#!/usr/bin/env python3 """Build 2023-2026 tender trend analysis from local historical Excel exports.""" from __future__ import annotations import csv import html import importlib.util import json import math import re import statistics from collections import Counter from datetime import datetime from pathlib import Path from typing import Any import pandas as pd from openpyxl import load_workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter ROOT = Path(__file__).resolve().parents[1] ANALYZE_SCRIPT = ROOT / "scripts" / "analyze_bid_data.py" OUT_DIR = ROOT / "docs" / "deliverables" / "20260617-multi-year-trend-analysis" OUT_HTML = OUT_DIR / "2023-2026智慧食堂招投标同比环比分析.html" OUT_XLSX = OUT_DIR / "2023-2026智慧食堂招投标同比环比分析.xlsx" SOURCES = [ { "year": 2023, "path": ROOT / "docs" / "deliverables" / "20260617-original-yearly-excel-reports" / "乙方宝智慧食堂2023年-2025年招投标数据_新版原始合并.xls", "label": "2023-2025 新版乙方宝合并导出", }, { "year": 2026, "path": ROOT / "data" / "raw" / "2026年1月-6月乙方宝中标通知20260612.xls", "label": "2026 1-6月乙方宝导出", }, ] OUTLIER_AMOUNT_WAN = 100000 OUTLIER_TITLE_PATTERN = "智慧食堂智能化建设及食材劳务总包项目" PRODUCT_KEYWORDS = { "软件平台/监管系统": ["平台", "系统", "软件", "监管", "一卡通", "信息化", "数字化", "大数据", "智慧校园", "营养", "膳食", "食安", "订餐", "结算", "支付"], "硬件终端/智能设备": ["设备", "终端", "刷脸", "消费机", "机具", "收银", "闸机", "留样机", "验货机", "晨检仪", "摄像头", "网关", "UPS", "机器人", "售饭台", "智能秤", "电子秤", "显示屏", "读卡器"], "餐饮运营/食材劳务": ["餐饮", "食堂服务", "运营", "劳务", "配送", "食材", "承包"], "工程建设/集成安装": ["建设", "改造", "装修", "安装", "集成", "弱电", "智慧园区", "工程"], } QUALIFICATION_KEYWORDS = [ "ISO9001", "ISO 9001", "ISO14001", "ISO 14001", "ISO45001", "ISO 45001", "CMMI", "ITSS", "信息安全管理体系", "质量管理体系", "环境管理体系", "职业健康安全管理体系", "软件著作权", "软著", "专利", "高新技术企业", "食品经营许可证", "安全生产许可证", "建筑工程施工总承包", "电子与智能化", "安全技术防范", "CMA", "CNAS", ] FONT = "Microsoft YaHei" BLUE = "1F4E79" LIGHT_BLUE = "D9EAF7" LIGHT_GREEN = "E8F4F2" 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"), ) 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 ANALYZE = load_analyze_module() def clean(value: Any) -> str: if value is None: return "" if isinstance(value, datetime): return value.strftime("%Y-%m-%d") text = str(value).strip().replace("\u3000", " ") text = " ".join(text.split()) return "" if text in {"--", "None", "nan", "null", "NULL"} else text def compact_excerpt(value: Any, limit: int = 6000) -> str: text = clean(value) if len(text) <= limit: return text return text[:limit] + "..." def normalize_company_name(value: Any) -> str: text = clean(value) text = text.replace("（", "(").replace("）", ")") text = re.sub(r"\s+", "", text) return text def parse_date(value: Any) -> datetime | None: if isinstance(value, datetime): return value text = clean(value) if not text: return None for fmt in ("%Y/%m/%d", "%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d %H:%M:%S", "%Y-%m-%d %H:%M:%S"): try: return datetime.strptime(text, fmt) except ValueError: pass return None def parse_amount(value: Any) -> float | None: if isinstance(value, (int, float)) and not isinstance(value, bool): if math.isnan(value): return None return float(value) text = clean(value).replace(",", "") for token in ["元", "￥", "¥"]: text = text.replace(token, "") if not text: return None try: return float(text) except ValueError: return None def parse_days(value: Any) -> int | str: text = clean(value).replace("天", "") if not text: return "" try: return int(float(text)) except ValueError: return "" def money_wan(value: float | int | None) -> str: if value is None: return "未披露" if abs(value) >= 10000: return f"{value / 10000:.2f}亿元" return f"{value:.2f}万元" def pct_text(value: float | None) -> str: if value is None or pd.isna(value): return "不可比" return f"{value * 100:.1f}%" def safe_pct(now: float | int | None, prev: float | int | None) -> float | None: if prev in (None, 0) or pd.isna(prev): return None return (float(now or 0) - float(prev)) / float(prev) def header_row(ws) -> int: for row_idx in range(1, min(ws.max_row or 1, 8) + 1): first = clean(ws.cell(row_idx, 1).value) second = clean(ws.cell(row_idx, 2).value) if first == "关键词" and second == "项目名称": return row_idx return 1 def normalize_raw_record(values: list[Any], source_year: int, source_label: str, source_file: Path, row_number: int) -> dict[str, Any] | None: if len(values) < 2 or not clean(values[1]): return None publish_date = parse_date(values[2] if len(values) > 2 else "") amount = parse_amount(values[8] if len(values) > 8 else "") title = clean(values[1]) buyer = clean(values[9] if len(values) > 9 else "") supplier = clean(values[12] if len(values) > 12 else "") keyword = clean(values[0] if values else "") notice_text = compact_excerpt(values[20] if len(values) > 20 else "") summary_text = compact_excerpt(values[19] if len(values) > 19 else "", 1200) row = { "来源年份": source_year, "来源文件": str(source_file), "来源说明": source_label, "源表行号": row_number, "关键词": keyword, "项目名称": title, "信息发布时间": publish_date.strftime("%Y-%m-%d") if publish_date else "", "项目编号": clean(values[3] if len(values) > 3 else ""), "发布省份": clean(values[4] if len(values) > 4 else ""), "发布市级": clean(values[5] if len(values) > 5 else ""), "发布区级": clean(values[6] if len(values) > 6 else ""), "中标阶段": clean(values[7] if len(values) > 7 else ""), "中标金额_元": amount if amount is not None else "", "中标金额_万元": round(amount / 10000, 4) if amount is not None else "", "招标单位": buyer, "中标单位": supplier, "供应商归一名称": normalize_company_name(supplier), "合同开始时间": clean(values[15] if len(values) > 15 else ""), "合同结束时间": clean(values[16] if len(values) > 16 else ""), "合同工期天数": parse_days(values[17] if len(values) > 17 else ""), "官网查看地址": clean(values[18] if len(values) > 18 else ""), "正文摘要": summary_text, "公告内容长度": len(clean(values[20] if len(values) > 20 else "")), "公告内容片段": notice_text, "包含附件": clean(values[21] if len(values) > 21 else ""), "需求类型": ANALYZE.classify_demand(title, keyword), "客户类型": ANALYZE.classify_client(buyer), } if publish_date: row["年份"] = publish_date.year row["月份"] = publish_date.strftime("%Y-%m") row["月份序号"] = publish_date.month row["季度"] = f"{publish_date.year}Q{((publish_date.month - 1) // 3) + 1}" row["半年度"] = "H1" if publish_date.month <= 6 else "H2" else: row["年份"] = source_year row["月份"] = f"{source_year}-未知" row["月份序号"] = "" row["季度"] = f"{source_year}Q未知" row["半年度"] = "未知" return row def read_xlsx_source(source: dict[str, Any]) -> list[dict[str, Any]]: path = source["path"] wb = load_workbook(path, read_only=True, data_only=True) ws = wb[wb.sheetnames[0]] start = header_row(ws) + 1 rows: list[dict[str, Any]] = [] for excel_row, values in enumerate(ws.iter_rows(min_row=start, values_only=True), start=start): item = normalize_raw_record(list(values), source["year"], source["label"], path, excel_row) if item: rows.append(item) wb.close() return rows def read_2026_source(source: dict[str, Any]) -> list[dict[str, Any]]: csv_path = ANALYZE.input_to_csv(source["path"]) rows = ANALYZE.read_rows(csv_path) output: list[dict[str, Any]] = [] for index, row in enumerate(rows, 1): amount = row.get("金额数值") publish_date = parse_date(row.get("发布时间") or row.get("信息发布时间")) output.append( { "来源年份": source["year"], "来源文件": str(source["path"]), "来源说明": source["label"], "源表行号": index + 2, "关键词": row.get("关键词", ""), "项目名称": row.get("项目名称", ""), "信息发布时间": row.get("发布时间", ""), "项目编号": row.get("项目编号", ""), "发布省份": row.get("发布省份", ""), "发布市级": row.get("发布市级", ""), "发布区级": row.get("发布区级", ""), "中标阶段": row.get("中标阶段", ""), "中标金额_元": amount if amount is not None else "", "中标金额_万元": row.get("金额_万元", ""), "招标单位": row.get("招标单位", ""), "中标单位": row.get("中标单位", ""), "供应商归一名称": normalize_company_name(row.get("中标单位", "")), "合同开始时间": row.get("合同开始时间", ""), "合同结束时间": row.get("合同结束时间", ""), "合同工期天数": row.get("合同工期天数", ""), "官网查看地址": row.get("官网查看地址", ""), "正文摘要": "", "公告内容长度": "", "公告内容片段": "", "包含附件": "", "需求类型": row.get("需求类型", ""), "客户类型": row.get("客户类型", ""), "年份": publish_date.year if publish_date else source["year"], "月份": row.get("月份", ""), "月份序号": publish_date.month if publish_date else "", "季度": f"{publish_date.year}Q{((publish_date.month - 1) // 3) + 1}" if publish_date else f"{source['year']}Q未知", "半年度": "H1" if publish_date and publish_date.month <= 6 else ("H2" if publish_date else "未知"), } ) return output def load_all_records() -> pd.DataFrame: records: list[dict[str, Any]] = [] for source in SOURCES: path = source["path"] if not path.exists(): raise FileNotFoundError(path) if path.suffix.lower() == ".xls": records.extend(read_2026_source(source)) else: records.extend(read_xlsx_source(source)) df = pd.DataFrame(records) df = df[df["项目名称"].astype(str).str.strip() != ""].copy() df["金额数值"] = pd.to_numeric(df["中标金额_元"], errors="coerce") df["金额_万元"] = pd.to_numeric(df["中标金额_万元"], errors="coerce") df["月份日期"] = pd.to_datetime(df["月份"] + "-01", errors="coerce") df["是否披露金额"] = df["金额数值"].notna() df["是否有链接"] = df["官网查看地址"].astype(str).str.startswith("http") df["是否有正文"] = pd.to_numeric(df["公告内容长度"], errors="coerce").fillna(0) > 0 return df def outlier_mask(df: pd.DataFrame) -> pd.Series: amounts = pd.to_numeric(df["金额_万元"], errors="coerce").fillna(0) return ( (pd.to_numeric(df["年份"], errors="coerce") == 2026) & (amounts >= OUTLIER_AMOUNT_WAN) & df["项目名称"].astype(str).str.contains(OUTLIER_TITLE_PATTERN, na=False) ) def outlier_projects(df: pd.DataFrame) -> pd.DataFrame: cols = [ "年份", "月份", "项目名称", "发布省份", "发布市级", "中标阶段", "金额_万元", "招标单位", "中标单位", "需求类型", "客户类型", "官网查看地址", "来源说明", "源表行号", ] out = df[outlier_mask(df)].copy() if out.empty: return pd.DataFrame(columns=["剔除原因", *cols]) out["剔除原因"] = "13亿级异常项目，金额远高于常规智慧食堂项目，影响同比/环比可比性" return out[["剔除原因", *cols]].sort_values("金额_万元", ascending=False) def summarize_group(df: pd.DataFrame, keys: list[str]) -> pd.DataFrame: grouped = ( df.groupby(keys, dropna=False) .agg( 项目数=("项目名称", "count"), 披露金额项目数=("是否披露金额", "sum"), 金额合计_万元=("金额_万元", "sum"), 平均金额_万元=("金额_万元", "mean"), 中位金额_万元=("金额_万元", "median"), 最大金额_万元=("金额_万元", "max"), 招标单位数=("招标单位", lambda s: s.replace("", pd.NA).dropna().nunique()), 中标单位数=("中标单位", lambda s: s.replace("", pd.NA).dropna().nunique()), 来源链接数=("是否有链接", "sum"), 正文记录数=("是否有正文", "sum"), ) .reset_index() ) for col in ["金额合计_万元", "平均金额_万元", "中位金额_万元", "最大金额_万元"]: grouped[col] = grouped[col].fillna(0).round(4) return grouped def top_values(series: pd.Series, limit: int = 3) -> str: values = [clean(value) for value in series if clean(value)] if not values: return "" return "；".join(f"{name}({count})" for name, count in Counter(values).most_common(limit)) def top_projects_text(df: pd.DataFrame, limit: int = 3) -> str: names = [] for _, row in df.sort_values("金额_万元", ascending=False).head(limit).iterrows(): amount = row.get("金额_万元") amount_text = money_wan(float(amount)) if pd.notna(amount) else "未披露" names.append(f"{clean(row.get('项目名称'))}（{amount_text}）") return "；".join(names) def extract_product_tags(title: Any, notice_text: Any, demand_type: Any) -> list[dict[str, str]]: text = f"{clean(title)} {clean(notice_text)}" rows: list[dict[str, str]] = [] seen: set[tuple[str, str]] = set() for category, keywords in PRODUCT_KEYWORDS.items(): matched = [kw for kw in keywords if kw in text] if not matched: continue key = (category, "、".join(matched[:8])) if key in seen: continue seen.add(key) rows.append( { "产品/能力类别": category, "抽取关键词": "、".join(matched[:8]), "证据状态": "公告正文/标题抽取" if clean(notice_text) else "项目标题/需求类型线索", } ) if not rows and clean(demand_type): rows.append( { "产品/能力类别": clean(demand_type), "抽取关键词": clean(demand_type), "证据状态": "需求类型分类线索", } ) return rows def evidence_excerpt(text: str, keyword: str, radius: int = 70) -> str: index = text.find(keyword) if index < 0: return clean(text[: radius * 2]) start = max(index - radius, 0) end = min(index + len(keyword) + radius, len(text)) return clean(text[start:end]) def extract_qualification_tags(notice_text: Any) -> list[dict[str, str]]: text = clean(notice_text) rows: list[dict[str, str]] = [] seen: set[str] = set() for keyword in QUALIFICATION_KEYWORDS: if keyword in text and keyword not in seen: seen.add(keyword) rows.append( { "资质/证书线索": keyword, "证据状态": "公告正文抽取", "证据摘录": evidence_excerpt(text, keyword), "下一步": "回到公告原文或企业库核验证书编号、有效期和发证机构。", } ) if "资格能力条件" in text and not rows: rows.append( { "资质/证书线索": "资格能力条件", "证据状态": "公告提到但未披露具体证书", "证据摘录": evidence_excerpt(text, "资格能力条件"), "下一步": "需要下载招标文件或登录天眼查/企查查/爱企查补齐证书明细。", } ) return rows def top_supplier_names(df: pd.DataFrame, limit: int = 10) -> pd.DataFrame: use = df[df["供应商归一名称"].astype(str).str.strip() != ""].copy() grouped = ( use.groupby("供应商归一名称") .agg( 供应商=("中标单位", lambda s: clean(s.dropna().iloc[0]) if len(s.dropna()) else ""), 项目数=("项目名称", "count"), 披露金额项目数=("是否披露金额", "sum"), 金额合计_万元=("金额_万元", "sum"), 年份数=("年份", "nunique"), 覆盖省份数=("发布省份", lambda s: s.replace("", pd.NA).dropna().nunique()), 覆盖城市数=("发布市级", lambda s: s.replace("", pd.NA).dropna().nunique()), 招标单位数=("招标单位", lambda s: s.replace("", pd.NA).dropna().nunique()), 最早披露日期=("信息发布时间", lambda s: s.replace("", pd.NA).dropna().min()), 最近披露日期=("信息发布时间", lambda s: s.replace("", pd.NA).dropna().max()), ) .reset_index() .sort_values(["项目数", "金额合计_万元"], ascending=False) .head(limit) .reset_index(drop=True) ) grouped.insert(0, "供应商排名", range(1, len(grouped) + 1)) grouped["金额合计_万元"] = grouped["金额合计_万元"].round(4) return grouped def supplier_year_changes(df: pd.DataFrame, top_suppliers: pd.DataFrame) -> pd.DataFrame: rows = [] ranks = dict(zip(top_suppliers["供应商归一名称"], top_suppliers["供应商排名"])) display_names = dict(zip(top_suppliers["供应商归一名称"], top_suppliers["供应商"])) for supplier_key in top_suppliers["供应商归一名称"]: previous_count: int | None = None previous_amount: float | None = None for year in [2023, 2024, 2025, 2026]: part = df[(df["供应商归一名称"] == supplier_key) & (df["年份"] == year)].copy() amounts = pd.to_numeric(part["金额_万元"], errors="coerce").dropna() count = int(part.shape[0]) amount_sum = float(amounts.sum()) if not amounts.empty else 0.0 rows.append( { "供应商排名": ranks[supplier_key], "供应商": display_names[supplier_key], "年份": year, "项目数": count, "项目数较上年变化": "" if previous_count is None else count - previous_count, "项目数同比": "" if previous_count in (None, 0) else round((count - previous_count) / previous_count, 4), "披露金额项目数": int(part["是否披露金额"].sum()) if count else 0, "金额合计_万元": round(amount_sum, 4), "金额较上年变化_万元": "" if previous_amount is None else round(amount_sum - previous_amount, 4), "金额同比": "" if previous_amount in (None, 0) else round((amount_sum - previous_amount) / previous_amount, 4), "平均金额_万元": round(float(amounts.mean()), 4) if not amounts.empty else "", "最高金额_万元": round(float(amounts.max()), 4) if not amounts.empty else "", "覆盖省份数": int(part["发布省份"].replace("", pd.NA).dropna().nunique()) if count else 0, "覆盖城市数": int(part["发布市级"].replace("", pd.NA).dropna().nunique()) if count else 0, "招标单位数": int(part["招标单位"].replace("", pd.NA).dropna().nunique()) if count else 0, "主要省份Top3": top_values(part["发布省份"]) if count else "", "主要客户类型Top3": top_values(part["客户类型"]) if count else "", "主要需求类型Top3": top_values(part["需求类型"]) if count else "", "代表项目Top3": top_projects_text(part) if count else "", "口径说明": "2026 为 1-6 月 YTD" if year == 2026 else "全年", } ) previous_count = count previous_amount = amount_sum return pd.DataFrame(rows) def supplier_project_details(df: pd.DataFrame, top_suppliers: pd.DataFrame) -> pd.DataFrame: ranks = dict(zip(top_suppliers["供应商归一名称"], top_suppliers["供应商排名"])) use = df[df["供应商归一名称"].isin(set(top_suppliers["供应商归一名称"]))].copy() use["供应商排名"] = use["供应商归一名称"].map(ranks) use["供应商"] = use["中标单位"] use = use.sort_values(["供应商归一名称", "年份", "金额_万元"], ascending=[True, True, False]) cols = [ "供应商排名", "供应商", "年份", "月份", "项目名称", "发布省份", "发布市级", "中标阶段", "金额_万元", "招标单位", "中标单位", "需求类型", "客户类型", "官网查看地址", "正文摘要", ] return use[cols] def supplier_product_changes(df: pd.DataFrame, top_suppliers: pd.DataFrame) -> pd.DataFrame: rows = [] ranks = dict(zip(top_suppliers["供应商归一名称"], top_suppliers["供应商排名"])) for _, item in df[df["供应商归一名称"].isin(set(top_suppliers["供应商归一名称"]))].iterrows(): supplier_key = item["供应商归一名称"] for product in extract_product_tags(item.get("项目名称"), item.get("公告内容片段"), item.get("需求类型")): rows.append( { "供应商排名": ranks.get(supplier_key, ""), "供应商": item.get("中标单位", ""), "年份": item.get("年份", ""), "产品/能力类别": product["产品/能力类别"], "抽取关键词": product["抽取关键词"], "证据状态": product["证据状态"], "项目名称": item.get("项目名称", ""), "金额_万元": item.get("金额_万元", ""), "发布省份": item.get("发布省份", ""), "招标单位": item.get("招标单位", ""), "来源链接": item.get("官网查看地址", ""), } ) out = pd.DataFrame(rows) if out.empty: return out return out.sort_values(["供应商排名", "年份", "产品/能力类别", "金额_万元"], ascending=[True, True, True, False]) def supplier_product_summary(product_details: pd.DataFrame) -> pd.DataFrame: if product_details.empty: return product_details rows = [] for keys, part in product_details.groupby(["供应商排名", "供应商", "年份", "产品/能力类别"], dropna=False): rank, supplier, year, category = keys amounts = pd.to_numeric(part["金额_万元"], errors="coerce").dropna() rows.append( { "供应商排名": rank, "供应商": supplier, "年份": year, "产品/能力类别": category, "关联项目数": int(part["项目名称"].nunique()), "金额合计_万元": round(float(amounts.sum()), 4) if not amounts.empty else 0, "关键词Top": top_values(part["抽取关键词"], 5), "证据状态Top": top_values(part["证据状态"], 3), "代表项目": top_projects_text(part.rename(columns={"项目名称": "项目名称"}), 1), } ) return pd.DataFrame(rows).sort_values(["供应商排名", "年份", "金额合计_万元"], ascending=[True, True, False]) def supplier_qualification_changes(df: pd.DataFrame, top_suppliers: pd.DataFrame) -> pd.DataFrame: rows = [] ranks = dict(zip(top_suppliers["供应商归一名称"], top_suppliers["供应商排名"])) display_names = dict(zip(top_suppliers["供应商归一名称"], top_suppliers["供应商"])) top_keys = set(top_suppliers["供应商归一名称"]) for _, item in df[df["供应商归一名称"].isin(top_keys)].iterrows(): extracted = extract_qualification_tags(item.get("公告内容片段")) for qual in extracted: rows.append( { "供应商排名": ranks.get(item["供应商归一名称"], ""), "供应商": item.get("中标单位", ""), "年份": item.get("年份", ""), "资质/证书线索": qual["资质/证书线索"], "证据状态": qual["证据状态"], "项目名称": item.get("项目名称", ""), "来源链接": item.get("官网查看地址", ""), "证据摘录": qual["证据摘录"], "下一步": qual["下一步"], } ) existing = {(row["供应商"], row["年份"]) for row in rows} for supplier_key in top_suppliers["供应商归一名称"]: for year in [2023, 2024, 2025, 2026]: part = df[(df["供应商归一名称"] == supplier_key) & (df["年份"] == year)] if part.empty or (display_names[supplier_key], year) in existing: continue rows.append( { "供应商排名": ranks[supplier_key], "供应商": display_names[supplier_key], "年份": year, "资质/证书线索": "公告未披露具体资质证书", "证据状态": "待企业库补抓", "项目名称": top_projects_text(part, 1), "来源链接": clean(part.iloc[0].get("官网查看地址", "")), "证据摘录": "当前源表未给出证书编号、有效期、发证机构等结构化信息。", "下一步": "登录天眼查/企查查/爱企查，或下载正式招标文件/候选人响应表补齐。", } ) out = pd.DataFrame(rows) if out.empty: return out return out.sort_values(["供应商排名", "年份", "证据状态"]) def supplier_qualification_summary(qualification_details: pd.DataFrame) -> pd.DataFrame: if qualification_details.empty: return qualification_details rows = [] for keys, part in qualification_details.groupby(["供应商排名", "供应商", "年份", "证据状态"], dropna=False): rank, supplier, year, status = keys rows.append( { "供应商排名": rank, "供应商": supplier, "年份": year, "证据状态": status, "线索/缺口数": int(part.shape[0]), "资质/证书线索Top": top_values(part["资质/证书线索"], 5), "代表项目": clean(part.iloc[0].get("项目名称", "")), "下一步": clean(part.iloc[0].get("下一步", "")), } ) return pd.DataFrame(rows).sort_values(["供应商排名", "年份", "线索/缺口数"], ascending=[True, True, False]) def add_mom_yoy(monthly: pd.DataFrame) -> pd.DataFrame: monthly = monthly.sort_values("月份").reset_index(drop=True).copy() monthly["上月项目数"] = monthly["项目数"].shift(1) monthly["上月金额_万元"] = monthly["金额合计_万元"].shift(1) monthly["环比项目数变化"] = monthly["项目数"] - monthly["上月项目数"] monthly["环比金额变化_万元"] = monthly["金额合计_万元"] - monthly["上月金额_万元"] monthly["环比项目数变化率"] = [ safe_pct(now, prev) for now, prev in zip(monthly["项目数"], monthly["上月项目数"]) ] monthly["环比金额变化率"] = [ safe_pct(now, prev) for now, prev in zip(monthly["金额合计_万元"], monthly["上月金额_万元"]) ] lookup = monthly.set_index("月份")[["项目数", "金额合计_万元"]].to_dict("index") last_year_counts = [] last_year_amounts = [] for month in monthly["月份"]: dt = pd.to_datetime(f"{month}-01", errors="coerce") prev_key = f"{dt.year - 1}-{dt.month:02d}" if pd.notna(dt) else "" prev = lookup.get(prev_key, {}) last_year_counts.append(prev.get("项目数")) last_year_amounts.append(prev.get("金额合计_万元")) monthly["去年同月项目数"] = last_year_counts monthly["去年同月金额_万元"] = last_year_amounts monthly["同比项目数变化"] = monthly["项目数"] - monthly["去年同月项目数"] monthly["同比金额变化_万元"] = monthly["金额合计_万元"] - monthly["去年同月金额_万元"] monthly["同比项目数变化率"] = [ safe_pct(now, prev) for now, prev in zip(monthly["项目数"], monthly["去年同月项目数"]) ] monthly["同比金额变化率"] = [ safe_pct(now, prev) for now, prev in zip(monthly["金额合计_万元"], monthly["去年同月金额_万元"]) ] return monthly def compare_h1(df: pd.DataFrame) -> pd.DataFrame: h1 = df[df["月份序号"].isin([1, 2, 3, 4, 5, 6])].copy() annual = summarize_group(h1, ["年份"]) annual = annual.sort_values("年份").reset_index(drop=True) annual["去年上半年项目数"] = annual["项目数"].shift(1) annual["去年上半年金额_万元"] = annual["金额合计_万元"].shift(1) annual["上半年项目数同比"] = [safe_pct(n, p) for n, p in zip(annual["项目数"], annual["去年上半年项目数"])] annual["上半年金额同比"] = [safe_pct(n, p) for n, p in zip(annual["金额合计_万元"], annual["去年上半年金额_万元"])] return annual def compare_dimension_h1(df: pd.DataFrame, dimension: str, current_year: int = 2026, base_year: int = 2025) -> pd.DataFrame: h1 = df[df["月份序号"].isin([1, 2, 3, 4, 5, 6]) & df["年份"].isin([base_year, current_year])].copy() summary = summarize_group(h1, ["年份", dimension]) pivot_count = summary.pivot(index=dimension, columns="年份", values="项目数").fillna(0) pivot_amount = summary.pivot(index=dimension, columns="年份", values="金额合计_万元").fillna(0) rows = [] labels = sorted(set(pivot_count.index) | set(pivot_amount.index)) for label in labels: count_base = float(pivot_count.get(base_year, pd.Series()).get(label, 0)) count_now = float(pivot_count.get(current_year, pd.Series()).get(label, 0)) amount_base = float(pivot_amount.get(base_year, pd.Series()).get(label, 0)) amount_now = float(pivot_amount.get(current_year, pd.Series()).get(label, 0)) rows.append( { dimension: label, "2025上半年项目数": count_base, "2026上半年项目数": count_now, "项目数变化": count_now - count_base, "项目数同比": safe_pct(count_now, count_base), "2025上半年金额_万元": round(amount_base, 4), "2026上半年金额_万元": round(amount_now, 4), "金额变化_万元": round(amount_now - amount_base, 4), "金额同比": safe_pct(amount_now, amount_base), } ) out = pd.DataFrame(rows) if out.empty: return out return out.sort_values(["金额变化_万元", "项目数变化"], ascending=False) def top_projects(df: pd.DataFrame, year: int | None = None, limit: int = 30) -> pd.DataFrame: use = df.copy() if year is not None: use = use[use["年份"] == year] cols = [ "年份", "月份", "项目名称", "发布省份", "发布市级", "中标阶段", "金额_万元", "招标单位", "中标单位", "需求类型", "客户类型", "官网查看地址", ] return use.sort_values("金额_万元", ascending=False)[cols].head(limit) def source_profile(df: pd.DataFrame) -> pd.DataFrame: rows = [] for source in SOURCES: part = df[df["来源说明"] == source["label"]] rows.append( { "来源说明": source["label"], "来源文件": str(source["path"]), "记录数": len(part), "最早日期": part["信息发布时间"].replace("", pd.NA).dropna().min() if not part.empty else "", "最晚日期": part["信息发布时间"].replace("", pd.NA).dropna().max() if not part.empty else "", "披露金额项目数": int(part["是否披露金额"].sum()) if not part.empty else 0, "来源链接数": int(part["是否有链接"].sum()) if not part.empty else 0, "正文记录数": int(part["是否有正文"].sum()) if not part.empty else 0, } ) return pd.DataFrame(rows) def build_tables(df: pd.DataFrame) -> dict[str, pd.DataFrame]: monthly = add_mom_yoy(summarize_group(df, ["月份"])) annual = summarize_group(df, ["年份"]) annual["口径说明"] = annual["年份"].apply(lambda y: "YTD 1-6月" if int(y) == 2026 else "全年") h1 = compare_h1(df) top_suppliers = top_supplier_names(df, 10) supplier_products = supplier_product_changes(df, top_suppliers) supplier_quals = supplier_qualification_changes(df, top_suppliers) tables = { "数据源体检": source_profile(df), "年度汇总": annual.sort_values("年份"), "上半年同比": h1, "月度环比同比": monthly, "季度汇总": summarize_group(df, ["季度"]), "省份年度": summarize_group(df, ["年份", "发布省份"]), "城市年度": summarize_group(df, ["年份", "发布省份", "发布市级"]), "需求类型年度": summarize_group(df, ["年份", "需求类型"]), "客户类型年度": summarize_group(df, ["年份", "客户类型"]), "供应商年度Top": summarize_group(df, ["年份", "中标单位"]).sort_values(["年份", "金额合计_万元"], ascending=[True, False]).head(300), "2026上半年_vs_2025上半年_省份": compare_dimension_h1(df, "发布省份"), "2026上半年_vs_2025上半年_需求": compare_dimension_h1(df, "需求类型"), "2026上半年_vs_2025上半年_客户": compare_dimension_h1(df, "客户类型"), "Top10供应商总览": top_suppliers, "Top10供应商年度变化": supplier_year_changes(df, top_suppliers), "Top10供应商产品年度汇总": supplier_product_summary(supplier_products), "Top10供应商产品变化": supplier_products, "Top10供应商资质年度汇总": supplier_qualification_summary(supplier_quals), "Top10供应商资质证书变化": supplier_quals, "Top10供应商项目明细": supplier_project_details(df, top_suppliers), "2026上半年高金额项目": top_projects(df, 2026, 40), "全量清洗明细": df[ [ "来源年份", "年份", "月份", "项目名称", "发布省份", "发布市级", "发布区级", "中标阶段", "金额_万元", "招标单位", "中标单位", "需求类型", "客户类型", "官网查看地址", "正文摘要", "公告内容长度", "公告内容片段", "包含附件", "来源说明", "源表行号", ] ].copy(), } return tables def attach_outlier_comparison_tables( filtered_tables: dict[str, pd.DataFrame], raw_tables: dict[str, pd.DataFrame], excluded: pd.DataFrame, ) -> dict[str, pd.DataFrame]: tables = dict(filtered_tables) tables["异常剔除项目"] = excluded tables["含异常_年度汇总"] = raw_tables["年度汇总"] tables["含异常_上半年同比"] = raw_tables["上半年同比"] tables["含异常_月度环比同比"] = raw_tables["月度环比同比"] tables["含异常_2026高金额项目"] = raw_tables["2026上半年高金额项目"] return tables def format_for_excel(df: pd.DataFrame) -> pd.DataFrame: out = df.copy() for col in out.columns: if "同比" in col or "环比" in col or "变化率" in col: out[col] = out[col].apply( lambda value: "" if value in ("", None) or pd.isna(value) else round(float(value), 4) ) return out def autosize_sheet(ws) -> None: ws.freeze_panes = "A2" ws.sheet_view.showGridLines = False header_fill = PatternFill("solid", fgColor=BLUE) header_font = Font(name=FONT, size=12, bold=True, color="FFFFFF") body_font = Font(name=FONT, size=12, color="1F1F1F") for row in ws.iter_rows(): for cell in row: cell.font = header_font if cell.row == 1 else body_font cell.alignment = Alignment(wrap_text=True, vertical="top") cell.border = BORDER if cell.row == 1: cell.fill = header_fill for col_idx, column_cells in enumerate(ws.columns, 1): max_len = 10 for cell in list(column_cells)[:80]: max_len = max(max_len, len(str(cell.value or "")) // 2) ws.column_dimensions[get_column_letter(col_idx)].width = min(max(max_len + 4, 12), 46) ws.auto_filter.ref = ws.dimensions def write_excel(tables: dict[str, pd.DataFrame]) -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) with pd.ExcelWriter(OUT_XLSX, engine="openpyxl") as writer: for name, table in tables.items(): safe_name = name[:31] format_for_excel(table).to_excel(writer, index=False, sheet_name=safe_name) wb = load_workbook(OUT_XLSX) for ws in wb.worksheets: autosize_sheet(ws) wb.save(OUT_XLSX) def html_escape(value: Any) -> str: return html.escape("" if value is None else str(value)) def html_cell_text(value: Any) -> str: if value is None: return "" if isinstance(value, float): if pd.isna(value): return "" if abs(value - round(value)) < 1e-9 and abs(value) < 10000: return str(int(round(value))) return f"{value:,.4f}" if isinstance(value, int): return str(value) return html_escape(value) def supplier_anchor(rank: Any) -> str: try: return f"supplier-{int(float(rank))}" except (TypeError, ValueError): return "" def signed_pct(value: float | None) -> str: if value is None or pd.isna(value): return "不可比" sign = "+" if value > 0 else "" return f"{sign}{value * 100:.1f}%" def signed_number(value: float | int | None, digits: int = 1) -> str: if value is None or pd.isna(value): return "不可比" sign = "+" if value > 0 else "" return f"{sign}{value:.{digits}f}" def table_html( df: pd.DataFrame, columns: list[str] | None = None, limit: int = 20, link_col: str | None = None, supplier_link_col: str | None = None, ) -> str: use = df.copy() if columns: use = use[columns] use = use.head(limit) header = "".join(f"{html_escape(col)}" for col in use.columns) rows = [] for _, row in use.iterrows(): cells = [] for col in use.columns: value = row[col] if link_col and col == link_col and isinstance(value, str) and value.startswith("http"): cells.append(f'打开链接') elif supplier_link_col and col == supplier_link_col and "供应商排名" in use.columns: anchor = supplier_anchor(row.get("供应商排名")) if anchor: cells.append(f'{html_cell_text(value)}') else: cells.append(f"{html_cell_text(value)}") else: cells.append(f"{html_cell_text(value)}") rows.append("" + "".join(cells) + "") return f"{header}{''.join(rows)}
" def bar_rows(title: str, rows: list[tuple[str, float]], suffix: str = "") -> str: if not rows: return "" max_value = max(abs(v) for _, v in rows) or 1 body = [] for name, value in rows: width = max(3, abs(value) / max_value * 100) body.append( f"""
{html_escape(name)}
{value:,.1f}{suffix}
""" ) return f'

{html_escape(title)}

{"".join(body)}
' def supplier_detail_sections(tables: dict[str, pd.DataFrame]) -> str: top_suppliers = tables["Top10供应商总览"] supplier_changes = tables["Top10供应商年度变化"] supplier_product_summary_table = tables["Top10供应商产品年度汇总"] supplier_qual_summary_table = tables["Top10供应商资质年度汇总"] supplier_projects = tables["Top10供应商项目明细"] sections = [] for _, supplier in top_suppliers.iterrows(): rank = int(supplier["供应商排名"]) name = clean(supplier["供应商"]) anchor = supplier_anchor(rank) yearly = supplier_changes[supplier_changes["供应商排名"] == rank].copy() active_yearly = yearly[pd.to_numeric(yearly["项目数"], errors="coerce").fillna(0) > 0].copy() products = supplier_product_summary_table[supplier_product_summary_table["供应商排名"] == rank].copy() quals = supplier_qual_summary_table[supplier_qual_summary_table["供应商排名"] == rank].copy() projects = supplier_projects[supplier_projects["供应商排名"] == rank].copy() active_year_text = "、".join(str(int(year)) for year in active_yearly["年份"].tolist()) if not active_yearly.empty else "未见有效年份" main_client = top_values(projects["客户类型"], 4) if not projects.empty else "" main_demand = top_values(projects["需求类型"], 4) if not projects.empty else "" main_region = top_values(projects["发布市级"], 5) or top_values(projects["发布省份"], 5) high_project = top_projects_text(projects, 3) if not projects.empty else "" sections.append( f"""

