proposa-ai/scripts/audit_step6_xml.py

"""用标准库 OOXML 审计大型 Step6 DOCX，避免 python-docx/lxml 原生崩溃。"""

from __future__ import annotations

import argparse
import json
import pickle
import re
import zipfile
from collections import defaultdict
from pathlib import Path
from xml.etree import ElementTree as ET

from _bootstrap import PROJECT_ROOT  # noqa: F401
from step3_outlining.scoring_structure import normalize_heading_text
from step6_exporting.docx_builder import _detect_heading_level


W = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}"


def _text(element) -> str:
    parts = []
    for node in element.iter():
        if node.tag in (W + "t", W + "instrText", W + "delText"):
            parts.append(node.text or "")
        elif node.tag == W + "tab":
            parts.append("\t")
        elif node.tag in (W + "br", W + "cr"):
            parts.append("\n")
    return "".join(parts)


def _style_id(paragraph) -> str:
    p_pr = paragraph.find(W + "pPr")
    p_style = p_pr.find(W + "pStyle") if p_pr is not None else None
    return p_style.get(W + "val", "") if p_style is not None else ""


def _style_levels(styles_root) -> dict[str, int]:
    result = {}
    for style in styles_root.findall(W + "style"):
        style_id = style.get(W + "styleId", "")
        name = style.find(W + "name")
        name_value = name.get(W + "val", "") if name is not None else ""
        combined = f"{style_id} {name_value}".lower().replace(" ", "")
        match = re.search(r"(?:heading|标题)([1-4])", combined)
        if match:
            result[style_id] = int(match.group(1))
    return result


def _normalize(text: str) -> str:
    return re.sub(r"\s+", "", text or "")


def audit(docx_path: Path, step2_path: Path, outline_path: Path) -> dict:
    with open(step2_path, "rb") as file:
        step2 = pickle.load(file)
    with open(outline_path, "rb") as file:
        outline = pickle.load(file)
    analysis = step2["analysis"]

    with zipfile.ZipFile(docx_path) as package:
        document_root = ET.fromstring(package.read("word/document.xml"))
        styles_root = ET.fromstring(package.read("word/styles.xml"))

    style_levels = _style_levels(styles_root)
    body = document_root.find(W + "body")
    body_children = list(body) if body is not None else []
    paragraphs = [node for node in body_children if node.tag == W + "p"]
    tables = [node for node in body_children if node.tag == W + "tbl"]
    paragraph_texts = [_text(node).strip() for node in paragraphs]

    headings = []
    h1 = []
    for paragraph, text in zip(paragraphs, paragraph_texts):
        if not text:
            continue
        level = style_levels.get(_style_id(paragraph), 0)
        if level == 1:
            h1.append(text)
        if level in (2, 3, 4) or (
            0 < _detect_heading_level(text) <= 3 and len(text) <= 80
        ):
            headings.append(text)

    expected_titles = {}
    expected_levels = {}
    for chapter in outline.flatten():
        for criterion_id in chapter.related_criteria or []:
            if chapter.level <= 1:
                continue
            previous = expected_levels.get(criterion_id, 0)
            if chapter.level > previous:
                expected_levels[criterion_id] = chapter.level
                expected_titles[criterion_id] = [chapter.title]
            elif chapter.level == previous:
                expected_titles.setdefault(criterion_id, []).append(chapter.title)
    normalized_headings = {normalize_heading_text(value) for value in headings}
    missing_scoring = []
    for criterion in analysis.scoring_criteria:
        titles = expected_titles.get(criterion.id) or [
            criterion.name or criterion.description
        ]
        if not any(
            normalize_heading_text(title) in normalized_headings
            for title in titles if title
        ):
            missing_scoring.append(criterion.id)

    table_texts = []
    table_signatures = []
    index_text = ""
    index_rows = []
    for table in tables:
        cells = [_text(cell).strip() for cell in table.iter(W + "tc")]
        table_text = "\n".join(cells)
        table_texts.append(table_text)
        table_signatures.append("|".join(_normalize(cell) for cell in cells))
        if not index_text:
            first_row = table.find(W + "tr")
            header = _text(first_row) if first_row is not None else ""
            if any(key in header for key in ("主要内容概述", "与评标有关")):
                index_text = table_text
                index_rows = [
                    [_text(cell).strip() for cell in row.iter(W + "tc")]
                    for row in table.findall(W + "tr")
                ]

    full_text = "\n".join(paragraph_texts + table_texts)
    normalized_full_text = _normalize(full_text)
    missing_rejections = [
        item.id
        for item in analysis.rejection_items
        if _normalize(item.description) not in normalized_full_text
    ]

    duplicate_groups = defaultdict(list)
    for index, signature in enumerate(table_signatures):
        if signature:
            duplicate_groups[signature].append(index)
    duplicate_indexes = [
        indexes for indexes in duplicate_groups.values() if len(indexes) > 1
    ]

    chapter3_text = ""
    in_chapter3 = False
    chapter3_parts = []
    for child in body_children:
        if child.tag != W + "p":
            if in_chapter3:
                chapter3_parts.append(_text(child))
            continue
        text = _text(child).strip()
        level = style_levels.get(_style_id(child), 0)
        if level == 1:
            if "需求理解" in text:
                in_chapter3 = True
                continue
            if in_chapter3:
                break
        if in_chapter3:
            chapter3_parts.append(text)
    chapter3_text = "\n".join(chapter3_parts)
    demand_scoring = [
        criterion for criterion in analysis.scoring_criteria
        if any(
            key in " ".join((criterion.category, criterion.name, criterion.description))
            for key in ("需求理解", "重点难点")
        )
    ]

    return {
        "paragraphs": len(paragraphs),
        "tables": len(tables),
        "h1": h1,
        "missing_scoring_headings": missing_scoring,
        "missing_rejections": missing_rejections,
        "index_sc_tokens": sorted(set(re.findall(r"SC-\d+", index_text, re.I))),
        "index_scoring_name_hits": sum(
            1 for item in analysis.scoring_criteria
            if _normalize(item.name) in _normalize(index_text)
        ),
        "index_missing_scoring": [
            {"id": item.id, "name": item.name}
            for item in analysis.scoring_criteria
            if _normalize(item.name) not in _normalize(index_text)
        ],
        "index_scoring_total": len(analysis.scoring_criteria),
        "index_rows": index_rows,
        "duplicate_table_index_groups": duplicate_indexes,
        "unresolved_placeholders": sorted(set(
            re.findall(r"%%[^%\n]+%%", full_text)
        )),
        "chapter3_characters": len(_normalize(chapter3_text)),
        "chapter3_relevant_scoring": {
            item.id: {
                "name": item.name,
                "name_found": _normalize(item.name) in _normalize(chapter3_text),
                "description_found": _normalize(item.description) in _normalize(chapter3_text),
            }
            for item in demand_scoring
        },
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("docx")
    parser.add_argument("--step2", required=True)
    parser.add_argument("--outline", required=True)
    args = parser.parse_args()
    result = audit(Path(args.docx), Path(args.step2), Path(args.outline))
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()