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

"""文档格式转换抽象，支持 LibreOffice 离线转换与测试用的 Fake 实现。"""

from __future__ import annotations

import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
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-"

# Windows 下宿主进程无控制台（如打包 exe）时启动控制台子程序（soffice.com）
# 会闪现黑框；CREATE_NO_WINDOW 使子进程在无窗口控制台运行，管道输出不受影响。
_NO_WINDOW_FLAGS = (
    subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
)

# 表格类扩展名：预览转 PDF 时启用 SinglePageSheets——每个工作表按实际
# 内容区域导出为单页，忽略纸张版型（宽表不再被 A4 等纸张分页截断）。
_SPREADSHEET_EXTENSIONS = {"xls", "xlsx", "et"}
_SINGLE_PAGE_FILTER = (
    'pdf:calc_pdf_Export:{"SinglePageSheets":{"type":"boolean","value":true}}'
)

# 版本探测失败后的重试间隔：失败结果只短期缓存，避免一次瞬时异常
# （如残留的挂起 soffice 进程占用默认 profile）导致整个进程生命周期回退分页。
_SINGLE_PAGE_DETECT_RETRY_SECONDS = 300.0


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()
        self._version_lock = threading.Lock()
        self._single_page_supported: bool | None = None
        self._single_page_detected_at: float | None = None

    @property
    def available(self) -> bool:
        if not self._enabled:
            return False
        if not self._executable:
            return False
        return Path(self._executable).is_file()

    def _version_probe_executable(self) -> str:
        """--version 探测用可执行文件。

        Windows 上 soffice.exe（GUI 子系统）执行 --version 会派生
        soffice.bin（控制台子系统），父链无控制台可继承时新建可见控制台
        窗口打印版本并停在 ``Press Enter to continue...`` 等待输入，
        导致探测超时挂起并闪现黑框。soffice.com 为控制台包装器，stdout
        走管道即打即退（配合 CREATE_NO_WINDOW 无窗口），故优先使用。
        """
        if sys.platform == "win32" and self._executable:
            candidate = Path(self._executable).with_name("soffice.com")
            if candidate.is_file():
                return str(candidate)
        return self._executable

    def _supports_single_page_sheets(self) -> bool:
        """检测 LibreOffice 是否支持 SinglePageSheets 滤镜（需 ≥7.4）。

        首次调用执行 ``--version`` 探测并缓存结果；探测成功永久缓存，
        失败结果仅缓存 ``_SINGLE_PAGE_DETECT_RETRY_SECONDS`` 秒后重试，
        避免一次瞬时失败使整个进程生命周期回退标准纸张分页转换。
        """
        with self._version_lock:
            if self._single_page_supported:
                return True
            if (
                self._single_page_supported is False
                and self._single_page_detected_at is not None
                and time.monotonic() - self._single_page_detected_at
                < _SINGLE_PAGE_DETECT_RETRY_SECONDS
            ):
                return False
            supported = False
            try:
                completed = subprocess.run(
                    [self._version_probe_executable(), "--version"],
                    capture_output=True,
                    text=True,
                    timeout=15,
                    check=False,
                    creationflags=_NO_WINDOW_FLAGS,
                )
                output = (completed.stdout or "") + (completed.stderr or "")
                supported = _version_at_least_7_4(output)
            except Exception:
                supported = False
            self._single_page_supported = supported
            self._single_page_detected_at = time.monotonic()
            if not supported:
                logger.info(
                    "LibreOffice 不支持 SinglePageSheets（版本<7.4 或探测失败），"
                    "表格预览回退标准纸张分页（%.0f 秒后重试探测）",
                    _SINGLE_PAGE_DETECT_RETRY_SECONDS,
                )
            return supported

    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="/:")
            # 表格类且 LibreOffice ≥7.4：按内容区域单页导出；其余维持标准 pdf 转换
            extension = source_path.suffix.lower().lstrip(".")
            if (
                extension in _SPREADSHEET_EXTENSIONS
                and self._supports_single_page_sheets()
            ):
                convert_target = _SINGLE_PAGE_FILTER
            else:
                convert_target = "pdf"
            args = [
                self._executable,
                "--headless",
                "--nologo",
                "--nodefault",
                "--nofirststartwizard",
                "--nolockcheck",
                "--convert-to",
                convert_target,
                "--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),
                        creationflags=_NO_WINDOW_FLAGS,
                    )
                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 _version_at_least_7_4(output: str) -> bool:
    """解析 ``soffice --version`` 输出，判定是否 ≥ 7.4（JSON 滤镜参数要求）。

    新版日历式版本号（如 24.8 / 25.2）主版本大于 7，同样支持。
    """
    match = re.search(r"(\d+)\.(\d+)", output or "")
    if not match:
        return False
    major, minor = int(match.group(1)), int(match.group(2))
    return major > 7 or (major == 7 and minor >= 4)


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",
]