{rank}. {html_escape(name)}

返回 Top10
{html_cell_text(supplier.get("项目数"))}合计项目数
{html_cell_text(supplier.get("金额合计_万元"))}金额合计_万元
{html_cell_text(supplier.get("招标单位数"))}招标单位数
{html_cell_text(supplier.get("覆盖城市数"))}覆盖城市数
高频中标线索：活跃年份为 {html_escape(active_year_text)}；主要地区为 {html_escape(main_region or "未识别")}；主要客户类型为 {html_escape(main_client or "未识别")}；主要需求类型为 {html_escape(main_demand or "未识别")}。代表项目：{html_escape(high_project or "暂无")}。

年度统计

{table_html(yearly, ["年份", "项目数", "项目数较上年变化", "项目数同比", "披露金额项目数", "金额合计_万元", "平均金额_万元", "最高金额_万元", "覆盖省份数", "覆盖城市数", "招标单位数", "主要省份Top3", "主要客户类型Top3", "主要需求类型Top3", "代表项目Top3", "口径说明"], 10)}

项目明细

{table_html(projects, ["年份", "月份", "项目名称", "发布省份", "发布市级", "中标阶段", "金额_万元", "招标单位", "需求类型", "客户类型", "官网查看地址"], 500, "官网查看地址")}

