proposa-ai/scripts/test_step6.py

"""Step 6 人工验证：聚合 Step5 章节内容为最终投标 DOCX。

输入（均使用已生成中间产物，不重跑 Step1~Step5）：
  - Step3 最终大纲与评分/废标映射（step5_outline.pkl，由 Step5 落盘，
    内含 heading_mappings 与 evaluation_index_entries）
  - Step5 各章节 DOCX（step5_chapters/，含 manifest.json）
  - 正式模板 DOCX（提供样式、页眉页脚部件与封面/目录骨架）

输出：
  - 一份聚合 DOCX：封面 → 自动目录 → 商务部分与评标索引表 → 商务章 →
    技术部分 → 技术章；并完成标题/字号/字体/颜色/行距/缩进检查与修复。

用法（按顺序运行）:
  1. uv run python scripts/test_step1.py；
  2. uv run python scripts/test_step2.py；
  3. uv run python scripts/test_step3.py；
  4. uv run python scripts/test_step4.py；
  5. uv run python scripts/test_step5.py；
  6. uv run python scripts/test_step6.py。
"""

import glob
import json
import logging
import os
import pickle
import re
import sys
import zipfile
import xml.etree.ElementTree as ET

from _bootstrap import PROJECT_ROOT


OUTPUT_DIR = os.environ.get(
    "PROPOSA_STEP6_OUTPUT_DIR",
    os.environ.get("PROPOSA_WORK_DIR", "output/171-上海群众艺术馆"),
)
STEP5_CHAPTERS_DIR = os.environ.get(
    "PROPOSA_STEP5_CHAPTERS_DIR",
    os.path.join(OUTPUT_DIR, "step5_chapters"),
)
STEP5_OUTLINE_FILE = os.environ.get(
    "PROPOSA_STEP5_REVIEWED_OUTLINE_FILE",
    os.path.join(OUTPUT_DIR, "step5_outline.pkl"),
)
STEP3_REPORT_FILE = os.environ.get(
    "PROPOSA_STEP3_REPORT_FILE",
    os.path.join(OUTPUT_DIR, "step3_outline_report.json"),
)
TEMPLATE_PATH = os.environ.get(
    "PROPOSA_TEMPLATE_PATH",
    "src/templates/申勤投标模板.docx",
)
REFERENCE_TABLES_DIR = os.environ.get(
    "PROPOSA_REFERENCE_TABLES_DIR",
    os.path.join(OUTPUT_DIR, "内容提取_参考投书"),
)
OUTPUT_FILE = os.environ.get(
    "PROPOSA_STEP6_OUTPUT",
    os.path.join(OUTPUT_DIR, "171-上海群众艺术馆.docx"),
)


_W_NS = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"


def _w(tag: str) -> str:
    return f"{{{_W_NS}}}{tag}"


def _element_text(element) -> str:
    return "".join(node.text or "" for node in element.iter(_w("t")))


def _find_manifest(chapters_dir):
    if not os.path.isdir(chapters_dir):
        raise FileNotFoundError(
            f"Step5 章节目录不存在: {chapters_dir}\n"
            "请先按顺序运行 uv run python scripts/test_step1.py 至 "
            "uv run python scripts/test_step5.py。"
        )
    for candidate in glob.glob(
        os.path.join(chapters_dir, "chapters", "*", "manifest.json")
    ):
        return candidate
    return ""


def _load_chapter_records(chapters_dir):
    manifest_path = _find_manifest(chapters_dir)
    if not manifest_path:
        raise FileNotFoundError(
            f"Step5 章节 manifest 不存在: {chapters_dir}\n"
            "请先运行 uv run python scripts/test_step5.py。"
        )
    with open(manifest_path, "r", encoding="utf-8") as file:
        manifest = json.load(file)
    records = [
        {
            "id": str(record.get("id", "")),
            "title": str(record.get("title", "")),
            "artifact_path": str(record.get("artifact_path", "")),
            "status": str(record.get("status", "complete")),
        }
        for record in manifest.get("chapters", [])
    ]
    if not records:
        raise RuntimeError(f"Step5 manifest 中没有章节记录: {manifest_path}")
    return records


def _load_outline(path):
    if not os.path.isfile(path):
        raise FileNotFoundError(
            f"Step5 大纲/映射不存在: {path}\n"
            "请先运行 uv run python scripts/test_step5.py。"
        )
    with open(path, "rb") as file:
        return pickle.load(file)


def _find_index_table_template(reference_tables_dir):
    """定位 Step1 从参考投书提取的评标索引表 DOCX。"""
    if not os.path.isdir(reference_tables_dir):
        return ""
    matches = []
    for candidate in glob.glob(
        os.path.join(reference_tables_dir, "*与评标有关的投标文件主要内容索引表*.docx")
    ):
        matches.append(candidate)
    if not matches:
        return ""
    matches.sort()
    return matches[0]


