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

# 字体名中的风格 token：剥离后不参与家族比较（加粗位单独保留在签名中）
_FONT_STYLE_TOKENS = {"bold", "italic", "oblique"}


def _font_signature(name: str, flags: int) -> str:
    """字体签名：剥离子集前缀与风格 token 的家族名（小写）+ 加粗位。

    返回空串表示无字体信息（不参与相邻行字体比较）。
    """
    cleaned = re.sub(r"^[A-Za-z]{6}\+", "", name or "")
    parts = [part for part in re.split(r"[-,]", cleaned) if part]
    family = "".join(
        part for part in parts if part.lower() not in _FONT_STYLE_TOKENS
    ).lower()
    if not family:
        return ""
    bold = bool(flags & 16) or any(part.lower() == "bold" for part in parts)
    return f"{family}|b" if bold else family


def _line_style(line) -> tuple[float, str]:
    """行主样式：(行内最大字号, 行内占非空白字符数最多的字体签名)。

    主字体按字符数加权取众数而非取首 span，避免行首少量西文数字
    （与中文不同字体嵌入）导致正文的行主字体来回漂移。
    """
    main_size = 0.0
    weights: dict[str, int] = {}
    for span in line["spans"]:
        text = span["text"].strip()
        if not text:
            continue
        main_size = max(main_size, span["size"])
        signature = _font_signature(span["font"], span["flags"])
        if signature:
            weights[signature] = weights.get(signature, 0) + len(text)
    dominant = max(weights, key=weights.get) if weights else ""
    return main_size, dominant


# 退化版面判定：逐行等距网格的间距容差（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, str]], float, float]] = []
    line_count = 0
    for block in data["blocks"]:
        if block.get("type") != 0:
            continue
        items: list[tuple[str, float, str]] = []
        for line in block["lines"]:
            text = "".join(span["text"] for span in line["spans"]).strip()
            if not text:
                continue
            size, font = _line_style(line)
            items.append((text, size, font))
        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, str]], 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"[0-9０-９]{1,3}[.、)）．]"          # 1. / 1、 / 1) / 1．
    r"|[（(][一二三四五六七八九十百0-9０-９]{1,4}[)）]"  # （一） / (1)
    r"|[一二三四五六七八九十]{1,3}、"   # 一、
    r"|第[一二三四五六七八九十百千0-9０-９]{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, str]]) -> list[str]:
    """将一组行（文本、主字号、主字体签名）合并为段落列表。

    新段落（切分）的情况：
    1. 下一行以条目/章节编号开头；
    2. 相邻行主字号相对差 ≥ _FONT_SIZE_SPLIT_RATIO，或主字体签名不同
       （标题与正文同块的兜底：公文字号体系下黑体/楷体标题与仿宋正文
       同字号，仅字体家族或加粗不同；字号/字体为空表示无版面信息，
       不参与比较。字体切分要求前一行至少 2 个字符——单字符行如项目
       符号"•"不可能是标题，避免符号与条目文本被字体差异拆开）；
    3. 当前缓冲以句末标点结尾（在追加该行后立即收尾）。
    其余情况视为段内软换行，按 _join_lines 规则合并。
    """
    paragraphs: list[str] = []
    buffer = ""
    previous_size = 0.0
    previous_font = ""
    previous_text = ""
    for text, size, font 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
            )
            or (
                previous_font
                and font
                and font != previous_font
                and len(previous_text) >= 2
            )
        ):
            paragraphs.append(buffer)
            buffer = text
        elif buffer:
            buffer = _join_lines(buffer, text)
        else:
            buffer = text
        previous_size = size
        previous_font = font
        previous_text = text
        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, str]] = []
    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",
]