产品/能力年度汇总

{table_html(products, ["年份", "产品/能力类别", "关联项目数", "金额合计_万元", "关键词Top", "证据状态Top", "代表项目"], 80)}

资质证书线索

{table_html(quals, ["年份", "证据状态", "线索/缺口数", "资质/证书线索Top", "代表项目", "下一步"], 80)}
""" ) return "".join(sections) def build_insights(tables: dict[str, pd.DataFrame]) -> dict[str, Any]: h1 = tables["上半年同比"].set_index("年份") monthly = tables["月度环比同比"] annual = tables["年度汇总"].set_index("年份") latest = monthly[monthly["月份"] == "2026-06"].iloc[0] if not monthly[monthly["月份"] == "2026-06"].empty else monthly.iloc[-1] h1_2026 = h1.loc[2026] if 2026 in h1.index else None h1_2025 = h1.loc[2025] if 2025 in h1.index else None return { "totalRecords": int(tables["全量清洗明细"].shape[0]), "annualYears": sorted(int(x) for x in annual.index), "h1_2026_count": int(h1_2026["项目数"]) if h1_2026 is not None else 0, "h1_2026_amount": float(h1_2026["金额合计_万元"]) if h1_2026 is not None else 0, "h1_count_yoy": float(h1_2026["上半年项目数同比"]) if h1_2026 is not None and pd.notna(h1_2026["上半年项目数同比"]) else None, "h1_amount_yoy": float(h1_2026["上半年金额同比"]) if h1_2026 is not None and pd.notna(h1_2026["上半年金额同比"]) else None, "h1_2025_count": int(h1_2025["项目数"]) if h1_2025 is not None else 0, "h1_2025_amount": float(h1_2025["金额合计_万元"]) if h1_2025 is not None else 0, "latestMonth": str(latest["月份"]), "latestProjectCount": int(latest["项目数"]), "latestAmount": float(latest["金额合计_万元"]), "latestMomCount": float(latest["环比项目数变化率"]) if pd.notna(latest["环比项目数变化率"]) else None, "latestMomAmount": float(latest["环比金额变化率"]) if pd.notna(latest["环比金额变化率"]) else None, "latestYoyCount": float(latest["同比项目数变化率"]) if pd.notna(latest["同比项目数变化率"]) else None, "latestYoyAmount": float(latest["同比金额变化率"]) if pd.notna(latest["同比金额变化率"]) else None, } def build_html(tables: dict[str, pd.DataFrame]) -> str: insights = build_insights(tables) annual_rows = [ ( str(int(row["年份"])) + (" YTD" if int(row["年份"]) == 2026 else ""), float(row["金额合计_万元"]), ) for _, row in tables["年度汇总"].iterrows() ] h1_rows = [(str(int(row["年份"])), float(row["金额合计_万元"])) for _, row in tables["上半年同比"].iterrows()] province_delta = tables["2026上半年_vs_2025上半年_省份"].copy().head(10) demand_delta = tables["2026上半年_vs_2025上半年_需求"].copy().head(8) top_suppliers = tables["Top10供应商总览"] supplier_changes = tables["Top10供应商年度变化"] supplier_product_summary_table = tables["Top10供应商产品年度汇总"] supplier_products = tables["Top10供应商产品变化"] supplier_qual_summary_table = tables["Top10供应商资质年度汇总"] supplier_quals = tables["Top10供应商资质证书变化"] top_projects_2026 = tables["2026上半年高金额项目"] month_2026 = tables["月度环比同比"][tables["月度环比同比"]["月份"].astype(str).str.startswith("2026-")] excluded_projects = tables.get("异常剔除项目", pd.DataFrame()) raw_annual = tables.get("含异常_年度汇总", pd.DataFrame()) raw_h1 = tables.get("含异常_上半年同比", pd.DataFrame()) raw_month_2026 = tables.get("含异常_月度环比同比", pd.DataFrame()) if not raw_month_2026.empty: raw_month_2026 = raw_month_2026[raw_month_2026["月份"].astype(str).str.startswith("2026-")] raw_top_projects = tables.get("含异常_2026高金额项目", pd.DataFrame()) supplier_details = supplier_detail_sections(tables) supplier_leader = top_suppliers.iloc[0].to_dict() if not top_suppliers.empty else {} excluded_count = int(excluded_projects.shape[0]) if not excluded_projects.empty else 0 excluded_amount = float(pd.to_numeric(excluded_projects.get("金额_万元", pd.Series(dtype=float)), errors="coerce").fillna(0).sum()) if not excluded_projects.empty else 0.0 raw_total_records = insights["totalRecords"] + excluded_count generated = datetime.now().strftime("%Y-%m-%d %H:%M:%S") return f""" 2023-2026智慧食堂招投标同比环比分析