def _assert_output_structure(output_path, outline):
    if not os.path.isfile(output_path):
        raise AssertionError(f"Step6 输出文件未生成: {output_path}")

    with zipfile.ZipFile(output_path, "r") as output_zip:
        names = set(output_zip.namelist())
        for required in ("word/document.xml", "word/header1.xml", "word/footer1.xml"):
            if required not in names:
                raise AssertionError(f"输出 DOCX 缺少部件: {required}")

        body = ET.fromstring(output_zip.read("word/document.xml")).find(_w("body"))
        if body is None:
            raise AssertionError("输出 DOCX 缺少 w:body")

        body_text = _element_text(body)
        instr_texts = [node.text or "" for node in body.iter(_w("instrText"))]
        instr_combined = " ".join(instr_texts)

        # 1. 每个数字顶层章必须作为 Heading 1 出现在正文中。
        numeric_chapters = [
            chapter
            for chapter in outline.chapters
            if str(getattr(chapter, "id", "")).isdigit()
        ]
        heading1_texts = []
        for paragraph in body.iter(_w("p")):
            p_pr = paragraph.find(_w("pPr"))
            style_id = ""
            if p_pr is not None:
                p_style = p_pr.find(_w("pStyle"))
                if p_style is not None:
                    style_id = (p_style.get(_w("val")) or "").strip().lower()
            if style_id in ("2", "heading1"):
                heading1_texts.append(_element_text(paragraph).strip())

        missing = [
            f"{chapter.id} {chapter.title}"
            for chapter in numeric_chapters
            if not any(
                chapter.title in text and re.match(r"^第\s*[一二三四五六七八九十\d]+\s*章", text)
                for text in heading1_texts
            )
        ]
        if missing:
            raise AssertionError("最终 DOCX 缺少顶层章节: " + ", ".join(missing))

        # 2. 自动目录域必须存在。
        if "TOC" not in instr_combined:
            raise AssertionError("最终 DOCX 未插入 Word 自动目录（TOC 域）")

        # 3. 评标索引表行数必须与 Step3 持久化的评分项条数一致（废标项不进入索引表）。
        expected_rows = len(
            [
                entry
                for entry in (outline.evaluation_index_entries or [])
                if str(entry.get("entry_type")) == "scoring"
            ]
        )
        tables = list(body.iter(_w("tbl")))
        if not tables:
            raise AssertionError("最终 DOCX 缺少评标索引表")
        index_table = tables[0]
        index_rows = index_table.findall(_w("tr"))
        if len(index_rows) != expected_rows + 1:
            raise AssertionError(
                f"评标索引表行数 {len(index_rows)} 与 Step3 评分项条数 {expected_rows} 不一致"
            )

        # 4. 页眉/页脚应保留 Step5 章节的原生内容（项目名 + 模板页眉）。
        header1_text = _element_text(ET.fromstring(output_zip.read("word/header1.xml")))
        footer1_text = _element_text(ET.fromstring(output_zip.read("word/footer1.xml")))
        if not header1_text.strip():
            raise AssertionError("商务部分页眉为空")
        if (outline.project_name or "").strip() and outline.project_name not in header1_text:
            raise AssertionError(
                f"商务部分页眉缺少项目名: {outline.project_name!r}"
            )
        if "PAGE" not in " ".join(
            node.text or "" for node in ET.fromstring(output_zip.read("word/footer1.xml")).iter(_w("instrText"))
        ):
            raise AssertionError("页脚缺少 PAGE 页码域")

    return {
        "chapters": len(numeric_chapters),
        "index_rows": expected_rows,
        "tables": len(tables),
        "heading1_count": len(heading1_texts),
    }


def main():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%H:%M:%S",
        stream=sys.stderr,
    )

    if not os.path.isfile(STEP3_REPORT_FILE):
        raise FileNotFoundError(
            f"Step3 大纲/评分映射不存在: {STEP3_REPORT_FILE}\n"
            "请先运行 uv run python scripts/test_step3.py。"
        )
    with open(STEP3_REPORT_FILE, "r", encoding="utf-8") as file:
        step3_report = json.load(file)

    outline = _load_outline(STEP5_OUTLINE_FILE)
    chapter_records = _load_chapter_records(STEP5_CHAPTERS_DIR)
    index_table_template_path = _find_index_table_template(REFERENCE_TABLES_DIR)

    print("=" * 60)
    print("Step 6: 聚合 Step5 章节内容为最终投标 DOCX")
    print("=" * 60)
    print(f"章节记录数: {len(chapter_records)}")
    print(f"Step3 评分/废标索引条数: {len(outline.evaluation_index_entries or [])}")
    print(f"Step3 标题关系映射条数: {len(outline.heading_mappings or [])}")
    print(
        f"Step3 报告索引条数: {len(step3_report.get('evaluation_index_entries', []) or [])}, "
        f"标题映射条数: {len(step3_report.get('heading_mappings', []) or [])}"
    )
    print(f"模板: {TEMPLATE_PATH}")
    print(
        f"评标索引表模板: {index_table_template_path or '未找到，使用兜底表格'}"
    )

    from step6_exporting.assembler import assemble_step5_document

    report = assemble_step5_document(
        chapter_records,
        outline,
        OUTPUT_FILE,
        template_path=TEMPLATE_PATH,
        index_table_template_path=index_table_template_path,
        project_name=outline.project_name,
    )

    print()
    print("=== 聚合结果 ===")
    print(f"输出文件: {report.output_path}")
    print(f"聚合章节数: {report.chapter_count}")
    print(f"索引表行数: {report.index_row_count}")
    print(f"格式修复 run 数: {report.fixed_runs}")
    for warning in report.warnings:
        print(f"[WARN] {warning}")

    structure = _assert_output_structure(report.output_path, outline)
    print()
    print("=== 结构校验 ===")
    print(
        f"[PASS] 顶层章 {structure['chapters']} 个，Heading 1 {structure['heading1_count']} 个，"
        f"索引 {structure['index_rows']} 行，表格 {structure['tables']} 张，自动目录已插入"
    )
    print("Step 6 测试完成 [OK]")


if __name__ == "__main__":
    main()