#!/usr/bin/env python3 """Build relational Excel and SQLite assets for tender intelligence.""" from __future__ import annotations import importlib.util import sqlite3 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] DETAIL_SCRIPT = ROOT / "scripts" / "build_full_candidate_tender_excels.py" OUT_DIR = ROOT / "docs" / "deliverables" / "20260615-tender-relational-model" 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"), ) SCHEMA_SQL = """ PRAGMA foreign_keys = ON; CREATE TABLE IF NOT EXISTS sources ( source_id TEXT PRIMARY KEY, source_type TEXT, title TEXT, url_or_path TEXT, evidence_level TEXT, captured_at TEXT, notes TEXT ); CREATE TABLE IF NOT EXISTS parties ( party_id TEXT PRIMARY KEY, name TEXT NOT NULL, identifier_scheme TEXT, identifier_id TEXT, roles TEXT, legal_form TEXT, registered_capital TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS contracting_processes ( process_id TEXT PRIMARY KEY, ocid TEXT, title TEXT NOT NULL, buyer_party_id TEXT, procuring_entity_party_id TEXT, region TEXT, status TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (buyer_party_id) REFERENCES parties(party_id), FOREIGN KEY (procuring_entity_party_id) REFERENCES parties(party_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS tenders ( tender_id TEXT PRIMARY KEY, process_id TEXT NOT NULL, title TEXT, tender_no TEXT, procurement_method TEXT, budget_amount REAL, budget_currency TEXT, tender_status TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (process_id) REFERENCES contracting_processes(process_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS lots ( lot_id TEXT PRIMARY KEY, tender_id TEXT NOT NULL, title TEXT, status TEXT, budget_amount REAL, notes TEXT, FOREIGN KEY (tender_id) REFERENCES tenders(tender_id) ); CREATE TABLE IF NOT EXISTS products ( product_id TEXT PRIMARY KEY, name TEXT NOT NULL, category TEXT, brand TEXT, model TEXT, classification_scheme TEXT, classification_id TEXT, notes TEXT ); CREATE TABLE IF NOT EXISTS tender_items ( item_id TEXT PRIMARY KEY, tender_id TEXT NOT NULL, lot_id TEXT, product_id TEXT, description TEXT, quantity REAL, unit TEXT, source_id TEXT, confidence TEXT, notes TEXT, FOREIGN KEY (tender_id) REFERENCES tenders(tender_id), FOREIGN KEY (lot_id) REFERENCES lots(lot_id), FOREIGN KEY (product_id) REFERENCES products(product_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS bids ( bid_id TEXT PRIMARY KEY, tender_id TEXT NOT NULL, bidder_party_id TEXT NOT NULL, lot_id TEXT, bid_amount REAL, currency TEXT, rank_no INTEGER, status TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (tender_id) REFERENCES tenders(tender_id), FOREIGN KEY (bidder_party_id) REFERENCES parties(party_id), FOREIGN KEY (lot_id) REFERENCES lots(lot_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS awards ( award_id TEXT PRIMARY KEY, process_id TEXT NOT NULL, tender_id TEXT, supplier_party_id TEXT NOT NULL, award_amount REAL, currency TEXT, award_date TEXT, status TEXT, source_id TEXT, reason_summary TEXT, FOREIGN KEY (process_id) REFERENCES contracting_processes(process_id), FOREIGN KEY (tender_id) REFERENCES tenders(tender_id), FOREIGN KEY (supplier_party_id) REFERENCES parties(party_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS contracts ( contract_id TEXT PRIMARY KEY, award_id TEXT, process_id TEXT NOT NULL, supplier_party_id TEXT, contract_amount REAL, currency TEXT, start_date TEXT, end_date TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (award_id) REFERENCES awards(award_id), FOREIGN KEY (process_id) REFERENCES contracting_processes(process_id), FOREIGN KEY (supplier_party_id) REFERENCES parties(party_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS qualifications ( qualification_id TEXT PRIMARY KEY, name TEXT NOT NULL, qualification_type TEXT, issuer TEXT, standard_or_code TEXT, notes TEXT ); CREATE TABLE IF NOT EXISTS party_qualifications ( party_qualification_id TEXT PRIMARY KEY, party_id TEXT NOT NULL, qualification_id TEXT NOT NULL, certificate_no TEXT, valid_from TEXT, valid_to TEXT, status TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (party_id) REFERENCES parties(party_id), FOREIGN KEY (qualification_id) REFERENCES qualifications(qualification_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS party_products ( party_product_id TEXT PRIMARY KEY, party_id TEXT NOT NULL, product_id TEXT NOT NULL, relation_type TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (party_id) REFERENCES parties(party_id), FOREIGN KEY (product_id) REFERENCES products(product_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS scores ( score_id TEXT PRIMARY KEY, bid_id TEXT, criterion TEXT, score REAL, max_score REAL, source_id TEXT, notes TEXT, FOREIGN KEY (bid_id) REFERENCES bids(bid_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); CREATE TABLE IF NOT EXISTS party_relationships ( relationship_id TEXT PRIMARY KEY, source_party_id TEXT NOT NULL, target_party_id TEXT NOT NULL, relationship_type TEXT, share_percent REAL, process_id TEXT, product_id TEXT, source_id TEXT, notes TEXT, FOREIGN KEY (source_party_id) REFERENCES parties(party_id), FOREIGN KEY (target_party_id) REFERENCES parties(party_id), FOREIGN KEY (process_id) REFERENCES contracting_processes(process_id), FOREIGN KEY (product_id) REFERENCES products(product_id), FOREIGN KEY (source_id) REFERENCES sources(source_id) ); """ TABLES = [ "sources", "parties", "contracting_processes", "tenders", "lots", "products", "tender_items", "bids", "awards", "contracts", "qualifications", "party_qualifications", "party_products", "scores", "party_relationships", ] def load_detail_module() -> Any: spec = importlib.util.spec_from_file_location("full_candidate_builder", DETAIL_SCRIPT) if spec is None or spec.loader is None: raise RuntimeError(f"Cannot load {DETAIL_SCRIPT}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def style(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, headers: list[str], rows: list[dict[str, Any]]) -> None: ws = wb.create_sheet(name) 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, 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(ws.cell(row_index, col_index, row.get(header, ""))) for idx, header in enumerate(headers, 1): ws.column_dimensions[get_column_letter(idx)].width = min(max(len(header) + 4, 16), 42) for row in range(1, ws.max_row + 1): ws.row_dimensions[row].height = 38 ws.freeze_panes = "A3" ws.auto_filter.ref = f"A2:{get_column_letter(len(headers))}{ws.max_row}" def safe_id(prefix: str, text: str) -> str: keep = "".join(ch for ch in text if ch.isalnum()) return f"{prefix}_{keep[:18]}" if keep else prefix def add_unique(rows: list[dict[str, Any]], key: str, item: dict[str, Any]) -> None: if not any(row.get(key) == item.get(key) for row in rows): rows.append(item) def build_rows() -> dict[str, list[dict[str, Any]]]: module = load_detail_module() rows = {table: [] for table in TABLES} for project_index, spec in enumerate(module.build_specs(), 1): project_code = f"P{project_index:03d}" project_title = spec["overview"][0][1] buyer = "" buyer_is_disclosed = False for field, value, *_ in spec["overview"]: if field in {"招标人", "采购人"}: buyer = value buyer_is_disclosed = True if not buyer: buyer = f"未披露采购人（{project_code}）" buyer_id = safe_id("party", buyer) if buyer_is_disclosed else f"party_unknown_buyer_{project_code}" source_id = f"S{project_index:03d}_official" process_id = f"CP{project_index:03d}" tender_id = f"T{project_index:03d}" add_unique(rows["sources"], "source_id", { "source_id": source_id, "source_type": "official_or_project_sources", "title": f"{project_title}来源集合", "url_or_path": "\n".join(spec["sources"]), "evidence_level": "mixed", "captured_at": "2026-06-15", "notes": "由详细版项目资料汇总生成", }) add_unique(rows["parties"], "party_id", { "party_id": buyer_id, "name": buyer, "identifier_scheme": "", "identifier_id": "", "roles": "buyer;procuringEntity", "legal_form": "", "registered_capital": "", "source_id": source_id, "notes": "采购/招标主体，参考 OCDS parties.roles", }) rows["contracting_processes"].append({ "process_id": process_id, "ocid": f"ocds-local-{project_code}", "title": project_title, "buyer_party_id": buyer_id, "procuring_entity_party_id": buyer_id, "region": "", "status": "compiled", "source_id": source_id, "notes": "参考 OCDS contracting process/release", }) rows["tenders"].append({ "tender_id": tender_id, "process_id": process_id, "title": project_title, "tender_no": next((value for field, value, *_ in spec["overview"] if field in {"招标编号", "项目编号"}), ""), "procurement_method": "", "budget_amount": "", "budget_currency": "CNY", "tender_status": "awarded_or_candidates_disclosed", "source_id": source_id, "notes": "项目级招标/采购记录", }) for bidder_index, bidder in enumerate(spec["bidders"], 1): rank, name, amount, score, role, analysis = bidder party_id = safe_id("party", name) add_unique(rows["parties"], "party_id", { "party_id": party_id, "name": name, "identifier_scheme": "", "identifier_id": "", "roles": "bidder;supplier" if bidder_index == 1 else "bidder", "legal_form": "", "registered_capital": "", "source_id": source_id, "notes": role, }) bid_id = f"B{project_index:03d}_{bidder_index:02d}" rows["bids"].append({ "bid_id": bid_id, "tender_id": tender_id, "bidder_party_id": party_id, "lot_id": "", "bid_amount": amount, "currency": "CNY", "rank_no": bidder_index, "status": role, "source_id": source_id, "notes": analysis, }) rows["party_relationships"].append({ "relationship_id": f"REL{project_index:03d}_{bidder_index:02d}", "source_party_id": buyer_id, "target_party_id": party_id, "relationship_type": "buyer-bidder" if bidder_index > 1 else "buyer-awardedSupplier", "share_percent": "", "process_id": process_id, "product_id": "", "source_id": source_id, "notes": "通过投标/候选/中标关系连接，参考 OCDS parties + bids/awards", }) if bidder_index == 1: award_amount = amount rows["awards"].append({ "award_id": f"A{project_index:03d}", "process_id": process_id, "tender_id": tender_id, "supplier_party_id": party_id, "award_amount": award_amount, "currency": "CNY", "award_date": "", "status": role, "source_id": source_id, "reason_summary": "；".join(item[1] for item in spec["reason"][:2]), }) if score and score != "未披露": rows["scores"].append({ "score_id": f"SC{project_index:03d}_{bidder_index:02d}", "bid_id": bid_id, "criterion": "total_or_public_score", "score": score, "max_score": "", "source_id": source_id, "notes": "公开得分或候选表得分字段", }) for item_index, product in enumerate(spec["products"], 1): product_id = f"PR{project_index:03d}_{item_index:03d}" product_name = product[1] rows["products"].append({ "product_id": product_id, "name": product_name, "category": product[1], "brand": product[7] if len(product) > 7 else "", "model": product[8] if len(product) > 8 else "", "classification_scheme": "local-smart-canteen", "classification_id": "", "notes": product[2] if len(product) > 2 else "", }) rows["tender_items"].append({ "item_id": f"I{project_index:03d}_{item_index:03d}", "tender_id": tender_id, "lot_id": "", "product_id": product_id, "description": product[2] if len(product) > 2 else "", "quantity": product[10] if len(product) > 10 else "", "unit": "", "source_id": source_id, "confidence": product[11] if len(product) > 11 else "", "notes": product[11] if len(product) > 11 else "", }) supplier_name = product[3] if len(product) > 3 else "" if supplier_name and supplier_name not in {"候选供应商", "五家入围候选"}: party_id = safe_id("party", supplier_name) add_unique(rows["parties"], "party_id", { "party_id": party_id, "name": supplier_name, "identifier_scheme": "", "identifier_id": "", "roles": "supplier;manufacturer", "legal_form": "", "registered_capital": "", "source_id": source_id, "notes": "产品清单供应商/制造商", }) rows["party_products"].append({ "party_product_id": f"PP{project_index:03d}_{item_index:03d}", "party_id": party_id, "product_id": product_id, "relation_type": "supplier_or_manufacturer", "source_id": source_id, "notes": "产品清单关系", }) for qual_index, qual in enumerate(spec["qualifications"], 1): qualification_id = f"Q{project_index:03d}_{qual_index:03d}" rows["qualifications"].append({ "qualification_id": qualification_id, "name": qual[1], "qualification_type": "tender_requirement_or_supplier_qualification", "issuer": "", "standard_or_code": "", "notes": qual[2], }) for bidder_index, bidder in enumerate(spec["bidders"][:3], 1): party_id = safe_id("party", bidder[1]) rows["party_qualifications"].append({ "party_qualification_id": f"PQ{project_index:03d}_{qual_index:03d}_{bidder_index:02d}", "party_id": party_id, "qualification_id": qualification_id, "certificate_no": "", "valid_from": "", "valid_to": "", "status": qual[3], "source_id": source_id, "notes": "若为招标资格要求，表示该公司需满足/候选公示显示满足；证书编号待企业库复核", }) return rows def create_sqlite(rows_by_table: dict[str, list[dict[str, Any]]], db_path: Path) -> None: if db_path.exists(): db_path.unlink() conn = sqlite3.connect(db_path) try: conn.executescript(SCHEMA_SQL) for table, rows in rows_by_table.items(): if not rows: continue headers = list(rows[0].keys()) placeholders = ", ".join(["?"] * len(headers)) sql = f"INSERT INTO {table} ({', '.join(headers)}) VALUES ({placeholders})" conn.executemany(sql, [[None if row.get(header, "") == "" else row.get(header, "") for header in headers] for row in rows]) conn.commit() finally: conn.close() def create_workbook(rows_by_table: dict[str, list[dict[str, Any]]], workbook_path: Path) -> None: wb = Workbook() wb.remove(wb.active) for table in TABLES: rows = rows_by_table[table] headers = list(rows[0].keys()) if rows else [] if not headers: continue write_sheet(wb, table, headers, rows) wb.save(workbook_path) def create_business_view_workbook(rows_by_table: dict[str, list[dict[str, Any]]], workbook_path: Path) -> None: parties = {row["party_id"]: row for row in rows_by_table["parties"]} tenders = {row["tender_id"]: row for row in rows_by_table["tenders"]} processes = {row["process_id"]: row for row in rows_by_table["contracting_processes"]} products = {row["product_id"]: row for row in rows_by_table["products"]} qualifications = {row["qualification_id"]: row for row in rows_by_table["qualifications"]} awards_by_tender = {row["tender_id"]: row for row in rows_by_table["awards"]} sheets: list[tuple[str, list[dict[str, Any]]]] = [] sheets.append(( "说明", [ { "用途": "业务视图", "说明": "这份 Excel 是给人工查看和补录用的精简版；底层标准表见“招投标关系模型模板.xlsx”和 SQLite。", "对应标准": "OCDS: parties/tender/bids/awards/items；BODS: party_relationships。", "注意": "不要用公司名称当唯一主键；拿到统一社会信用代码后填入 parties.identifier_id。", } ], )) sheets.append(( "资质表", [ { "资质ID": row["qualification_id"], "资质名称": row["name"], "资质类型": row["qualification_type"], "发证机构": row["issuer"], "标准或编号": row["standard_or_code"], "备注": row["notes"], } for row in rows_by_table["qualifications"] ], )) sheets.append(( "公司表", [ { "公司ID": row["party_id"], "公司名称": row["name"], "角色": row["roles"], "统一社会信用代码或平台ID": row["identifier_id"], "企业类型": row["legal_form"], "注册资本": row["registered_capital"], "来源ID": row["source_id"], "备注": row["notes"], } for row in rows_by_table["parties"] ], )) sheets.append(( "招标公司表", [ { "公司ID": row["party_id"], "招标公司": row["name"], "关联项目数": sum(1 for process in rows_by_table["contracting_processes"] if process["buyer_party_id"] == row["party_id"]), "来源ID": row["source_id"], "备注": row["notes"], } for row in rows_by_table["parties"] if "buyer" in row["roles"] or "procuringEntity" in row["roles"] ], )) sheets.append(( "投标公司表", [ { "公司ID": row["party_id"], "投标公司": row["name"], "投标次数": sum(1 for bid in rows_by_table["bids"] if bid["bidder_party_id"] == row["party_id"]), "中标次数": sum(1 for award in rows_by_table["awards"] if award["supplier_party_id"] == row["party_id"]), "角色": row["roles"], "来源ID": row["source_id"], "备注": row["notes"], } for row in rows_by_table["parties"] if "bidder" in row["roles"] or "supplier" in row["roles"] ], )) sheets.append(( "项目表", [ { "项目ID": process["process_id"], "项目名称": process["title"], "招标公司": parties.get(process["buyer_party_id"], {}).get("name", ""), "区域": process["region"], "状态": process["status"], "来源ID": process["source_id"], } for process in rows_by_table["contracting_processes"] ], )) sheets.append(( "项目-投标关系", [ { "投标ID": bid["bid_id"], "项目名称": tenders.get(bid["tender_id"], {}).get("title", ""), "投标公司": parties.get(bid["bidder_party_id"], {}).get("name", ""), "报价": bid["bid_amount"], "排名": bid["rank_no"], "状态": bid["status"], "是否中标": "是" if awards_by_tender.get(bid["tender_id"], {}).get("supplier_party_id") == bid["bidder_party_id"] else "否", "来源ID": bid["source_id"], "说明": bid["notes"], } for bid in rows_by_table["bids"] ], )) sheets.append(( "公司-资质关系", [ { "关系ID": row["party_qualification_id"], "公司": parties.get(row["party_id"], {}).get("name", ""), "资质": qualifications.get(row["qualification_id"], {}).get("name", ""), "证书编号": row["certificate_no"], "有效期起": row["valid_from"], "有效期止": row["valid_to"], "状态": row["status"], "来源ID": row["source_id"], "备注": row["notes"], } for row in rows_by_table["party_qualifications"] ], )) sheets.append(( "产品表", [ { "产品ID": row["product_id"], "产品名称": row["name"], "类别": row["category"], "品牌": row["brand"], "型号": row["model"], "分类体系": row["classification_scheme"], "备注": row["notes"], } for row in rows_by_table["products"] ], )) sheets.append(( "项目-产品关系", [ { "明细ID": row["item_id"], "项目名称": tenders.get(row["tender_id"], {}).get("title", ""), "产品": products.get(row["product_id"], {}).get("name", ""), "参数/说明": row["description"], "数量": row["quantity"], "单位": row["unit"], "置信度": row["confidence"], "来源ID": row["source_id"], } for row in rows_by_table["tender_items"] ], )) sheets.append(( "公司-产品关系", [ { "关系ID": row["party_product_id"], "公司": parties.get(row["party_id"], {}).get("name", ""), "产品": products.get(row["product_id"], {}).get("name", ""), "关系类型": row["relation_type"], "来源ID": row["source_id"], "备注": row["notes"], } for row in rows_by_table["party_products"] ], )) sheets.append(( "公司-公司关系", [ { "关系ID": row["relationship_id"], "源公司": parties.get(row["source_party_id"], {}).get("name", ""), "目标公司": parties.get(row["target_party_id"], {}).get("name", ""), "关系类型": row["relationship_type"], "项目": processes.get(row["process_id"], {}).get("title", ""), "关联产品": products.get(row["product_id"], {}).get("name", ""), "来源ID": row["source_id"], "备注": row["notes"], } for row in rows_by_table["party_relationships"] ], )) wb = Workbook() wb.remove(wb.active) for sheet_name, rows in sheets: headers = list(rows[0].keys()) if rows else [] if headers: write_sheet(wb, sheet_name, headers, rows) wb.save(workbook_path) def write_schema_files() -> None: (OUT_DIR / "schema.sql").write_text(SCHEMA_SQL.strip() + "\n", encoding="utf-8") html = """ 招投标关系模型说明