2023-2026 智慧食堂招投标同比环比分析

合并新版 2023-2025 乙方宝原始 Excel 与项目内 2026 年 1-6 月数据；默认剔除 2 个 13 亿级异常项目后，分析月度环比、同月同比、上半年同比、地区、客户类型、需求类型和供应商变化。

{insights["totalRecords"]:,}剔除后合并记录数
{insights["h1_2026_count"]:,}剔除后 2026 上半年项目数
{money_wan(insights["h1_2026_amount"])}剔除后 2026 上半年金额
{signed_pct(insights["h1_amount_yoy"])}剔除后 2026 上半年金额同比

核心结论

生成时间 {generated}
口径说明：这里原先写的 H1 指 Half 1，即上半年（1-6月）。2023、2024、2025 使用新版 2023-2025 合并原始 Excel；2026 只有 1-6 月。默认分析已剔除 2 个 13 亿级异常项目，避免金额同比和环比被极端值拉偏；含异常口径可在下方按钮切换查看。
  • 剔除异常项目后，2026 上半年（1-6月）项目数为 {insights["h1_2026_count"]:,}，对比 2025 上半年的 {insights["h1_2025_count"]:,}，项目数同比 {signed_pct(insights["h1_count_yoy"])}。
  • 剔除异常项目后，2026 上半年（1-6月）金额为 {money_wan(insights["h1_2026_amount"])}，对比 2025 上半年的 {money_wan(insights["h1_2025_amount"])}，金额同比 {signed_pct(insights["h1_amount_yoy"])}。
  • {html_escape(insights["latestMonth"])} 项目数 {insights["latestProjectCount"]:,}，金额 {money_wan(insights["latestAmount"])}；项目数环比 {signed_pct(insights["latestMomCount"])}，金额环比 {signed_pct(insights["latestMomAmount"])}。
  • {html_escape(insights["latestMonth"])} 对去年同月，项目数同比 {signed_pct(insights["latestYoyCount"])}，金额同比 {signed_pct(insights["latestYoyAmount"])}。
  • 按 2023-2026 合计项目数排序，Top 10 供应商已单独列出年度项目数、金额、地区、客户类型、产品/能力线索和资质证书缺口；当前项目数最高为 {html_escape(supplier_leader.get("供应商", ""))}，合计 {html_escape(supplier_leader.get("项目数", ""))} 个项目。

