proposa-ai/scripts/test_step4.py

"""Step 4 人工验证：按 Step3 最终目录逐章生成并输出块式写作报告。

本脚本会调用当前配置的 LLM。运行：uv run python scripts/test_step4.py
"""

import json
import logging
import os
import pickle
import re
import sys
import zipfile
import xml.etree.ElementTree as ET
from collections import Counter

from _bootstrap import PROJECT_ROOT

REFERENCE_BID = os.environ.get(
    "PROPOSA_REFERENCE_BID",
    "test_data/171-上海群众艺术馆/参考投标文件/物业管理费项目投标文件.docx",
)

TEMPLATE_PATH = os.environ.get(
    "PROPOSA_TEMPLATE_PATH", "src/templates/申勤投标模板.docx"
)

OUTPUT_DIR = os.environ.get(
    "PROPOSA_STEP4_OUTPUT_DIR",
    os.environ.get("PROPOSA_WORK_DIR", "output/171-上海群众艺术馆"),
)
STEP2_INFO_FILE = os.environ.get(
    "PROPOSA_STEP2_INFO_FILE",
    os.path.join(OUTPUT_DIR, "step2_info.pkl"),
)
STEP3_OUTLINE_FILE = os.environ.get(
    "PROPOSA_STEP3_OUTLINE_FILE",
    os.path.join(OUTPUT_DIR, "step3_outline.pkl"),
)
CHAPTER_OUTPUT_DIR = os.path.join(OUTPUT_DIR, "step4_chapters")
os.makedirs(OUTPUT_DIR, exist_ok=True)
os.environ["BID_OUTLINE_CACHE_DIR"] = os.path.join(OUTPUT_DIR, ".outline_cache")
os.environ["BID_LLM_CACHE_DIR"] = os.path.join(OUTPUT_DIR, ".llm_cache")

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


def _load_reference_bid(path, project_data, output_dir):
    """参考投书载入入口，保留给既有回归测试兼容。

    当前手工流程已在 test_step2.py 完成参考投书正文与表格载入并写入
    step2_info.pkl；test_step4.py 主流程不再调用本函数。
    """
    if not path or not os.path.isfile(path):
        return 0

    from doc_reader.reader import read_docx_paragraph_texts_et
    from models import ParsedDocument
    from step1_parsing.table_extractor import extract_reference_tables_with_llm

    reference_content = "\n".join(read_docx_paragraph_texts_et(path))
    if reference_content:
        project_data.reference_bids.append(ParsedDocument(
            file_path=path,
            file_name=os.path.basename(path),
            content=reference_content,
            doc_type="reference_bid",
        ))
    reference_table_dir = os.path.join(output_dir, "内容提取_参考投书")
    project_data.reference_tables = extract_reference_tables_with_llm(
        path, output_dir=reference_table_dir
    )
    return len(project_data.reference_tables)


def _walk(chapters):
    for chapter in chapters:
        yield chapter
        yield from _walk(chapter.children)


def _inspect_chapter_artifact(path):
    result = {
        "native_table_count": 0,
        "native_image_count": 0,
        "unresolved_placeholders": [],
    }
    if not path or not os.path.isfile(path):
        return result
    word_ns = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
    with zipfile.ZipFile(path, "r") as package:
        root = ET.fromstring(package.read("word/document.xml"))
    text = "".join(
        node.text or "" for node in root.iter(f"{{{word_ns}}}t")
    )
    result["native_table_count"] = sum(
        1 for _ in root.iter(f"{{{word_ns}}}tbl")
    )
    result["native_image_count"] = sum(
        1 for _ in root.iter(f"{{{word_ns}}}drawing")
    )
    result["unresolved_placeholders"] = sorted(set(
        re.findall(r"%%[^%\n]{1,80}%%", text)
    ))
    return result


def _chapter_record(chapter, entry_counts):
    blocks = list(chapter.content_blocks or [])
    supplements = [
        node for node in _walk([chapter]) if node.supplement_content.strip()
    ]
    record = {
        "id": chapter.id,
        "title": chapter.title,
        "chapter_type": chapter.chapter_type.value,
        "template_original_id": chapter.template_original_id,
        "template_original_title": chapter.template_original_title,
        "preserve_template_layout": bool(chapter.preserve_template_layout),
        "template_fill_completed": chapter.template_fill_completed,
        "generated_chars": len(chapter.generated_content or ""),
        "scoring_binding_count": sum(
            len(node.direct_scoring_bindings or node.direct_scoring_criteria)
            for node in _walk([chapter])
        ),
        "rejection_binding_count": sum(
            len(node.direct_rejection_bindings) for node in _walk([chapter])
        ),
        "index_entry_count": entry_counts.get(chapter.id, 0),
        "supplement_node_count": len(supplements),
        "supplement_chars": sum(len(node.supplement_content) for node in supplements),
        "content_blocks": blocks,
        "artifact_path": chapter.artifact_path,
        "index_deferred_until_pagination": any(
            block.get("block_type") == "deferred_evaluation_index"
            for block in blocks
        ),
        "nodes": [
            {
                "id": node.id,
                "title": node.title,
                "template_original_id": node.template_original_id,
                "content_generation_mode": node.content_generation_mode,
                "scoring_bindings": list(
                    node.direct_scoring_bindings or node.direct_scoring_criteria
                ),
                "rejection_bindings": list(node.direct_rejection_bindings),
                "supplement_chars": len(node.supplement_content or ""),
                "supplement_preview": (node.supplement_content or "")[:240],
            }
            for node in _walk([chapter])
            if (
                node.direct_scoring_bindings
                or node.direct_scoring_criteria
                or node.direct_rejection_bindings
            )
        ],
    }
    record.update(_inspect_chapter_artifact(chapter.artifact_path))
    return record


