document-management/backend/tests/dms/services/test_document_content_service.py

"""正文提取服务（document_content_service）纯单元测试。

不依赖数据库与请求上下文：
- _merge_wrapped_lines：PDF 段内软换行合并启发式（纯文本兜底）；
- _merge_line_items：块内合并规则（含字号突变切分）；
- sanitize_extract_error：失败原因脱敏（去除绝对路径）；
- extract_document_content：UNSUPPORTED/FAILED 分支、PDF 版面块分段与退化回退。
"""

from __future__ import annotations

from pathlib import Path

import fitz

from dms.common.enums import ContentExtractStatus
from dms.services.document_content_service import (
    _merge_line_items,
    _merge_wrapped_lines,
    extract_document_content,
    sanitize_extract_error,
)


def test_merge_chinese_wrapped_line_without_space():
    text = "这是一段在段内被换行\n打断的中文正文，合并后不应有空格"
    assert _merge_wrapped_lines(text) == "这是一段在段内被换行打断的中文正文，合并后不应有空格"


def test_merge_english_words_with_space():
    text = "emergency response\nplan continues here"
    assert _merge_wrapped_lines(text) == "emergency response plan continues here"


def test_sentence_end_punctuation_keeps_paragraph_break():
    text = "第一条意见如下。\n第二条意见如下：\n具体内容见附件。"
    assert _merge_wrapped_lines(text) == "第一条意见如下。\n第二条意见如下：\n具体内容见附件。"


def test_blank_line_keeps_paragraph_break():
    text = "第一段未结束\n仍在同一段\n\n第二段开始"
    assert _merge_wrapped_lines(text) == "第一段未结束仍在同一段\n第二段开始"


def test_list_marker_start_keeps_paragraph_break():
    text = "总体要求如下\n1. 第一条\n（一）细分条款\n一、中文序号\n第2章范围"
    assert _merge_wrapped_lines(text) == "总体要求如下\n1. 第一条\n（一）细分条款\n一、中文序号\n第2章范围"


def test_mixed_cjk_and_ascii_merge_directly():
    # 一侧为中文时不补空格
    text = "使用 PDF\n格式存储文件"
    assert _merge_wrapped_lines(text) == "使用 PDF格式存储文件"


def test_sanitize_error_strips_windows_path():
    exc = RuntimeError("cannot open D:/dms/storage/uploads/a/b.pdf: permission denied")
    sanitized = sanitize_extract_error(exc)
    assert sanitized.startswith("RuntimeError:")
    assert "D:/" not in sanitized
    assert "<path>" in sanitized


def test_sanitize_error_strips_posix_path_and_truncates():
    exc = ValueError("failed at /var/lib/dms/storage/file.pdf " + "x" * 600)
    sanitized = sanitize_extract_error(exc)
    assert "/var/lib" not in sanitized
    assert len(sanitized) <= 480


def test_sanitize_error_multiline_uses_first_line():
    exc = RuntimeError("第一行原因\n第二行堆栈")
    assert sanitize_extract_error(exc) == "RuntimeError: 第一行原因"


def test_unsupported_extension_returns_unsupported(tmp_path: Path):
    target = tmp_path / "image.png"
    target.write_bytes(b"\x89PNG\r\n\x1a\n")
    result = extract_document_content(target, "png")
    assert result.status == ContentExtractStatus.UNSUPPORTED.value
    assert result.text is None
    assert result.error is None
    assert result.extracted_at is None


def test_corrupt_pdf_returns_failed_with_error(tmp_path: Path):
    target = tmp_path / "broken.pdf"
    target.write_bytes(b"not a real pdf content")
    result = extract_document_content(target, "pdf")
    assert result.status == ContentExtractStatus.FAILED.value
    assert result.text is None
    assert result.error  # 必须携带失败原因
    assert result.extracted_at is not None


def _write_pdf(path: Path, lines: list[str], gap: int = 16, fontsize: int = 11) -> None:
    document = fitz.open()
    page = document.new_page()
    y = 72
    for line in lines:
        # china-s：PyMuPDF 内嵌简体中文字体（默认 helv 无法编码 CJK）
        page.insert_text((72, y), line, fontsize=fontsize, fontname="china-s")
        y += gap
    document.save(path)
    document.close()


def test_pdf_extraction_merges_wrapped_lines(tmp_path: Path):
    target = tmp_path / "wrapped.pdf"
    _write_pdf(
        target,
        ["这是一段在段内被换行", "打断的中文正文。", "另一段独立内容。"],
    )
    result = extract_document_content(target, "pdf")
    assert result.status == ContentExtractStatus.SUCCESS.value
    assert result.error is None
    assert result.text is not None
    # 第一行未以句末标点结尾：与第二行合并；第二行以"。"结尾：保留段落边界
    assert "这是一段在段内被换行打断的中文正文。" in result.text
    assert "打断的中文正文。\n另一段独立内容。" in result.text


def test_pdf_heading_in_separate_block_not_merged_with_body(tmp_path: Path):
    """标题行与正文行距较大（独立成块）时，即使标题无句末标点也不并入正文。"""
    target = tmp_path / "heading.pdf"
    document = fitz.open()
    page = document.new_page()
    page.insert_text((72, 72), "第一章 总则", fontsize=11, fontname="china-s")
    page.insert_text((72, 72 + 24), "为规范方案管理，特制定本办法。", fontsize=11, fontname="china-s")
    document.save(target)
    document.close()
    result = extract_document_content(target, "pdf")
    assert result.status == ContentExtractStatus.SUCCESS.value
    assert result.text is not None
    assert "第一章 总则\n为规范方案管理，特制定本办法。" in result.text