异常项目剔除与口径切换

默认剔除
已将 {excluded_count} 个 13 亿级项目从默认分析口径中剔除，剔除金额合计 {money_wan(excluded_amount)}。这些项目仍保留在 Excel 的“异常剔除项目”和“含异常”对照 sheet 中，页面也可以用按钮查看含异常口径。
{insights["totalRecords"]:,}剔除后记录数
{raw_total_records:,}含异常记录数
{money_wan(excluded_amount)}本次剔除金额

被剔除项目

{table_html(excluded_projects, ["剔除原因", "月份", "项目名称", "发布省份", "发布市级", "金额_万元", "招标单位", "中标单位", "官网查看地址"], 10, "官网查看地址")}

剔除后口径对照

{table_html(tables["年度汇总"], ["年份", "项目数", "金额合计_万元", "平均金额_万元", "最大金额_万元", "口径说明"], 10)}
{table_html(tables["上半年同比"], ["年份", "项目数", "金额合计_万元", "去年上半年项目数", "上半年项目数同比", "去年上半年金额_万元", "上半年金额同比"], 10)}
{table_html(month_2026, ["月份", "项目数", "金额合计_万元", "环比金额变化率", "去年同月金额_万元", "同比金额变化率"], 12)}

年度与上半年金额走势

单位：万元
{bar_rows("年度金额合计，2026 为 YTD", annual_rows, " 万")} {bar_rows("上半年（1-6月）金额合计，可比口径", h1_rows, " 万")}