def _build_report(outline, analysis=None):
    nodes = {node.id: node for node in _walk(outline.chapters)}
    errors = []
    entry_counts = Counter()
    for entry in outline.evaluation_index_entries:
        heading_id = str(entry.get("final_heading_id", ""))
        entry_counts[heading_id.split(".", 1)[0]] += 1
        node = nodes.get(heading_id)
        if node is None:
            errors.append(f"索引落点不存在: {entry.get('entry_id')} -> {heading_id}")
        elif not node.supplement_content.strip():
            errors.append(
                f"直接落点缺少补充正文: {entry.get('entry_id')} -> "
                f"{heading_id} {node.title}"
            )

    heading_pattern = re.compile(
        r"(?m)^\s*(?:#{1,6}\s+|[一二三四五六七八九十]+、|"
        r"（[一二三四五六七八九十]+）|\d+[.、])"
    )
    for node in nodes.values():
        supplement = node.supplement_content or ""
        if supplement and heading_pattern.search(supplement):
            errors.append(f"补充块含疑似新标题: {node.id} {node.title}")
        if supplement and re.search(r"(?m)^\s*\|.*\|\s*$", supplement):
            errors.append(f"补充块含 Markdown 表格: {node.id} {node.title}")

    for chapter in outline.chapters:
        if not chapter.preserve_template_layout:
            continue
        kinds = [str(block.get("block_type", "")) for block in chapter.content_blocks]
        if not chapter.template_fill_completed:
            errors.append(f"模板章未完成项目信息填充: {chapter.id} {chapter.title}")
        allowed_prefixes = (
            ["template_base", "native_table_plan"],
            ["template_base", "deferred_evaluation_index"],
        )
        if kinds[:2] not in allowed_prefixes:
            errors.append(f"模板章块顺序错误: {chapter.id} {chapter.title}: {kinds}")

    records = [_chapter_record(chapter, entry_counts) for chapter in outline.chapters]
    for record in records:
        if record["unresolved_placeholders"]:
            errors.append(
                f"章节产物仍有未解析占位符: {record['id']} {record['title']}: "
                + ", ".join(record["unresolved_placeholders"][:10])
            )
    project_fields = dict(
        getattr(analysis, "project_fields", {}) or {}
    ) if analysis is not None else {}
    if analysis is not None and getattr(analysis, "agency_name", ""):
        project_fields["采购机构名称"] = analysis.agency_name
    return {
        "passed": not errors,
        "errors": errors,
        "summary": {
            "chapter_count": len(outline.chapters),
            "heading_count": len(nodes),
            "evaluation_index_entry_count": len(outline.evaluation_index_entries),
            "template_preserved_chapter_count": sum(
                1 for chapter in outline.chapters if chapter.preserve_template_layout
            ),
            "supplement_node_count": sum(
                1 for node in nodes.values() if node.supplement_content.strip()
            ),
        },
        "project_fields": project_fields,
        "chapters": records,
    }