招投标关系模型说明

基于 OCDS 的采购过程结构，参考 BODS 的企业关系思想，并保留 Excel/SQLite 两种落地形式。

借鉴来源

来源采用内容
Open Contracting Data Standard (OCDS)采用 parties、tender、bids、awards、contracts、items、documents 的基本结构。
Beneficial Ownership Data Standard (BODS)企业之间的控制/关联关系用 statement-like 的关系表表达。
OpenDataServices flatten-tool保留“结构化数据可平铺到 Excel、再回到数据库/JSON”的思路。
天眼查/企查查类产品只借鉴实体-资质-产品-投标关系的组织方式，不复制其专有字段或数据。

核心关系

parties 记录招标公司、投标公司、供应商、制造商；contracting_processes 和 tenders 记录项目；bids 连接投标公司与项目；awards 连接中标公司与项目；products、tender_items、party_products 连接产品、项目与公司；qualifications、party_qualifications 连接资质与公司；party_relationships 用于表达招标公司、投标公司、供应商和产品之间的补充关系。

交付文件

招投标关系模型模板.xlsx 是标准关系表；招投标关系模型业务视图.xlsx 是人工查看的精简业务表；tender_relational_schema.sqlite 和 schema.sql 用于数据库查询和后续程序化采集。

""" (OUT_DIR / "招投标关系模型说明.html").write_text(html, encoding="utf-8") def main() -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) rows = build_rows() create_workbook(rows, OUT_DIR / "招投标关系模型模板.xlsx") create_business_view_workbook(rows, OUT_DIR / "招投标关系模型业务视图.xlsx") create_sqlite(rows, OUT_DIR / "tender_relational_schema.sqlite") write_schema_files() for table in TABLES: print(table, len(rows[table])) if __name__ == "__main__": main()