document-management/backend/dms/services/document_converter.py

"""文档格式转换抽象，支持 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",
]