def test_pdf_different_font_size_splits_paragraph(tmp_path: Path):
    """标题与正文字号不同时独立成块（或块内字号突变切分），不合并为一段。"""
    target = tmp_path / "fontsize.pdf"
    document = fitz.open()
    page = document.new_page()
    page.insert_text((72, 72), "方案名称", fontsize=18, fontname="china-s")
    page.insert_text((72, 72 + 24), "本方案用于说明字号差异的段落切分", fontsize=11, fontname="china-s")
    document.save(target)
    document.close()
    result = extract_document_content(target, "pdf")
    assert result.status == ContentExtractStatus.SUCCESS.value
    assert result.text is not None
    assert "方案名称\n本方案用于说明字号差异的段落切分" in result.text


def test_pdf_degenerate_layout_falls_back_to_heuristic(tmp_path: Path):
    """每行独立成块（行距≥20pt 逐行定位）时回退纯文本启发式，行为与现状一致。"""
    target = tmp_path / "degenerate.pdf"
    _write_pdf(
        target,
        ["这是一段在段内被换行", "打断的中文正文。", "另一段独立内容。"],
        gap=24,
    )
    result = extract_document_content(target, "pdf")
    assert result.status == ContentExtractStatus.SUCCESS.value
    assert result.text is not None
    assert "这是一段在段内被换行打断的中文正文。" in result.text
    assert "打断的中文正文。\n另一段独立内容。" in result.text


def test_merge_line_items_splits_on_font_size_jump():
    items = [("方案名称", 18.0, ""), ("本方案用于说明字号差异", 11.0, "")]
    assert _merge_line_items(items) == ["方案名称", "本方案用于说明字号差异"]


def test_merge_line_items_merges_same_font_size():
    items = [("段内被换行", 11.0, "fangsong"), ("打断的中文正文", 11.0, "fangsong")]
    assert _merge_line_items(items) == ["段内被换行打断的中文正文"]


def test_merge_line_items_ignores_font_size_when_unavailable():
    # 字号为 0（纯文本兜底路径）不参与字号比较，仍按启发式合并
    items = [("段内被换行", 0.0, ""), ("打断的中文正文", 0.0, "")]
    assert _merge_line_items(items) == ["段内被换行打断的中文正文"]


def test_merge_line_items_splits_on_font_family_change():
    # 公文字号体系：黑体标题与仿宋正文同字号，仅家族不同
    items = [("一、总体要求", 16.0, "simhei"), ("为保证方案落地，制定本要求。", 16.0, "fangsong")]
    assert _merge_line_items(items) == ["一、总体要求", "为保证方案落地，制定本要求。"]


def test_merge_line_items_splits_on_bold_change():
    # 同家族仅加粗不同（仿宋加粗三级标题 → 仿宋正文）
    items = [("主要职责范围", 16.0, "fangsong|b"), ("各成员单位按照职责分工落实。", 16.0, "fangsong")]
    assert _merge_line_items(items) == ["主要职责范围", "各成员单位按照职责分工落实。"]


def test_merge_line_items_ignores_font_when_unavailable():
    # 字体为空（纯文本兜底路径）不参与字体比较，仍按启发式合并
    items = [("段内被换行", 11.0, ""), ("打断的中文正文", 11.0, "")]
    assert _merge_line_items(items) == ["段内被换行打断的中文正文"]


def test_merge_line_items_keeps_single_char_symbol_with_following_text():
    # 前一行为单字符项目符号"•"（SymbolMT）：不可能是标题，字体差异不切分
    items = [("•", 11.0, "symbolmt"), ("组织机构；", 11.0, "dengxian")]
    assert _merge_line_items(items) == ["•组织机构；"]


def test_merge_line_items_splits_when_current_line_single_char():
    # 前一行为多字符标题、当前行为单字符（如表头"周"）：字体差异仍应切分
    items = [("表1  工作计划", 12.0, "simhei"), ("周", 10.6, "calibri|b")]
    assert _merge_line_items(items) == ["表1  工作计划", "周"]


def test_font_signature_normalization():
    from dms.services.document_content_service import _font_signature

    # 子集前缀剥离 + 风格 token 剥离 + 加粗位保留
    assert _font_signature("ABCDEE+Calibri-Bold", 0) == "calibri|b"
    assert _font_signature("ABCDEE+Calibri-Bold", 16) == "calibri|b"
    assert _font_signature("SimHei", 0) == "simhei"
    assert _font_signature("DengXian", 16) == "dengxian|b"
    assert _font_signature("", 0) == ""


def test_list_start_matches_fullwidth_number_and_period():
    from dms.services.document_content_service import _LIST_START_RE

    assert _LIST_START_RE.match("1．主要任务")
    assert _LIST_START_RE.match("１．主要任务")
    assert _LIST_START_RE.match("（２）保障措施")
    assert _LIST_START_RE.match("第３章组织指挥")
    assert not _LIST_START_RE.match("2025 年度工作安排")