#!/usr/bin/env python3 """Build an interactive HTML dashboard from extracted tender entity tables.""" from __future__ import annotations import json from collections import Counter, defaultdict from pathlib import Path from typing import Any from openpyxl import load_workbook ROOT = Path(__file__).resolve().parents[1] SOURCE_XLSX = ROOT / "docs" / "deliverables" / "20260617-raw-excel-entity-tables" / "原始Excel招投标实体抽取表.xlsx" OUT_DIR = ROOT / "docs" / "deliverables" / "20260617-entity-map-dashboard" OUT_HTML = OUT_DIR / "招投标实体地图看板.html" PROVINCE_META = { "北京": ("北京市", "110000"), "天津": ("天津市", "120000"), "河北": ("河北省", "130000"), "山西": ("山西省", "140000"), "内蒙古": ("内蒙古自治区", "150000"), "辽宁": ("辽宁省", "210000"), "吉林": ("吉林省", "220000"), "黑龙江": ("黑龙江省", "230000"), "上海": ("上海市", "310000"), "江苏": ("江苏省", "320000"), "浙江": ("浙江省", "330000"), "安徽": ("安徽省", "340000"), "福建": ("福建省", "350000"), "江西": ("江西省", "360000"), "山东": ("山东省", "370000"), "河南": ("河南省", "410000"), "湖北": ("湖北省", "420000"), "湖南": ("湖南省", "430000"), "广东": ("广东省", "440000"), "广西": ("广西壮族自治区", "450000"), "海南": ("海南省", "460000"), "重庆": ("重庆市", "500000"), "四川": ("四川省", "510000"), "贵州": ("贵州省", "520000"), "云南": ("云南省", "530000"), "西藏": ("西藏自治区", "540000"), "陕西": ("陕西省", "610000"), "甘肃": ("甘肃省", "620000"), "青海": ("青海省", "630000"), "宁夏": ("宁夏回族自治区", "640000"), "新疆": ("新疆维吾尔自治区", "650000"), "台湾": ("台湾省", "710000"), "香港": ("香港特别行政区", "810000"), "澳门": ("澳门特别行政区", "820000"), } MUNICIPALITIES = {"北京", "天津", "上海", "重庆"} def clean(value: Any) -> str: if value is None: return "" text = str(value).strip() return "" if text in {"--", "None", "nan"} else text def as_float(value: Any) -> float: if isinstance(value, (int, float)): return float(value) text = clean(value).replace(",", "") if not text: return 0.0 try: return float(text) except ValueError: return 0.0 def read_sheet(sheet_name: str) -> list[dict[str, Any]]: wb = load_workbook(SOURCE_XLSX, read_only=True, data_only=True) ws = wb[sheet_name] headers = [clean(cell.value) for cell in ws[2]] rows = [] for row in ws.iter_rows(min_row=3, values_only=True): item = {headers[idx]: row[idx] for idx in range(len(headers)) if headers[idx]} if any(clean(value) for value in item.values()): rows.append(item) wb.close() return rows def province_display(raw: str) -> str: return PROVINCE_META.get(raw, (raw, ""))[0] def city_display(raw: str) -> str: if not raw: return "" if raw.endswith(("市", "州", "盟", "地区", "自治州")): return raw return f"{raw}市" def normalize_project(row: dict[str, Any]) -> dict[str, Any]: province = clean(row.get("发布省份")) city = clean(row.get("发布市级")) district = clean(row.get("发布区级")) amount_wan = as_float(row.get("中标金额_万元")) return { "projectId": clean(row.get("项目ID")), "title": clean(row.get("项目名称")), "keyword": clean(row.get("关键词")), "projectNo": clean(row.get("项目编号")), "date": clean(row.get("信息发布时间")), "month": clean(row.get("月份")), "province": province, "provinceName": province_display(province), "provinceCode": PROVINCE_META.get(province, ("", ""))[1], "city": city, "cityName": city_display(city), "district": district, "stage": clean(row.get("中标阶段")), "amountWan": amount_wan, "buyer": clean(row.get("招标公司")), "supplier": clean(row.get("投标公司/中标单位")), "demandType": clean(row.get("需求类型")), "clientType": clean(row.get("客户类型")), "durationDays": row.get("合同工期天数") or "", "url": clean(row.get("官网查看地址")), } def aggregate(rows: list[dict[str, Any]], key: str) -> list[dict[str, Any]]: groups: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in rows: name = clean(row.get(key)) if name: groups[name].append(row) output = [] for name, items in groups.items(): amount = sum(row["amountWan"] for row in items) output.append( { "name": name, "projectCount": len(items), "amountWan": round(amount, 4), "avgAmountWan": round(amount / len(items), 4) if items else 0, } ) return sorted(output, key=lambda item: (item["amountWan"], item["projectCount"]), reverse=True) def build_payload() -> dict[str, Any]: projects = [normalize_project(row) for row in read_sheet("招投标项目表")] companies = read_sheet("招标公司表") suppliers = read_sheet("投标公司表") products = read_sheet("投标产品表") qualifications = read_sheet("投标公司资质表") province_rows = aggregate(projects, "provinceName") demand_counter = Counter(row["demandType"] or "未识别" for row in projects) client_counter = Counter(row["clientType"] or "未识别" for row in projects) supplier_amount: dict[str, float] = defaultdict(float) supplier_count: Counter[str] = Counter() for row in projects: if row["supplier"]: supplier_amount[row["supplier"]] += row["amountWan"] supplier_count[row["supplier"]] += 1 top_suppliers = sorted( [ {"name": name, "amountWan": round(amount, 4), "projectCount": supplier_count[name]} for name, amount in supplier_amount.items() ], key=lambda item: (item["amountWan"], item["projectCount"]), reverse=True, )[:20] return { "meta": { "source": str(SOURCE_XLSX.relative_to(ROOT)), "projectCount": len(projects), "buyerCount": len(companies), "supplierCount": len(suppliers), "productCount": len(products), "qualificationTodoCount": len(qualifications), "totalAmountWan": round(sum(row["amountWan"] for row in projects), 4), }, "provinceMeta": { raw: {"name": display, "code": code} for raw, (display, code) in PROVINCE_META.items() }, "municipalities": sorted(MUNICIPALITIES), "projects": projects, "provinceSummary": province_rows, "demandSummary": [{"name": k, "value": v} for k, v in demand_counter.most_common()], "clientSummary": [{"name": k, "value": v} for k, v in client_counter.most_common()], "topSuppliers": top_suppliers, } def html_escape_json(payload: dict[str, Any]) -> str: return json.dumps(payload, ensure_ascii=False).replace(" str: data_json = html_escape_json(payload) return f""" 招投标实体地图看板

招投标实体地图看板

把原始 Excel 抽取出的项目、招标公司、投标公司、产品需求和来源链接作为页面内置数据库，按地区进行地图钻取和报表联动。

地区金额地图
全国项目金额与项目数分布
全国
地图加载中。
地区排行
随地图层级切换
中标单位排行
按披露金额
需求类型分布
项目标题与关键词分类
当前区域报表详情
区域、项目数、金额、主要招标公司和中标单位。
区域项目数金额合计主要招标公司主要中标单位
""" def main() -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) payload = build_payload() OUT_HTML.write_text(build_html(payload), encoding="utf-8") print(OUT_HTML) print(json.dumps(payload["meta"], ensure_ascii=False, indent=2)) if __name__ == "__main__": main()