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

"""不依赖AI或Office进程的PDF/DOCX正文提取与检索文本构建。"""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

import fitz
from docx import Document as WordDocument

from dms.common.enums import ContentExtractStatus

logger = logging.getLogger(__name__)


@dataclass(frozen=True, slots=True)
class ContentExtraction:
    text: str | None
    status: str
    extracted_at: datetime | None
    # 提取失败原因（脱敏：不含绝对路径，仅异常类型与首行信息），供管理员排障
    error: str | None = None


_ERROR_MAX_LENGTH = 480


def sanitize_extract_error(exc: Exception) -> str:
    message = str(exc).splitlines()[0] if str(exc).strip() else ""
    # 去掉可能的绝对路径（Windows 盘符 / POSIX 根路径），避免泄露部署目录
    message = re.sub(r"[A-Za-z]:[\\/][^\s,;]*", "<path>", message)
    message = re.sub(r"(?<![\w])/[A-Za-z0-9_./-]+", "<path>", message)
    text = f"{type(exc).__name__}: {message}" if message else type(exc).__name__
    return text[:_ERROR_MAX_LENGTH]


def build_search_text(
    name: str,
    summary: str | None,
    tags: list[str] | None,
    content_text: str | None,
) -> str:
    return " ".join(
        part for part in [name, summary or "", *(tags or []), content_text or ""]
        if part
    ).strip()


def _word_text(path: Path) -> str:
    document = WordDocument(path)
    values: list[str] = []

    def add_paragraphs(paragraphs) -> None:
        values.extend(
            paragraph.text.strip()
            for paragraph in paragraphs
            if paragraph.text.strip()
        )

    def add_tables(tables) -> None:
        for table in tables:
            for row in table.rows:
                for cell in row.cells:
                    add_paragraphs(cell.paragraphs)
                    add_tables(cell.tables)

    add_paragraphs(document.paragraphs)
    add_tables(document.tables)
    for section in document.sections:
        add_paragraphs(section.header.paragraphs)
        add_tables(section.header.tables)
        add_paragraphs(section.footer.paragraphs)
        add_tables(section.footer.tables)
    return "\n".join(values).strip()


# 块内相邻行主字号相对差达到该比例时视为段落边界（标题与正文同块的兜底）
_FONT_SIZE_SPLIT_RATIO = 0.2


def _line_main_size(line) -> float:
    """取行内最大 span 字号作为该行主字号。"""
    return max(
        (span["size"] for span in line["spans"] if span["text"].strip()),
        default=0.0,
    )


# 退化版面判定：逐行等距网格的间距容差（pt）与字号容差（pt）
_DEGENERATE_GAP_TOLERANCE = 2.0
_DEGENERATE_SIZE_TOLERANCE = 0.5


def _pdf_page_text(page) -> str:
    """按版面块识别段落：块边界即段落边界，块内合并软换行。

    PyMuPDF 按版面邻近度将行分组为块，标题/落款等因字号或间距不同独立成块，
    无需依赖标点猜测即可与正文分开。块内沿用启发式合并（句末标点/条目编号切分，
    相邻行字号突变亦视为段落边界）。整页为逐行等距同字号的网格（如扫描件文本层
    逐行定位）时块结构无分组价值，回退纯文本启发式。
    """
    data = page.get_text("dict", flags=fitz.TEXTFLAGS_TEXT)
    # 每块：(行列表[(文本, 主字号)], bbox顶, bbox底)
    blocks: list[tuple[list[tuple[str, float]], float, float]] = []
    line_count = 0
    for block in data["blocks"]:
        if block.get("type") != 0:
            continue
        items = [
            (text, _line_main_size(line))
            for line in block["lines"]
            if (text := "".join(span["text"] for span in line["spans"]).strip())
        ]
        if items:
            blocks.append((items, block["bbox"][1], block["bbox"][3]))
            line_count += len(items)
    if line_count == 0:
        return ""
    if _is_degenerate_line_grid(blocks, line_count):
        return _merge_wrapped_lines(page.get_text("text"))
    paragraphs: list[str] = []
    for items, _, _ in blocks:
        paragraphs.extend(_merge_line_items(items))
    return "\n".join(paragraphs)


def _is_degenerate_line_grid(
    blocks: list[tuple[list[tuple[str, float]], float, float]],
    line_count: int,
) -> bool:
    """判断整页是否为"每行一块、等距同字号"的退化网格（扫描件文本层典型特征）。

    需同时满足：每块仅一行、块数≥3（至少两个间距可比较均匀性）、
    相邻块间距一致、各行字号一致。标题+正文等少量块或间距/字号有变化的
    页面信任版面块结构，不视为退化。
    """
    if len(blocks) < 3 or len(blocks) < line_count:
        return False
    if any(len(items) != 1 for items, _, _ in blocks):
        return False
    gaps = [
        blocks[index + 1][1] - top for index, (_, top, _) in enumerate(blocks[:-1])
    ]
    if max(gaps) - min(gaps) > _DEGENERATE_GAP_TOLERANCE:
        return False
    sizes = [items[0][1] for items, _, _ in blocks]
    return max(sizes) - min(sizes) <= _DEGENERATE_SIZE_TOLERANCE


