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

"""统一预览服务：PDF直接返回；DOC/DOCX按需转换为PDF并缓存。"""

from __future__ import annotations

import logging
import os
import tempfile
from pathlib import Path

import fitz
from flask import current_app

from dms.common.enums import ContentExtractStatus
from dms.common.errors import (
    PreviewConversionFailedError,
    PreviewConversionTimeoutError,
    PreviewConverterUnavailableError,
    PreviewUnavailableError,
)
from dms.services.document_converter import DocumentConverter, create_converter
from dms.storage.paths import (
    UnsafeStoragePathError,
    is_path_inside_preview,
    preview_cache_path,
    resolve_storage_path,
)

logger = logging.getLogger(__name__)

_CONVERTIBLE_EXTENSIONS = {"doc", "docx"}


def _get_converter() -> DocumentConverter:
    """从 Flask 应用配置获取当前转换器。"""
    config = current_app.config
    # 应用初始化时若已挂载转换器实例则优先复用，否则动态创建
    converter = current_app.extensions.get("dms_document_converter")
    if converter is None:
        converter = create_converter(config)
        current_app.extensions["dms_document_converter"] = converter
    return converter


class PreviewService:
    """文档预览服务：管理 PDF 直接预览与 Office 转换后预览。"""

    def __init__(self, storage_root: Path, converter: DocumentConverter) -> None:
        self.storage_root = storage_root
        self.converter = converter

    @classmethod
    def current(cls) -> "PreviewService":
        root = Path(current_app.config["DMS_STORAGE_ROOT"]).expanduser().resolve()
        return cls(root, _get_converter())

    def preview_path_for(self, document) -> Path:
        """返回最终用于发送文件的磁盘路径。

        - PDF: 原始文件路径
        - DOC/DOCX: 缓存的 PDF 路径（按需转换）
        - 其他: 直接抛出 PreviewUnavailableError
        """
        extension = document.file_extension.lower().lstrip(".")
        original = resolve_storage_path(
            document.file_relative_path,
            self.storage_root,
        )

        if extension == "pdf":
            return original

        if extension not in _CONVERTIBLE_EXTENSIONS:
            raise PreviewUnavailableError()

        return self._convertible_preview_path(
            original, document.file_hash, extension
        )

    def _convertible_preview_path(
        self, source_path: Path, file_hash: str, extension: str
    ) -> Path:
        if not self.converter.available:
            raise PreviewConverterUnavailableError()

        cache = preview_cache_path(file_hash, self.storage_root)
        if self.validate_cached_pdf(cache):
            return cache

        try:
            pdf_bytes = self.converter.convert_to_pdf(source_path, file_hash)
        except (PreviewConverterUnavailableError, PreviewConversionTimeoutError):
            raise
        except Exception as exc:
            logger.exception("文档转换失败")
            raise PreviewConversionFailedError() from exc

        if not pdf_bytes or not pdf_bytes.startswith(b"%PDF-"):
            raise PreviewConversionFailedError()

        # 原子写入：先写临时文件再重命名
        cache.parent.mkdir(parents=True, exist_ok=True)
        temp_file: Path | None = None
        try:
            with tempfile.NamedTemporaryFile(
                dir=cache.parent,
                prefix=f".{file_hash}.",
                suffix=".tmp",
                delete=False,
            ) as temp_stream:
                temp_stream.write(pdf_bytes)
                temp_file = Path(temp_stream.name)
            os.replace(temp_file, cache)
        except OSError as exc:
            if temp_file is not None:
                temp_file.unlink(missing_ok=True)
            logger.exception("写入预览缓存失败")
            raise PreviewConversionFailedError() from exc

        return cache

    def validate_cached_pdf(self, cache_path: Path) -> bool:
        """验证缓存文件是否为有效 PDF 且位于 preview 目录内。"""
        try:
            if not is_path_inside_preview(cache_path, self.storage_root):
                return False
            if not cache_path.is_file() or cache_path.stat().st_size == 0:
                return False
            header = cache_path.read_bytes()[:8]
            if not header.startswith(b"%PDF-"):
                return False
            with fitz.open(stream=cache_path.read_bytes(), filetype="pdf") as _:
                return True
        except Exception:
            return False


def extract_doc_content_via_pdf(
    source_path: Path,
    converter: DocumentConverter,
    file_hash: str,
) -> tuple[str | None, str]:
    """将 DOC 转换为 PDF 后提取正文。

    返回 (text, status)。status 为 SUCCESS、EMPTY 或 FAILED。
    """
    from dms.services.document_content_service import _pdf_text

    if not converter.available:
        return None, ContentExtractStatus.FAILED.value

    try:
        pdf_bytes = converter.convert_to_pdf(source_path, file_hash)
    except Exception:
        logger.exception("DOC 转 PDF 提取正文失败")
        return None, ContentExtractStatus.FAILED.value

    if not pdf_bytes or not pdf_bytes.startswith(b"%PDF-"):
        return None, ContentExtractStatus.FAILED.value

    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as temp_file:
            temp_file.write(pdf_bytes)
            temp_path = Path(temp_file.name)
        text = _pdf_text(temp_path)
    except Exception:
        logger.exception("DOC 转换后的 PDF 提取正文失败")
        return None, ContentExtractStatus.FAILED.value
    finally:
        if temp_path is not None:
            temp_path.unlink(missing_ok=True)

    return text or None, (
        ContentExtractStatus.SUCCESS.value if text
        else ContentExtractStatus.EMPTY.value
    )


__all__ = [
    "PreviewService",
    "extract_doc_content_via_pdf",
]