"""文档格式转换抽象，支持 LibreOffice 离线转换与测试用的 Fake 实现。""" from __future__ import annotations import logging import os import shutil import subprocess import tempfile import threading from abc import ABC, abstractmethod from pathlib import Path from typing import Any from urllib.parse import quote import fitz from dms.common.errors import ( PreviewConversionFailedError, PreviewConverterUnavailableError, PreviewConversionTimeoutError, ) logger = logging.getLogger(__name__) _PDF_MAGIC = b"%PDF-" class DocumentConverter(ABC): """文档转换器接口；业务层只依赖此接口，方便后续替换实现。""" @abstractmethod def convert_to_pdf(self, source_path: Path, file_hash: str) -> bytes: """将源文件转换为 PDF 并返回二进制内容。 实现必须保证： - 不依赖当前工作目录； - 异常转换为 PreviewConversion* 业务错误； - 不泄露临时目录或命令行。 """ raise NotImplementedError @property @abstractmethod def available(self) -> bool: """转换器是否可用。""" raise NotImplementedError class LibreOfficeDocumentConverter(DocumentConverter): """使用 LibreOffice headless 将 DOC/DOCX 转换为 PDF。""" def __init__( self, executable: str, timeout_seconds: int, max_concurrency: int, enabled: bool = True, ) -> None: self._executable = executable self._timeout_seconds = timeout_seconds self._semaphore = threading.Semaphore(max(1, max_concurrency)) self._enabled = enabled self._in_flight: dict[str, LibreOfficeDocumentConverter._ConversionTask] = {} self._lock = threading.Lock() @property def available(self) -> bool: if not self._enabled: return False if not self._executable: return False return Path(self._executable).is_file() class _ConversionTask: __slots__ = ("event", "result") def __init__(self) -> None: self.event = threading.Event() self.result: bytes | None = None def _join_or_register(self, file_hash: str) -> tuple[_ConversionTask, bool]: """注册新的转换任务或加入已有任务。 返回 (task, is_owner)。is_owner=True 表示当前线程需要执行转换。 非 owner 线程持有 task 引用，即使 owner 清理 _in_flight 也能读取结果。 """ with self._lock: if file_hash in self._in_flight: return self._in_flight[file_hash], False task = self._ConversionTask() self._in_flight[file_hash] = task return task, True def _set_result(self, task: _ConversionTask, result: bytes | None) -> None: task.result = result task.event.set() def convert_to_pdf(self, source_path: Path, file_hash: str) -> bytes: if not self.available: raise PreviewConverterUnavailableError() task, is_owner = self._join_or_register(file_hash) if not is_owner: if not task.event.wait(self._timeout_seconds): raise PreviewConversionTimeoutError() if task.result is None: raise PreviewConversionFailedError() return task.result try: result = self._convert_locked(source_path, file_hash) self._set_result(task, result) return result except Exception: self._set_result(task, None) raise finally: with self._lock: self._in_flight.pop(file_hash, None) def _convert_locked(self, source_path: Path, file_hash: str) -> bytes: work_dir = Path(tempfile.mkdtemp(prefix=f"dms-lo-{file_hash[:8]}-")) user_dir = Path(tempfile.mkdtemp(prefix=f"dms-locfg-{file_hash[:8]}-")) try: output_dir = work_dir / "out" output_dir.mkdir(parents=True, exist_ok=True) input_file = work_dir / source_path.name input_file.write_bytes(source_path.read_bytes()) # LibreOffice 在 headless 模式下输出文件名与输入文件名一致，扩展名改为 pdf base_name = source_path.stem expected_output = output_dir / f"{base_name}.pdf" user_url = quote(str(user_dir.as_posix()), safe="/:") args = [ self._executable, "--headless", "--nologo", "--nodefault", "--nofirststartwizard", "--nolockcheck", "--convert-to", "pdf", "--outdir", str(output_dir), f"-env:UserInstallation=file:///{user_url}", str(input_file), ] with self._semaphore: try: process = subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=str(work_dir), ) except OSError as exc: logger.error("启动 LibreOffice 失败：%s", exc.__class__.__name__) raise PreviewConverterUnavailableError() from exc try: stdout, stderr = process.communicate( timeout=self._timeout_seconds ) except subprocess.TimeoutExpired as exc: logger.warning("LibreOffice 转换超时") _terminate_process(process) raise PreviewConversionTimeoutError() from exc finally: if process.poll() is None: _terminate_process(process) if process.returncode != 0: logger.error( "LibreOffice 退出码非零：returncode=%s", process.returncode, ) raise PreviewConversionFailedError() if not expected_output.is_file(): logger.error("LibreOffice 未生成预期 PDF 文件") raise PreviewConversionFailedError() pdf_bytes = expected_output.read_bytes() if not pdf_bytes.startswith(_PDF_MAGIC): logger.error("LibreOffice 输出文件头不是 PDF") raise PreviewConversionFailedError() return pdf_bytes finally: try: shutil.rmtree(work_dir, ignore_errors=True) shutil.rmtree(user_dir, ignore_errors=True) except Exception: logger.exception("清理 LibreOffice 临时目录失败") def _terminate_process(process: subprocess.Popen[Any]) -> None: """终止 LibreOffice 进程及其子进程。""" try: process.terminate() process.wait(timeout=5) except Exception: try: process.kill() except Exception: pass class FakeDocumentConverter(DocumentConverter): """测试用转换器：将源文件内容复制为 PDF 字节或按规则生成 PDF。""" def __init__(self, enabled: bool = True) -> None: self._enabled = enabled self.calls: list[tuple[Path, str]] = [] @property def available(self) -> bool: return self._enabled def convert_to_pdf(self, source_path: Path, file_hash: str) -> bytes: self.calls.append((source_path, file_hash)) if not self._enabled: raise PreviewConverterUnavailableError() source_text = source_path.read_bytes().decode("utf-8", errors="replace") document = fitz.open() try: page = document.new_page() page.insert_text((72, 72), source_text[:4096]) return document.tobytes() finally: document.close() def create_converter(config: dict[str, Any]) -> DocumentConverter: """根据 Flask 配置创建默认转换器。""" return LibreOfficeDocumentConverter( executable=config.get("DMS_LIBREOFFICE_EXECUTABLE", ""), timeout_seconds=config.get("DMS_LIBREOFFICE_TIMEOUT_SECONDS", 60), max_concurrency=config.get("DMS_LIBREOFFICE_MAX_CONCURRENCY", 2), enabled=config.get("DMS_OFFICE_PREVIEW_ENABLED", True), ) __all__ = [ "DocumentConverter", "LibreOfficeDocumentConverter", "FakeDocumentConverter", "create_converter", ]