"""用标准库 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()