2026 月度环比与同比

同月同比对 2025
{table_html(month_2026, ["月份", "项目数", "金额合计_万元", "环比项目数变化", "环比项目数变化率", "环比金额变化_万元", "环比金额变化率", "去年同月项目数", "同比项目数变化率", "去年同月金额_万元", "同比金额变化率"], 12)}

上半年（1-6月）同比表

2023-2026
{table_html(tables["上半年同比"], ["年份", "项目数", "金额合计_万元", "去年上半年项目数", "上半年项目数同比", "去年上半年金额_万元", "上半年金额同比"], 10)}

数据源体检

字段与覆盖
{table_html(tables["数据源体检"], ["来源说明", "记录数", "最早日期", "最晚日期", "披露金额项目数", "来源链接数", "正文记录数"], 10)}

2026 上半年 vs 2025 上半年省份变化

按金额变化排序
{table_html(province_delta, ["发布省份", "2025上半年项目数", "2026上半年项目数", "项目数变化", "项目数同比", "2025上半年金额_万元", "2026上半年金额_万元", "金额变化_万元", "金额同比"], 12)}

2026 上半年 vs 2025 上半年需求变化

按金额变化排序
{table_html(demand_delta, ["需求类型", "2025上半年项目数", "2026上半年项目数", "项目数变化", "项目数同比", "2025上半年金额_万元", "2026上半年金额_万元", "金额变化_万元", "金额同比"], 12)}