def _write_report(report):
    json_path = os.path.join(OUTPUT_DIR, "step4_generation_report.json")
    md_path = os.path.join(OUTPUT_DIR, "step4_generation_report.md")
    with open(json_path, "w", encoding="utf-8") as file:
        json.dump(report, file, ensure_ascii=False, indent=2)

    summary = report["summary"]
    lines = [
        "# Step4 逐章生成检查报告", "",
        f"- 门禁结果：{'通过' if report['passed'] else '失败'}",
        f"- 一级章：{summary['chapter_count']}；全部标题：{summary['heading_count']}",
        f"- 评分/废标索引条目：{summary['evaluation_index_entry_count']}",
        f"- 模板保留章：{summary['template_preserved_chapter_count']}；"
        f"实际补充节点：{summary['supplement_node_count']}", "",
    ]
    if report["errors"]:
        lines.extend(["## 门禁错误", ""])
        lines.extend(f"- {error}" for error in report["errors"])
        lines.append("")
    if report.get("project_fields"):
        lines.extend(["## Step2 项目填表字段", ""])
        lines.extend(
            f"- {key}：{value or '未提取'}"
            for key, value in report["project_fields"].items()
        )
        lines.append("")
    lines.extend(["## 各章生成情况", ""])
    for chapter in report["chapters"]:
        block_order = ", ".join(
            str(block.get("block_type", "")) for block in chapter["content_blocks"]
        ) or "无"
        lines.extend([
            f"### {chapter['id']} {chapter['title']}", "",
            f"- 类型：{chapter['chapter_type']}；模板保留：{chapter['preserve_template_layout']}；"
            f"项目填充完成：{chapter['template_fill_completed']}",
            f"- 模板原节点：{chapter['template_original_id']} {chapter['template_original_title']}",
            f"- 基础正文：{chapter['generated_chars']} 字；补充节点："
            f"{chapter['supplement_node_count']}；补充正文：{chapter['supplement_chars']} 字",
            f"- 评分绑定：{chapter['scoring_binding_count']}；废标绑定："
            f"{chapter['rejection_binding_count']}；索引条目：{chapter['index_entry_count']}",
            f"- 内容块顺序：{block_order}",
            f"- 原生表格：{chapter['native_table_count']}；原生图片："
            f"{chapter['native_image_count']}；未解析占位符："
            f"{', '.join(chapter['unresolved_placeholders']) or '无'}",
            f"- 最终分页后构建索引：{chapter['index_deferred_until_pagination']}",
            f"- 章节产物：{chapter['artifact_path'] or '无'}", "",
        ])
        for node in chapter["nodes"]:
            preview = node["supplement_preview"].replace("\n", " ") or "无"
            lines.extend([
                f"- `{node['id']}` {node['title']}（模板原节点 "
                f"`{node['template_original_id'] or '-'}`；模式 "
                f"`{node['content_generation_mode'] or '-'}`；评分 "
                f"{', '.join(node['scoring_bindings']) or '-'}；废标 "
                f"{', '.join(node['rejection_bindings']) or '-'}；补充 "
                f"{node['supplement_chars']} 字）",
                f"  - 预览：{preview}",
            ])
        if chapter["nodes"]:
            lines.append("")
    with open(md_path, "w", encoding="utf-8") as file:
        file.write("\n".join(lines).rstrip() + "\n")
    return md_path, json_path


def main():
    from step4_writing import write_content

    print("=" * 60)
    print("Step 2: 复用已落盘的分析与项目资料（Step 4 的前置）")
    print("=" * 60)
    if not os.path.isfile(STEP2_INFO_FILE):
        raise FileNotFoundError(
            f"Step2 分析结果不存在: {STEP2_INFO_FILE}\n"
            "请先运行 uv run python scripts/test_step2.py。"
        )
    with open(STEP2_INFO_FILE, "rb") as file:
        step2_payload = pickle.load(file)
    analysis = step2_payload["analysis"]
    project_data = step2_payload["project_data"]

    print(f"通用资料: {len(project_data.general_materials)} 个文件")
    print(f"参考投标书: {len(project_data.reference_bids)} 篇")
    print(f"参考投书表格: {len(project_data.reference_tables)} 张")
    print(
        f"复用 Step2 分析: {len(analysis.scoring_criteria)} 评分项, "
        f"{len(analysis.rejection_items)} 废标项"
    )

    print("\n" + "=" * 60)
    print("Step 3: 复用已落盘的最终目录（Step 4 的唯一结构来源）")
    print("=" * 60)
    if not os.path.isfile(STEP3_OUTLINE_FILE):
        raise FileNotFoundError(
            f"Step3 大纲结果不存在: {STEP3_OUTLINE_FILE}\n"
            "请先运行 uv run python scripts/test_step3.py。"
        )
    with open(STEP3_OUTLINE_FILE, "rb") as file:
        outline = pickle.load(file)
    print(
        f"最终目录: {len(outline.chapters)} 章, "
        f"{len(outline.evaluation_index_entries)} 条索引映射"
    )

    print("\n" + "=" * 60)
    print("Step 4: 按模板基块 → 项目/表格填充 → 评分/废标补充块生成")
    print("=" * 60)
    outline = write_content(
        outline, analysis, project_data,
        template_path=TEMPLATE_PATH,
        reference_bid_path=REFERENCE_BID,
        chapter_output_dir=CHAPTER_OUTPUT_DIR,
    )
    report = _build_report(outline, analysis)
    md_path, json_path = _write_report(report)
    print(f"逐章可读报告: {md_path}")
    print(f"机器可读报告: {json_path}")
    for chapter in report["chapters"]:
        print(
            f"  {chapter['id']} {chapter['title']}: 基础 {chapter['generated_chars']:,} 字, "
            f"补充 {chapter['supplement_chars']:,} 字/{chapter['supplement_node_count']} 节点, "
            f"评分 {chapter['scoring_binding_count']}, 废标 {chapter['rejection_binding_count']}"
        )
    if report["errors"]:
        raise AssertionError(
            "Step4 块式生成门禁失败；详见报告：\n- "
            + "\n- ".join(report["errors"][:30])
        )
    print("\nStep 4 测试完成 [OK]")


if __name__ == "__main__":
    main()