def _pdf_text(path: Path) -> str:
    with fitz.open(stream=path.read_bytes(), filetype="pdf") as document:
        pages = [_pdf_page_text(page) for page in document]
    # 页与页之间以空行分隔（页边界即段落边界）
    return "\n\n".join(page for page in pages if page).strip()


# 句末标点：一行以此结尾视为自然段落结束，不再与下一行合并
_SENTENCE_END_RE = re.compile(r"[。！？!?；;：:…．.\"”’」』）)\]}>》]+$")
# 条目/章节起始标记：下一行以此开头视为新段落（如 "1." "（一）" "第2章" "一、"）
_LIST_START_RE = re.compile(
    r"^(?:"
    r"\d{1,3}[.、)）]"          # 1. / 1、 / 1)
    r"|[（(][一二三四五六七八九十百\d]{1,4}[)）]"  # （一） / (1)
    r"|[一二三四五六七八九十]{1,3}、"   # 一、
    r"|第[一二三四五六七八九十百千\d]{1,5}[章节条部分篇]"  # 第3章 / 第五节
    r")"
)


def _join_lines(previous: str, following: str) -> str:
    """合并段内被换行符断开的两行：两侧均为 ASCII 字母数字时补空格，其余直接相连。"""
    if previous and following and previous[-1].isascii() and previous[-1].isalnum() \
            and following[0].isascii() and following[0].isalnum():
        return f"{previous} {following}"
    return previous + following


def _merge_line_items(items: list[tuple[str, float]]) -> list[str]:
    """将一组行（文本、主字号）合并为段落列表。

    新段落（切分）的情况：
    1. 下一行以条目/章节编号开头；
    2. 相邻行主字号相对差 ≥ _FONT_SIZE_SPLIT_RATIO（标题与正文同块的兜底，
       字号为 0 表示无版面信息，不参与比较）；
    3. 当前缓冲以句末标点结尾（在追加该行后立即收尾）。
    其余情况视为段内软换行，按 _join_lines 规则合并。
    """
    paragraphs: list[str] = []
    buffer = ""
    previous_size = 0.0
    for text, size in items:
        if buffer and _LIST_START_RE.match(text):
            paragraphs.append(buffer)
            buffer = text
        elif (
            buffer
            and previous_size
            and size
            and abs(size - previous_size) / max(size, previous_size)
            >= _FONT_SIZE_SPLIT_RATIO
        ):
            paragraphs.append(buffer)
            buffer = text
        elif buffer:
            buffer = _join_lines(buffer, text)
        else:
            buffer = text
        previous_size = size
        if buffer and _SENTENCE_END_RE.search(buffer):
            paragraphs.append(buffer)
            buffer = ""
    if buffer:
        paragraphs.append(buffer)
    return paragraphs


def _merge_wrapped_lines(page_text: str) -> str:
    """纯文本兜底：PDF 复制文本时会在段内软换行处插入 \\n，此处按启发式规则合并。

    换行保留（视为新段落）的情况：
    1. 空行（原始段落边界）；
    2. 上一行以句末标点结尾；
    3. 下一行以条目/章节编号开头。
    其余情况视为段内软换行，直接合并（中文不加空格，英文单词间补一个空格）。
    """
    paragraphs: list[str] = []
    group: list[tuple[str, float]] = []
    for raw_line in page_text.splitlines():
        line = raw_line.strip()
        if not line:
            paragraphs.extend(_merge_line_items(group))
            group = []
        else:
            group.append((line, 0.0))
    paragraphs.extend(_merge_line_items(group))
    return "\n".join(paragraphs)


def extract_document_content(path: Path, extension: str) -> ContentExtraction:
    normalized = extension.lower().lstrip(".")
    if normalized not in {"pdf", "docx"}:
        return ContentExtraction(
            text=None,
            status=ContentExtractStatus.UNSUPPORTED.value,
            extracted_at=None,
        )
    extracted_at = datetime.now(timezone.utc).replace(tzinfo=None)
    try:
        text = _pdf_text(path) if normalized == "pdf" else _word_text(path)
    except Exception as exc:
        logger.exception("文档正文提取失败：extension=%s", normalized)
        return ContentExtraction(
            text=None,
            status=ContentExtractStatus.FAILED.value,
            extracted_at=extracted_at,
            error=sanitize_extract_error(exc),
        )
    return ContentExtraction(
        text=text or None,
        status=(
            ContentExtractStatus.SUCCESS.value
            if text
            else ContentExtractStatus.EMPTY.value
        ),
        extracted_at=extracted_at,
        error=None,
    )


__all__ = [
    "ContentExtraction",
    "build_search_text",
    "extract_document_content",
    "sanitize_extract_error",
]