Top 10 供应商总览

按 2023-2026 合计项目数排序
{table_html(top_suppliers, ["供应商排名", "供应商", "项目数", "披露金额项目数", "金额合计_万元", "年份数", "覆盖省份数", "覆盖城市数", "招标单位数", "最早披露日期", "最近披露日期"], 10, supplier_link_col="供应商")}

Top 10 供应商年度中标变化

2026 为 1-6 月 YTD
{table_html(supplier_changes, ["供应商排名", "供应商", "年份", "项目数", "项目数较上年变化", "项目数同比", "金额合计_万元", "金额较上年变化_万元", "金额同比", "覆盖省份数", "主要省份Top3", "主要客户类型Top3", "主要需求类型Top3", "代表项目Top3", "口径说明"], 60, supplier_link_col="供应商")}

Top 10 供应商产品/能力年度汇总

公告正文优先，标题线索兜底
{table_html(supplier_product_summary_table, ["供应商排名", "供应商", "年份", "产品/能力类别", "关联项目数", "金额合计_万元", "关键词Top", "证据状态Top", "代表项目"], 100)}

Top 10 供应商资质证书年度汇总

不编证书，缺口显式标注
{table_html(supplier_qual_summary_table, ["供应商排名", "供应商", "年份", "证据状态", "线索/缺口数", "资质/证书线索Top", "代表项目", "下一步"], 100)}

