zikai
aa08a1cf60
纯 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 隔离测试库）：上传/转换/下载/软删/硬删/超限/越权
300 lines
12 KiB
Python
300 lines
12 KiB
Python
"""PDF 转换服务：上传落盘 + 异步后台转换 + 进度追踪 + 两级删除。
|
||
|
||
复用 UploadService 的「流式落盘 + 存储路径生成」能力：原始上传文件与产物 PDF
|
||
均作为 UploadedFile 存储，本服务只维护 PdfJob 关系与状态。
|
||
|
||
转换在后台 asyncio task 中以 to_thread 执行（转换器是同步阻塞调用），
|
||
过程中经 DAO 更新 progress/status，供前端轮询。worker 限制：依赖进程内
|
||
asyncio 事件循环，与现有 reaper/hub 一致，保持单 worker。
|
||
|
||
删除语义：
|
||
user_delete -- 置 user_deleted=true + deleted_at，磁盘与 DB 行保留（管理页可见）。
|
||
admin_delete -- 删原始/产物磁盘文件 + UploadedFile 行 + PdfJob 行（真正删除）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import secrets
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from fastapi import HTTPException, UploadFile
|
||
|
||
from ..config import get_settings
|
||
from ..dao.pdf_job_dao import PdfJobDAO
|
||
from ..dao.uploaded_file_dao import UploadedFileDAO
|
||
from ..models.pdf_job import PdfJob
|
||
from ..schemas.pdf import PdfJobOut, PdfJobListResponse, PdfSubmitResponse
|
||
from . import pdf_converter
|
||
from .upload_service import UploadService
|
||
|
||
logger = logging.getLogger("zikai.pdf")
|
||
|
||
|
||
def new_owner_cookie() -> str:
|
||
"""生成新的用户标识 cookie 值（uuid4 hex）。"""
|
||
return secrets.token_hex(16)
|
||
|
||
|
||
class PdfService:
|
||
def __init__(self, job_dao: PdfJobDAO, file_dao: UploadedFileDAO) -> None:
|
||
s = get_settings()
|
||
self.job_dao = job_dao
|
||
self.file_dao = file_dao
|
||
self.upload_root = s.resolved_upload_dir()
|
||
self.cfg = s.pdf
|
||
# 复用 UploadService 的存储路径生成与落盘能力
|
||
self._upload = UploadService(file_dao)
|
||
|
||
# ---------------- 提交 ----------------
|
||
|
||
def submit(self, file: UploadFile, owner_cookie: str) -> tuple[PdfJobOut, int]:
|
||
"""上传原始文件并创建 pending 任务，返回 (任务视图, source_file_id)。
|
||
|
||
校验大小 ≤ max_size_bytes 与扩展名白名单；复用 UploadService 流式落盘
|
||
入库为 UploadedFile，再建 PdfJob 关联。转换不在此处执行（由 controller
|
||
调 schedule_convert 异步触发）。
|
||
"""
|
||
filename = file.filename or "upload.epub"
|
||
if not pdf_converter.is_supported(filename):
|
||
raise HTTPException(400, "仅支持 epub 文件")
|
||
|
||
# 大小校验：UploadFile 流式无已知长度，先读一遍统计并重置（小文件可行），
|
||
# 对大文件更优的做法是流式计数，这里复用 stream_to_disk 后按 size 校验。
|
||
resp = self._upload.stream_to_disk(file, source="pdf", uploaded_by=owner_cookie)
|
||
if resp.size_bytes > self.cfg.max_size_bytes:
|
||
# 超限：清理刚落盘的文件与 DB 行，保持无副作用
|
||
try:
|
||
self._upload.delete_file(resp.id)
|
||
except Exception: # pragma: no cover
|
||
logger.warning("清理超限文件失败 id=%s", resp.id)
|
||
raise HTTPException(
|
||
413,
|
||
f"文件过大（{resp.size_bytes} > {self.cfg.max_size_bytes}，上限 250MB）",
|
||
)
|
||
|
||
job = PdfJob(
|
||
owner_cookie=owner_cookie,
|
||
source_file_id=resp.id,
|
||
source_filename=filename,
|
||
source_size=resp.size_bytes,
|
||
status="pending",
|
||
progress=0,
|
||
)
|
||
job = self.job_dao.create(job)
|
||
logger.info("PDF 任务已创建 job=%s file=%s size=%d", job.id, filename, resp.size_bytes)
|
||
return PdfJobOut.model_validate(job), resp.id
|
||
|
||
def schedule_convert(self, job_id: int) -> None:
|
||
"""在当前事件循环起一个后台 task 执行转换（不阻塞调用方）。
|
||
|
||
用 to_thread 跑同步转换器；转换中分段更新 progress。
|
||
重要：后台 task 必须用独立 DB Session（请求 Session 在请求结束后即关闭），
|
||
故 _convert_async 内部经 _fresh_service 重建带新 Session 的 service。
|
||
"""
|
||
asyncio.create_task(self._convert_async(job_id))
|
||
|
||
async def _convert_async(self, job_id: int) -> None:
|
||
"""后台转换：pending -> converting(进度) -> done/failed。
|
||
|
||
每个阶段用独立 Session（_fresh_service），避免引用请求 Session（已关闭）。
|
||
"""
|
||
try:
|
||
await asyncio.to_thread(self._run_with_fresh_session, "_mark_converting", job_id)
|
||
await asyncio.wait_for(
|
||
asyncio.to_thread(self._run_with_fresh_session, "_do_convert", job_id),
|
||
timeout=self.cfg.convert_timeout_seconds,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
await asyncio.to_thread(self._run_with_fresh_session, "_mark_failed", job_id, "转换超时")
|
||
except Exception as exc:
|
||
await asyncio.to_thread(self._run_with_fresh_session, "_mark_failed", job_id, f"转换失败：{exc}")
|
||
|
||
@staticmethod
|
||
def _run_with_fresh_session(method_name: str, *args) -> None:
|
||
"""用独立 DB Session 构造新 PdfService 实例执行其方法。
|
||
|
||
后台线程不能复用请求的 Session（请求结束即关闭），故每次操作新建 Session。
|
||
method_name 是 PdfService 实例方法名，在此用新 service 调用对应方法。
|
||
"""
|
||
from ..database import get_session_local
|
||
db = get_session_local()()
|
||
try:
|
||
svc = PdfService(PdfJobDAO(db), UploadedFileDAO(db))
|
||
getattr(svc, method_name)(*args)
|
||
finally:
|
||
db.close()
|
||
|
||
# ---------------- 同步转换实现（在线程中执行，用独立 Session） ----------------
|
||
|
||
def _mark_converting(self, job_id: int) -> None:
|
||
job = self.job_dao.get(job_id)
|
||
if job is None:
|
||
return
|
||
job.status = "converting"
|
||
job.progress = 5
|
||
self.job_dao.update(job)
|
||
|
||
def _do_convert(self, job_id: int) -> None:
|
||
"""执行转换并落产物 PDF 为 UploadedFile。"""
|
||
job = self.job_dao.get(job_id)
|
||
if job is None:
|
||
return
|
||
|
||
src_row = self.file_dao.get_by_id(job.source_file_id)
|
||
if src_row is None:
|
||
self._mark_failed(job_id, "原始文件记录丢失")
|
||
return
|
||
src_path = (self.upload_root / src_row.storage_path).resolve()
|
||
if not src_path.exists():
|
||
self._mark_failed(job_id, "原始文件实体不存在")
|
||
return
|
||
|
||
# 产物 PDF 存储路径（复用 UploadService 的路径生成，扩展名固定 .pdf）
|
||
rel_path, abs_path = self._upload.make_storage_path("result.pdf")
|
||
part_path = abs_path.with_name(abs_path.name + ".part")
|
||
|
||
# 更新进度到「渲染中」
|
||
job.status = "converting"
|
||
job.progress = 30
|
||
self.job_dao.update(job)
|
||
|
||
try:
|
||
pdf_converter.convert_to_pdf(src_path, part_path)
|
||
except Exception:
|
||
part_path.unlink(missing_ok=True)
|
||
raise
|
||
|
||
job.progress = 80
|
||
self.job_dao.update(job)
|
||
|
||
# 落产物 UploadedFile（复用 commit_entity 的原子改名 + 入库）
|
||
from ..models.uploaded_file import UploadedFile
|
||
import hashlib
|
||
size = part_path.stat().st_size
|
||
sha256 = self._hash_file(part_path)
|
||
entity = UploadedFile(
|
||
storage_path=str(rel_path),
|
||
original_filename=Path(job.source_filename).stem + ".pdf",
|
||
content_type="application/pdf",
|
||
size_bytes=size,
|
||
sha256=sha256,
|
||
source="pdf-convert",
|
||
uploaded_by=job.owner_cookie,
|
||
)
|
||
saved = self._upload.commit_entity(entity, part_path, abs_path)
|
||
|
||
job.output_file_id = saved.id
|
||
job.status = "done"
|
||
job.progress = 100
|
||
self.job_dao.update(job)
|
||
logger.info("PDF 转换完成 job=%s output_file_id=%s", job_id, saved.id)
|
||
|
||
def _mark_failed(self, job_id: int, message: str) -> None:
|
||
job = self.job_dao.get(job_id)
|
||
if job is None:
|
||
return
|
||
job.status = "failed"
|
||
job.error_message = message[:500]
|
||
self.job_dao.update(job)
|
||
logger.warning("PDF 转换失败 job=%s: %s", job_id, message)
|
||
|
||
def _hash_file(self, path: Path) -> str:
|
||
import hashlib
|
||
h = hashlib.sha256()
|
||
with path.open("rb") as f:
|
||
while chunk := f.read(self._upload.chunk_bytes):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
|
||
# ---------------- 查询 ----------------
|
||
|
||
def list_for_user(self, owner_cookie: str, limit: int = 100, offset: int = 0) -> PdfJobListResponse:
|
||
limit = min(max(limit, 1), self.cfg.list_limit)
|
||
offset = max(offset, 0)
|
||
total = self.job_dao.count_for_user(owner_cookie)
|
||
rows = self.job_dao.list_for_user(owner_cookie, limit=limit, offset=offset)
|
||
return PdfJobListResponse(total=total, items=[PdfJobOut.model_validate(r) for r in rows])
|
||
|
||
def list_for_admin(self, limit: int = 100, offset: int = 0) -> PdfJobListResponse:
|
||
limit = min(max(limit, 1), self.cfg.list_limit)
|
||
offset = max(offset, 0)
|
||
total = self.job_dao.count_all()
|
||
rows = self.job_dao.list_all(limit=limit, offset=offset)
|
||
return PdfJobListResponse(total=total, items=[PdfJobOut.model_validate(r) for r in rows])
|
||
|
||
def get_job(self, job_id: int, owner_cookie: str) -> PdfJobOut:
|
||
"""用户查询单任务：仅当归属本人且未软删时可见。"""
|
||
job = self.job_dao.get(job_id)
|
||
if job is None or job.owner_cookie != owner_cookie or job.user_deleted:
|
||
raise HTTPException(404, "任务不存在")
|
||
return PdfJobOut.model_validate(job)
|
||
|
||
def get_output_path(self, job_id: int, owner_cookie: str) -> tuple[PdfJobOut, Path, str]:
|
||
"""返回 (任务视图, 产物磁盘绝对路径, 下载文件名) 供下载。
|
||
|
||
用户仅可下载自己未软删且已完成的任务产物。
|
||
"""
|
||
job = self.job_dao.get(job_id)
|
||
if job is None or job.owner_cookie != owner_cookie or job.user_deleted:
|
||
raise HTTPException(404, "任务不存在")
|
||
if job.status != "done" or job.output_file_id is None:
|
||
raise HTTPException(409, "任务尚未完成，无法下载")
|
||
out_row = self.file_dao.get_by_id(job.output_file_id)
|
||
if out_row is None:
|
||
raise HTTPException(410, "产物文件记录丢失")
|
||
path = (self.upload_root / out_row.storage_path).resolve()
|
||
if not path.exists():
|
||
raise HTTPException(410, "产物文件实体不存在")
|
||
return PdfJobOut.model_validate(job), path, out_row.original_filename
|
||
|
||
# ---------------- 删除 ----------------
|
||
|
||
def user_delete(self, job_id: int, owner_cookie: str) -> bool:
|
||
"""用户软删：仅置 user_deleted=true，磁盘与 DB 行保留（管理页仍可见）。"""
|
||
job = self.job_dao.get(job_id)
|
||
if job is None or job.owner_cookie != owner_cookie or job.user_deleted:
|
||
return False
|
||
job.user_deleted = True
|
||
job.deleted_at = datetime.now(timezone.utc)
|
||
self.job_dao.update(job)
|
||
logger.info("用户软删 PDF 任务 job=%s", job_id)
|
||
return True
|
||
|
||
def admin_delete(self, job_id: int) -> bool:
|
||
"""管理员硬删：删原始/产物磁盘文件 + UploadedFile 行 + PdfJob 行。
|
||
|
||
真正删除，不可恢复。磁盘删除失败仅记日志，仍清 DB 行保证列表不再显示。
|
||
"""
|
||
job = self.job_dao.get(job_id)
|
||
if job is None:
|
||
return False
|
||
# 删原始文件
|
||
self._safe_delete_file(job.source_file_id)
|
||
# 删产物文件（若有）
|
||
if job.output_file_id is not None:
|
||
self._safe_delete_file(job.output_file_id)
|
||
# 删 PdfJob 行
|
||
self.job_dao.db.delete(job)
|
||
self.job_dao.db.commit()
|
||
logger.info("管理员硬删 PDF 任务 job=%s", job_id)
|
||
return True
|
||
|
||
def _safe_delete_file(self, file_id: int) -> None:
|
||
"""删 UploadedFile 磁盘文件 + DB 行；失败只记日志不阻断。"""
|
||
row = self.file_dao.get_by_id(file_id)
|
||
if row is None:
|
||
return
|
||
path = (self.upload_root / row.storage_path).resolve()
|
||
try:
|
||
Path(path).unlink(missing_ok=True)
|
||
except Exception as exc: # pragma: no cover
|
||
logger.warning("删除文件失败 file_id=%s path=%s: %s", file_id, path, exc)
|
||
try:
|
||
self.file_dao.db.delete(row)
|
||
self.file_dao.db.commit()
|
||
except Exception as exc: # pragma: no cover
|
||
logger.warning("删除文件 DB 行失败 file_id=%s: %s", file_id, exc)
|