""" 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]")