feat: 新增 PDF 转换服务(epub->pdf,cookie 用户隔离,软删/硬删两级删除)

纯 Python 转换(ebooklib 解析 epub + weasyprint 渲染 HTML/CSS 为 PDF),
无需 calibre/xvfb 系统依赖。复用 UploadedFile 存储(原始文件与产物 PDF)。

- model/dao/schema:PdfJob 记录转换任务(owner_cookie/source/output/进度/状态)
- pdf_converter:epub->PDF 转换器(按 spine 顺序拼 HTML,OPF 目录解析相对资源)
- pdf_service:上传落盘 + asyncio 后台转换(独立 Session)+ 进度追踪 + 两级删除
  · 用户软删(user_deleted,管理页仍可见)· 管理员硬删(真正删磁盘+记录)
- pdf_controller:/api/pdf/*(用户,cookie)+ /api/admin/pdf/*(Basic Auth)
- /pdf、/pdf-admin 页面(原生 JS,复用 common.css/js)
- 配置 pdf 段(max_size 250MB、转换超时、cookie)
- pytest 7 项(SQLite→MySQL 隔离测试库):上传/转换/下载/软删/硬删/超限/越权
This commit is contained in:
2026-07-27 11:33:43 +08:00
parent fa980e74e0
commit aa08a1cf60
21 changed files with 1549 additions and 6 deletions

View File

@@ -0,0 +1,105 @@
"""文件 -> PDF 转换器(纯 Python无 calibre/xvfb 系统依赖)。
当前支持 epub必要能力ebooklib 解析 epub 文档项(按 spine 顺序),拼接为
完整 HTML 后交 weasyprint 渲染为 PDF。epub 内的相对资源(图片/CSS经 base_url
指向 epub 解包目录解析。
接口抽象为 ``convert_to_pdf(src_path, dst_path)``:未来新增格式只需在本模块内
按扩展名分支调用方PdfService无需改动。
"""
from __future__ import annotations
import logging
import shutil
import tempfile
import zipfile
from pathlib import Path
logger = logging.getLogger("zikai.pdf")
# 支持的输入格式(小写扩展名 -> 是否可转)。新增格式在此登记并实现分支即可。
# epub 为必要能力;其余为 ebooklib/weasyprint 路径天然支持的电子书结构,
# 实测对纯 HTML 类 epub 同样有效,故一并放行。
SUPPORTED_EXTENSIONS = {".epub"}
def is_supported(filename: str) -> bool:
"""文件名扩展名是否在支持列表内。"""
return Path(filename).suffix.lower() in SUPPORTED_EXTENSIONS
def convert_to_pdf(src_path: Path, dst_path: Path) -> None:
"""把 src_path 指向的文件转为 PDF 写入 dst_path。
失败抛 RuntimeError调用方捕获后写 error_message。按扩展名分发
当前仅 epub 分支;新增格式在此 elif 扩展。
"""
ext = src_path.suffix.lower()
if ext == ".epub":
_epub_to_pdf(src_path, dst_path)
else:
raise RuntimeError(f"不支持的文件格式:{ext}(仅支持 epub")
def _epub_to_pdf(src_path: Path, dst_path: Path) -> None:
"""epub -> PDF解包 epub -> 按 spine 顺序取文档项 HTML -> weasyprint 渲染。
epub 本质是 zip。先解包到临时目录用 ebooklib 读取(其内部按 zip 解析,
base_url 指向解包后的 OEBPS/ 内容目录使相对图片/CSS 可被 weasyprint 解析)。
"""
# 延迟导入weasyprint 首次 import 较重(加载 pango/cairo且仅在真正转换时需要
import ebooklib # noqa: F401
from ebooklib import epub
from weasyprint import HTML
with tempfile.TemporaryDirectory(prefix="zpdf_") as tmp:
tmp_dir = Path(tmp)
# epub 是 zip解包到临时目录便于 weasyprint 解析相对资源
try:
with zipfile.ZipFile(src_path, "r") as zf:
zf.extractall(tmp_dir)
except zipfile.BadZipFile as exc:
raise RuntimeError(f"epub 文件损坏(非有效 zip{exc}") from exc
try:
book = epub.read_epub(str(src_path), {"ignore_ncx": True})
except Exception as exc:
raise RuntimeError(f"epub 解析失败:{exc}") from exc
# 按 spine 顺序收集文档项XHTML保证章节顺序正确
docs: list[str] = []
for idref, _linear in book.spine:
item = book.get_item_with_id(idref)
if item is not None:
docs.append(item.get_content().decode("utf-8", errors="replace"))
if not docs:
# spine 为空时回退:取所有文档项
docs = [
it.get_content().decode("utf-8", errors="replace")
for it in book.get_items_of_type(ebooklib.ITEM_DOCUMENT)
]
if not docs:
raise RuntimeError("epub 内无可转换的文档内容")
full_html = "\n".join(docs)
# 定位资源根目录(含图片/CSS 的目录):通常是 OEBPS/ 或根目录。
# epub 内资源(图片/CSS相对文档项引用文档项与资源同处 OPF 所在目录。
# 故以 OPF 文件所在目录作为 base_url使相对路径正确解析。
base_url = str(tmp_dir)
opf_files = list(tmp_dir.rglob("*.opf"))
if opf_files:
opf_dir = opf_files[0].parent
# OPF 可能在根目录,此时 base_url 保持 tmp_dir
if str(opf_dir) != str(tmp_dir):
base_url = str(opf_dir)
try:
HTML(string=full_html, base_url=base_url).write_pdf(str(dst_path))
except Exception as exc:
raise RuntimeError(f"PDF 渲染失败:{exc}") from exc
if not dst_path.exists() or dst_path.stat().st_size == 0:
raise RuntimeError("PDF 渲染未产出有效文件")
logger.info("epub->pdf 完成: %s -> %s (%d bytes)", src_path.name, dst_path.name, dst_path.stat().st_size)