proposa-ai/scripts/test_step2.py

"""
Step 2 测试：招标文件深度分析

读取文档 → 构建 ProjectData → 调用 analyze_tender → 打印分析结果。
用法: uv run python scripts/test_step2.py
"""

import logging
import os
import pickle
import sys

from _bootstrap import PROJECT_ROOT

# ---------- 测试数据（环境变量可覆盖） ----------
TENDER_FILE = os.environ.get(
    "PROPOSA_TENDER_FILE",
    "test_data/171-上海群众艺术馆/上海市群众艺术馆物业管理服务采购项目招标文件.pdf",
)
PROCUREMENT_FILE = os.environ.get(
    "PROPOSA_PROCUREMENT_FILE",
    "test_data/171-上海群众艺术馆/采购需求.docx",
)
CLARIFICATION_FILE = os.environ.get("PROPOSA_CLARIFICATION_FILE", "")
COMPANY_INFO_DIR = os.environ.get(
    "PROPOSA_COMPANY_INFO_DIR",
    "src/templates/通用标书生成资料",
)
REFERENCE_BID = os.environ.get(
    "PROPOSA_REFERENCE_BID",
    "test_data/171-上海群众艺术馆/参考投标文件/物业管理费项目投标文件.docx",
)
OUTPUT_DIR = os.environ.get(
    "PROPOSA_STEP2_OUTPUT_DIR",
    os.environ.get("PROPOSA_WORK_DIR", "output/171-上海群众艺术馆"),
)
STEP1_ITEMS_FILE = os.environ.get(
    "PROPOSA_STEP1_ITEMS_FILE",
    os.path.join(OUTPUT_DIR, "step1_extracted_items.pkl"),
)
STEP2_INFO_FILE = os.environ.get(
    "PROPOSA_STEP2_INFO_FILE",
    os.path.join(OUTPUT_DIR, "step2_info.pkl"),
)

# ---------- 日志 ----------
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):
    """载入参考投书正文及表格，供 Step3/Step4/Step5 复用。"""
    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",
        ))
        print(
            f"参考投标书: {os.path.basename(path)} "
            f"({len(reference_content):,} 字符)"
        )

    reference_table_dir = os.path.join(output_dir, "内容提取_参考投书")
    project_data.reference_tables = extract_reference_tables_with_llm(
        path, output_dir=reference_table_dir
    )
    print(
        f"参考投书表格(含LLM增强): {len(project_data.reference_tables)} 张 "
        f"-> {reference_table_dir}"
    )
    return len(project_data.reference_tables)

# ---------- 1. 读取文档 ----------
print("=" * 60)
print("Step 2: 加载文档并构建 ProjectData")
print("=" * 60)

from doc_reader import read_file
from models import ProjectData, ParsedDocument

pd = ProjectData(project_id="test", project_name="测试项目")

# ---------- 0. 复用 Step1 已提取的表/函/表格 ----------
if not os.path.isfile(STEP1_ITEMS_FILE):
    raise FileNotFoundError(
        f"Step1 提取结果不存在: {STEP1_ITEMS_FILE}\n"
        "请先运行 uv run python scripts/test_step1.py。"
    )
with open(STEP1_ITEMS_FILE, "rb") as file:
    pd.extracted_items = pickle.load(file)
print(f"复用 Step1 表/函/表格: {len(pd.extracted_items)} 项")

# 读招标文件
if os.path.isfile(TENDER_FILE):
    content = read_file(TENDER_FILE)
    if content:
        pd.tender_docs.append(ParsedDocument(
            file_path=TENDER_FILE,
            file_name=os.path.basename(TENDER_FILE),
            content=content,
            doc_type="tender",
        ))
        print(f"招标文件: {os.path.basename(TENDER_FILE)} ({len(content):,} 字符)")

# 读采购需求
if PROCUREMENT_FILE and os.path.isfile(PROCUREMENT_FILE):
    content = read_file(PROCUREMENT_FILE)
    if content:
        pd.procurement_docs.append(ParsedDocument(
            file_path=PROCUREMENT_FILE,
            file_name=os.path.basename(PROCUREMENT_FILE),
            content=content,
            doc_type="procurement",
        ))
        print(f"采购需求: {os.path.basename(PROCUREMENT_FILE)} ({len(content):,} 字符)")

# 读澄清公告
if CLARIFICATION_FILE and os.path.isfile(CLARIFICATION_FILE):
    content = read_file(CLARIFICATION_FILE)
    if content:
        pd.clarification_docs.append(ParsedDocument(
            file_path=CLARIFICATION_FILE,
            file_name=os.path.basename(CLARIFICATION_FILE),
            content=content,
            doc_type="clarification",
        ))
        print(f"澄清公告: {os.path.basename(CLARIFICATION_FILE)} ({len(content):,} 字符)")

# 读通用资料
if os.path.isdir(COMPANY_INFO_DIR):
    count = 0
    for root, dirs, files in os.walk(COMPANY_INFO_DIR):
        for fn in files:
            if fn.endswith(".md"):
                fp = os.path.join(root, fn)
                content = read_file(fp)
                if content:
                    pd.general_materials.append(ParsedDocument(
                        file_path=fp,
                        file_name=fn,
                        content=content,
                        doc_type="general_info",
                    ))
                    count += 1
    print(f"通用资料: {count} 个文件")

# 参考投书在 Step2 载入后保存到 step2_info.pkl，Step3/Step4/Step5 不再重复读取。
_load_reference_bid(REFERENCE_BID, pd, OUTPUT_DIR)

# ---------- 2. 深度分析 ----------
print()
print("=" * 60)
print("Step 2: 深度分析招标文件")
print("=" * 60)

os.environ.setdefault("BID_LLM_CACHE_DIR", os.path.join(OUTPUT_DIR, ".llm_cache"))
from step2_analysis import analyze_tender

analysis = analyze_tender(pd, output_dir=OUTPUT_DIR)
print(f"Step2 分析结果已落盘: {STEP2_INFO_FILE}")

# ---------- 3. 打印结果 ----------
print()
print("=== 分析结果 ===")
print(f"项目名称: {analysis.project_name}")
print(f"采购人:   {analysis.buyer_name}")
print(f"包件数:   {len(analysis.packages)}")
print(f"服务需求: {analysis.service_requirements[:120]}...")
print()

print(f"评分细则 ({len(analysis.scoring_criteria)} 项):")
for c in analysis.scoring_criteria:
    print(f"  [{c.id}] {c.name} | {c.category} | {c.max_score}分")
    if c.description:
        print(f"       {c.description[:100]}")

print()
print(f"废标项 ({len(analysis.rejection_items)} 项):")
for r in analysis.rejection_items:
    print(f"  [{r.id}] {r.description[:80]}...")

print()
print(f"响应清单 ({len(analysis.response_requirements)} 项):")
for r in analysis.response_requirements:
    print(f"  [{r.id}] [{r.category}] {r.name}")

print()
print(f"文档模板 ({len(analysis.templates)} 个):")
for t in analysis.templates:
    print(f"  - {t.name}")

print()
print("Step 2 测试完成 [OK]")