产品/能力明细抽样

完整明细见 Excel
{table_html(supplier_products, ["供应商排名", "供应商", "年份", "产品/能力类别", "抽取关键词", "证据状态", "项目名称", "金额_万元", "来源链接"], 40, "来源链接")}

资质证书明细抽样

完整明细见 Excel
{table_html(supplier_quals, ["供应商排名", "供应商", "年份", "资质/证书线索", "证据状态", "项目名称", "证据摘录", "下一步", "来源链接"], 40, "来源链接")}

Top 10 供应商公司详情

点击公司名可跳到这里
每个公司详情包含年度统计、全部项目明细、产品/能力年度汇总和资质证书线索。比如点击“邵阳市大祥区佳友信息技术服务中心”，可以看到它集中在 2024 年、湖南地区、学校客户、软件运营服务类项目中标。
{supplier_details}

2026 上半年高金额项目

可打开来源链接
{table_html(top_projects_2026, ["月份", "项目名称", "发布省份", "发布市级", "金额_万元", "招标单位", "中标单位", "需求类型", "客户类型", "官网查看地址"], 40, "官网查看地址")}
""" def write_html(tables: dict[str, pd.DataFrame]) -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) OUT_HTML.write_text(build_html(tables), encoding="utf-8") def write_run_note(summary: dict[str, Any]) -> None: note = ROOT / "docs" / "runs" / "20260617-multi-year-tender-trend-analysis.html" note.write_text( f""" 20260617 2023-2026招投标同比环比分析记录

2023-2026 招投标同比环比分析记录

用户目标

把新版 2023-2025 年乙方宝合并原始 Excel 与项目内 2026 年数据合并，从环比和同比角度分析智慧食堂招投标情况变化。

数据源

主口径选择 docs/deliverables/20260617-original-yearly-excel-reports/乙方宝智慧食堂2023年-2025年招投标数据_新版原始合并.xls 与 data/raw/2026年1月-6月乙方宝中标通知20260612.xls。该组文件字段与原分析逻辑兼容，均为乙方宝项目级导出。

输出文件

HTML：docs/deliverables/20260617-multi-year-trend-analysis/2023-2026智慧食堂招投标同比环比分析.html

Excel：docs/deliverables/20260617-multi-year-trend-analysis/2023-2026智慧食堂招投标同比环比分析.xlsx

脚本：scripts/build_multi_year_trend_analysis.py

验证结果

原始合并记录数：{summary['rawRecords']}；剔除异常项目数：{summary['excludedRecords']}；默认分析记录数：{summary['analysisRecords']}。输出工作簿 sheet 数：{summary['sheetCount']}。已使用 Python 解析 HTML 和 Excel 工作簿，并验证页面包含异常项目剔除按钮、剔除项目明细、含异常口径对照和 Top10 供应商页内跳转。

口径说明

2023、2024、2025 使用新版 2023-2025 合并原始表；2026 只有 1-6 月。默认分析口径剔除两条 13 亿级异常项目，报告中保留按钮和 Excel 对照 sheet 用于切换查看含异常口径。此前页面中的 H1 即 Half 1，上半年。

""", encoding="utf-8", ) def main() -> None: df = load_all_records() excluded = outlier_projects(df) analysis_df = df[~outlier_mask(df)].copy() raw_tables = build_tables(df) filtered_tables = build_tables(analysis_df) tables = attach_outlier_comparison_tables(filtered_tables, raw_tables, excluded) write_excel(tables) write_html(tables) summary = { "rawRecords": int(df.shape[0]), "excludedRecords": int(excluded.shape[0]), "analysisRecords": int(analysis_df.shape[0]), "sheetCount": len(tables), "html": str(OUT_HTML.relative_to(ROOT)), "xlsx": str(OUT_XLSX.relative_to(ROOT)), } write_run_note(summary) print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()