"""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)