#!/usr/bin/env python3 """Extract entity tables from the raw Yifangbao tender Excel workbook.""" from __future__ import annotations import hashlib import importlib.util from collections import defaultdict from pathlib import Path from typing import Any from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter ROOT = Path(__file__).resolve().parents[1] RAW_PATH = ROOT / "data" / "raw" / "2026年1月-6月乙方宝中标通知20260612.xls" ANALYZE_SCRIPT = ROOT / "scripts" / "analyze_bid_data.py" OUT_DIR = ROOT / "docs" / "deliverables" / "20260617-raw-excel-entity-tables" OUT_XLSX = OUT_DIR / "原始Excel招投标实体抽取表.xlsx" FONT = "Microsoft YaHei" BLUE = "1F4E79" LIGHT_BLUE = "D9EAF7" WHITE = "FFFFFF" BORDER = Border( left=Side(style="thin", color="B7C9DA"), right=Side(style="thin", color="B7C9DA"), top=Side(style="thin", color="B7C9DA"), bottom=Side(style="thin", color="B7C9DA"), ) PRODUCT_KEYWORDS = [ "智慧食堂", "智慧食堂系统", "监管系统", "食堂监管", "软硬件系统", "软硬件", "系统维保", "维保服务", "设备采购", "刷脸收银", "收银设备", "食堂设备", "餐饮服务", "食堂服务", "一卡通", "机器人", "终端", "平台", "系统升级", ] def load_analyze_module() -> Any: spec = importlib.util.spec_from_file_location("bid_analyze", ANALYZE_SCRIPT) if spec is None or spec.loader is None: raise RuntimeError(f"Cannot load {ANALYZE_SCRIPT}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def stable_id(prefix: str, value: str) -> str: text = value.strip() or "unknown" digest = hashlib.sha1(text.encode("utf-8")).hexdigest()[:10].upper() return f"{prefix}{digest}" def money_wan(value: Any) -> float | str: if isinstance(value, (int, float)): return round(value / 10000, 4) return "" def clean(value: Any) -> str: if value is None: return "" text = str(value).strip() return "" if text in {"--", "暂未公布", "nan", "None"} else text def product_keywords(title: str, keyword: str) -> str: text = f"{title} {keyword}" found = [token for token in PRODUCT_KEYWORDS if token in text] return "、".join(dict.fromkeys(found)) def product_confidence(title: str) -> str: if any(token in title for token in ["设备", "系统", "平台", "维保", "服务", "软硬件", "监管", "收银"]): return "中：由项目标题/关键词推断" return "低：原始Excel无产品明细，仅保留关键词" def style_cell(cell, fill: str | None = None, bold: bool = False, color: str = "1F1F1F") -> None: cell.font = Font(name=FONT, size=10, bold=bold, color=color) cell.alignment = Alignment(wrap_text=True, vertical="top") cell.border = BORDER if fill: cell.fill = PatternFill("solid", fgColor=fill) def write_sheet(wb: Workbook, name: str, rows: list[dict[str, Any]]) -> None: ws = wb.create_sheet(name[:31]) if not rows: rows = [{"说明": "无可抽取记录"}] headers = list(rows[0].keys()) ws.sheet_view.showGridLines = False ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=len(headers)) title = ws.cell(1, 1, name) title.font = Font(name=FONT, size=15, bold=True, color=WHITE) title.fill = PatternFill("solid", fgColor=BLUE) title.alignment = Alignment(horizontal="center", vertical="center") title.border = BORDER for col, header in enumerate(headers, 1): cell = ws.cell(2, col, header) style_cell(cell, LIGHT_BLUE, True, "14344A") cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) for row_index, row in enumerate(rows, 3): for col_index, header in enumerate(headers, 1): style_cell(ws.cell(row_index, col_index, row.get(header, ""))) for idx, header in enumerate(headers, 1): max_len = len(str(header)) for row in rows[:100]: max_len = max(max_len, len(str(row.get(header, ""))) // 2) ws.column_dimensions[get_column_letter(idx)].width = min(max(max_len + 4, 14), 46) for row in range(1, ws.max_row + 1): ws.row_dimensions[row].height = 34 ws.freeze_panes = "A3" ws.auto_filter.ref = f"A2:{get_column_letter(len(headers))}{ws.max_row}" def group_unique(rows: list[dict[str, Any]], name_key: str) -> dict[str, list[dict[str, Any]]]: groups: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in rows: name = clean(row.get(name_key)) if name: groups[name].append(row) return groups def joined(values: list[str], limit: int = 8) -> str: unique = [v for v in dict.fromkeys(clean(v) for v in values) if v] suffix = "" if len(unique) <= limit else f" 等{len(unique)}项" return "、".join(unique[:limit]) + suffix def amount_stats(items: list[dict[str, Any]]) -> dict[str, Any]: amounts = [row["金额数值"] for row in items if isinstance(row.get("金额数值"), (int, float))] total = sum(amounts) return { "披露金额项目数": len(amounts), "金额合计_元": round(total, 2), "金额合计_万元": money_wan(total), "平均金额_万元": money_wan(total / len(amounts)) if amounts else "", "最高金额_万元": money_wan(max(amounts)) if amounts else "", } def build_tables(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: project_rows: list[dict[str, Any]] = [] relation_rows: list[dict[str, Any]] = [] product_rows: list[dict[str, Any]] = [] contact_rows: list[dict[str, Any]] = [] source_rows: list[dict[str, Any]] = [] for idx, row in enumerate(rows, 1): project_id = f"P{idx:04d}" buyer = clean(row.get("招标单位")) supplier = clean(row.get("中标单位")) buyer_id = stable_id("BUY", buyer) supplier_id = stable_id("BID", supplier) title = clean(row.get("项目名称")) keywords = product_keywords(title, clean(row.get("关键词"))) project_rows.append( { "项目ID": project_id, "项目名称": title, "关键词": clean(row.get("关键词")), "项目编号": clean(row.get("项目编号")), "信息发布时间": clean(row.get("发布时间") or row.get("信息发布时间")), "月份": clean(row.get("月份")), "发布省份": clean(row.get("发布省份")), "发布市级": clean(row.get("发布市级")), "发布区级": clean(row.get("发布区级")), "中标阶段": clean(row.get("中标阶段")), "中标金额_元": row.get("金额数值") or "", "中标金额_万元": row.get("金额_万元") or "", "招标公司ID": buyer_id if buyer else "", "招标公司": buyer, "投标公司ID": supplier_id if supplier else "", "投标公司/中标单位": supplier, "需求类型": clean(row.get("需求类型")), "客户类型": clean(row.get("客户类型")), "合同开始时间": clean(row.get("合同开始时间")), "合同结束时间": clean(row.get("合同结束时间")), "合同工期天数": row.get("合同工期天数") or "", "官网查看地址": clean(row.get("官网查看地址")), } ) if buyer and supplier: relation_rows.append( { "关系ID": f"REL{idx:04d}", "项目ID": project_id, "项目名称": title, "招标公司ID": buyer_id, "招标公司": buyer, "投标公司ID": supplier_id, "投标公司": supplier, "关系类型": "招标方-中标方", "披露阶段": clean(row.get("中标阶段")), "中标金额_万元": row.get("金额_万元") or "", "产品/需求类型": clean(row.get("需求类型")), "来源链接": clean(row.get("官网查看地址")), "说明": "原始Excel仅披露中标单位；未披露全部参与投标公司名单。", } ) product_rows.append( { "产品记录ID": f"PROD{idx:04d}", "项目ID": project_id, "项目名称": title, "投标公司ID": supplier_id if supplier else "", "投标公司": supplier, "招标公司": buyer, "产品/需求类型": clean(row.get("需求类型")), "抽取关键词": keywords, "软硬件/服务判断": clean(row.get("需求类型")), "中标金额_万元": row.get("金额_万元") or "", "抽取置信度": product_confidence(title), "来源字段": "项目名称、关键词", "备注": "原始Excel无逐项设备/软件明细；本表为项目级产品/需求抽取。", } ) for role, company, company_id, person, phone in [ ("招标公司", buyer, buyer_id, row.get("招标单位联系人"), row.get("招标单位联系人电话")), ("投标公司/中标单位", supplier, supplier_id, row.get("中标单位联系人"), row.get("中标单位联系人电话")), ]: person_text = clean(person) phone_text = clean(phone) if company and (person_text or phone_text): contact_rows.append( { "联系人ID": stable_id("CT", f"{company}|{person_text}|{phone_text}"), "公司ID": company_id, "公司名称": company, "公司角色": role, "联系人": person_text, "联系电话": phone_text, "来源项目ID": project_id, "来源项目": title, "来源链接": clean(row.get("官网查看地址")), } ) source_rows.append( { "来源ID": f"SRC{idx:04d}", "项目ID": project_id, "项目名称": title, "来源类型": "乙方宝中标通知列表", "来源链接": clean(row.get("官网查看地址")), "采集字段": "项目名称、招标单位、中标单位、金额、地区、联系人、合同工期", "证据等级": "列表字段", "备注": "需要进入详情页/附件才能补齐全部投标人、资质证书、设备参数。", } ) buyer_groups = group_unique(rows, "招标单位") supplier_groups = group_unique(rows, "中标单位") tender_company_rows = [] for name, items in sorted(buyer_groups.items(), key=lambda kv: (-len(kv[1]), kv[0])): stats = amount_stats(items) tender_company_rows.append( { "招标公司ID": stable_id("BUY", name), "招标公司名称": name, "客户类型": joined([row.get("客户类型", "") for row in items], 5), "项目数": len(items), **stats, "覆盖省份": joined([row.get("发布省份", "") for row in items], 8), "覆盖城市": joined([row.get("发布市级", "") for row in items], 8), "联系人": joined([row.get("招标单位联系人", "") for row in items], 5), "联系电话": joined([row.get("招标单位联系人电话", "") for row in items], 5), "来源项目示例": clean(items[0].get("项目名称")), "来源链接示例": clean(items[0].get("官网查看地址")), "抽取状态": "原始Excel直接抽取", } ) bidder_company_rows = [] for name, items in sorted(supplier_groups.items(), key=lambda kv: (-len(kv[1]), kv[0])): stats = amount_stats(items) dates = sorted([clean(row.get("发布时间") or row.get("信息发布时间")) for row in items if clean(row.get("发布时间") or row.get("信息发布时间"))]) bidder_company_rows.append( { "投标公司ID": stable_id("BID", name), "投标公司名称": name, "原始披露角色": "中标单位", "中标项目数": len(items), **stats, "服务客户类型": joined([row.get("客户类型", "") for row in items], 8), "覆盖省份": joined([row.get("发布省份", "") for row in items], 8), "需求类型": joined([row.get("需求类型", "") for row in items], 8), "最早披露日期": dates[0] if dates else "", "最近披露日期": dates[-1] if dates else "", "联系人": joined([row.get("中标单位联系人", "") for row in items], 5), "联系电话": joined([row.get("中标单位联系人电话", "") for row in items], 5), "来源项目示例": clean(items[0].get("项目名称")), "来源链接示例": clean(items[0].get("官网查看地址")), "说明": "原始Excel字段名为中标单位；未披露未中标投标公司。", } ) qualification_rows = [ { "投标公司ID": row["投标公司ID"], "投标公司名称": row["投标公司名称"], "资质名称": "", "资质类型": "", "证书编号": "", "有效期起": "", "有效期止": "", "状态": "待补抓", "来源项目示例": row["来源项目示例"], "来源链接示例": row["来源链接示例"], "抽取状态": "原始Excel未披露资质字段", "备注": "需从企查查/天眼查/爱企查、招标公告、投标文件或中标公示附件补充。", } for row in bidder_company_rows ] region_groups: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) for row in rows: region_groups[(clean(row.get("发布省份")), clean(row.get("发布市级")), clean(row.get("发布区级")))].append(row) region_rows = [] for (province, city, district), items in sorted(region_groups.items(), key=lambda kv: (-len(kv[1]), kv[0])): stats = amount_stats(items) region_rows.append( { "发布省份": province, "发布市级": city, "发布区级": district, "项目数": len(items), **stats, "主要招标公司": joined([row.get("招标单位", "") for row in items], 5), "主要中标单位": joined([row.get("中标单位", "") for row in items], 5), } ) coverage_rows = [ {"表名": "招标公司表", "记录数": len(tender_company_rows), "抽取方式": "原始Excel直接抽取并按招标单位聚合", "覆盖情况": "可用"}, {"表名": "投标公司表", "记录数": len(bidder_company_rows), "抽取方式": "原始Excel中标单位聚合", "覆盖情况": "仅含中标单位，不含全部投标人"}, {"表名": "投标产品表", "记录数": len(product_rows), "抽取方式": "项目名称/关键词/需求类型推断", "覆盖情况": "项目级产品/需求可用，缺逐项参数"}, {"表名": "投标公司资质表", "记录数": len(qualification_rows), "抽取方式": "生成待补抓清单", "覆盖情况": "原始Excel未披露资质证书"}, {"表名": "招投标项目表", "记录数": len(project_rows), "抽取方式": "原始Excel直接抽取", "覆盖情况": "可用"}, {"表名": "招标-投标关系表", "记录数": len(relation_rows), "抽取方式": "招标单位 -> 中标单位", "覆盖情况": "可用，但非完整投标名单"}, {"表名": "联系人表", "记录数": len(contact_rows), "抽取方式": "联系人/电话非空记录", "覆盖情况": "部分可用"}, {"表名": "来源链接表", "记录数": len(source_rows), "抽取方式": "官网查看地址", "覆盖情况": "可用"}, {"表名": "地区统计表", "记录数": len(region_rows), "抽取方式": "省市区聚合", "覆盖情况": "可用"}, ] return { "抽取覆盖说明": coverage_rows, "招投标项目表": project_rows, "招标公司表": tender_company_rows, "投标公司表": bidder_company_rows, "投标产品表": product_rows, "投标公司资质表": qualification_rows, "招标-投标关系表": relation_rows, "联系人表": contact_rows, "来源链接表": source_rows, "地区统计表": region_rows, } def build_workbook(tables: dict[str, list[dict[str, Any]]]) -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) wb = Workbook() wb.remove(wb.active) for sheet, rows in tables.items(): write_sheet(wb, sheet, rows) wb.save(OUT_XLSX) def main() -> None: module = load_analyze_module() csv_path = module.input_to_csv(RAW_PATH) rows = module.read_rows(csv_path) tables = build_tables(rows) build_workbook(tables) for sheet, sheet_rows in tables.items(): print(sheet, len(sheet_rows)) print(OUT_XLSX) if __name__ == "__main__": main()