#!/usr/bin/env python3 """Generate a full bid-award analysis report from CSV/XLS/XLSX input. The script is intentionally dependency-light. It uses LibreOffice/soffice for Excel conversion and the Python standard library for profiling, scoring, and report generation. """ from __future__ import annotations import argparse import csv import json import math import shutil import statistics import subprocess import tempfile from collections import Counter, defaultdict from datetime import datetime from pathlib import Path from typing import Any RAW_COLUMNS = [ "关键词", "项目名称", "信息发布时间", "项目编号", "发布省份", "发布市级", "发布区级", "中标阶段", "中标金额（元）", "招标单位", "招标单位联系人", "招标单位联系人电话", "中标单位", "中标单位联系人", "中标单位联系人电话", "合同开始时间", "合同结束时间", "合同工期", "官网查看地址", ] SENSITIVE_COLUMNS = { "招标单位联系人", "招标单位联系人电话", "中标单位联系人", "中标单位联系人电话", } MISSING_MARKERS = {"", "--", "暂未公布", "nan", "None", "null", "NULL"} def clean_text(value: Any) -> str: if value is None: return "" text = str(value).strip() text = text.replace("\u3000", " ") text = " ".join(text.split()) return "" if text in MISSING_MARKERS else text def display_text(value: Any) -> str: text = clean_text(value) return text if text else "未填" def parse_amount(value: Any) -> float | None: text = clean_text(value).replace(",", "") if not text: return None for token in ["元", "￥", "¥"]: text = text.replace(token, "") try: return float(text) except ValueError: return None def parse_date(value: Any) -> datetime | None: text = clean_text(value) if not text: return None for fmt in ("%Y/%m/%d", "%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d %H:%M:%S"): try: return datetime.strptime(text, fmt) except ValueError: pass return None def parse_days(value: Any) -> int | None: text = clean_text(value) if not text: return None text = text.replace("天", "") try: return int(float(text)) except ValueError: return None def money(value: float | int | None) -> str: if value is None: return "未披露" if abs(value) >= 100000000: return f"{value / 100000000:.2f}亿元" if abs(value) >= 10000: return f"{value / 10000:.2f}万元" return f"{value:,.0f}元" def pct(value: float | int | None) -> str: if value is None: return "0.0%" return f"{value * 100:.1f}%" def quantile(values: list[float], q: float) -> float | None: if not values: return None ordered = sorted(values) if len(ordered) == 1: return ordered[0] pos = (len(ordered) - 1) * q low = math.floor(pos) high = math.ceil(pos) if low == high: return ordered[int(pos)] return ordered[low] + (ordered[high] - ordered[low]) * (pos - low) def convert_excel_to_csv(path: Path) -> Path: soffice = shutil.which("soffice") or shutil.which("libreoffice") if not soffice: raise RuntimeError("Need soffice/libreoffice to convert Excel input.") tmpdir = Path(tempfile.mkdtemp(prefix="bid-analysis-")) before = set(tmpdir.rglob("*")) subprocess.run( [soffice, "--headless", "--convert-to", "csv", "--outdir", str(tmpdir), str(path)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) candidates = [ p for p in tmpdir.rglob("*") if p.is_file() and p not in before and p.suffix.lower() == ".csv" ] if not candidates: candidates = [p for p in tmpdir.rglob("*") if p.is_file() and p.suffix.lower() == ".csv"] if not candidates: raise RuntimeError("soffice conversion completed but no CSV was produced.") return candidates[0] def input_to_csv(path: Path) -> Path: suffix = path.suffix.lower() if suffix == ".csv": return path if suffix in {".xls", ".xlsx"}: return convert_excel_to_csv(path) raise RuntimeError(f"Unsupported file type: {suffix}") def normalize_row(raw: list[str]) -> dict[str, Any]: row = {name: clean_text(raw[i]) if i < len(raw) else "" for i, name in enumerate(RAW_COLUMNS)} amount = parse_amount(row["中标金额（元）"]) publish_date = parse_date(row["信息发布时间"]) days = parse_days(row["合同工期"]) row["金额数值"] = amount row["金额_万元"] = round(amount / 10000, 4) if amount is not None else "" row["发布时间"] = publish_date.strftime("%Y-%m-%d") if publish_date else "" row["月份"] = publish_date.strftime("%Y-%m") if publish_date else "未知" row["合同工期天数"] = days if days is not None else "" row["招标联系人可用"] = bool(row["招标单位联系人"] or row["招标单位联系人电话"]) row["中标联系人可用"] = bool(row["中标单位联系人"] or row["中标单位联系人电话"]) row["需求类型"] = classify_demand(row["项目名称"], row["关键词"]) row["客户类型"] = classify_client(row["招标单位"]) return row def read_rows(csv_path: Path) -> list[dict[str, Any]]: with csv_path.open("r", encoding="utf-8-sig", newline="") as f: reader = list(csv.reader(f)) data_rows = reader[2:] if len(reader) >= 2 and reader[1][:2] == ["关键词", "项目名称"] else reader[1:] rows: list[dict[str, Any]] = [] for raw in data_rows: if not any(clean_text(cell) for cell in raw): continue if clean_text(raw[0] if raw else "") == "关键词" and clean_text(raw[1] if len(raw) > 1 else "") == "项目名称": continue rows.append(normalize_row(raw)) return rows def classify_demand(title: str, keyword: str) -> str: text = f"{title} {keyword}" if any(token in text for token in ["维保", "维护", "运维", "保养"]): return "系统维保/运维" if any(token in text for token in ["餐饮服务", "食堂服务", "运营", "劳务", "配送", "食材"]): return "餐饮/运营服务" if any(token in text for token in ["设备", "刷脸", "机器人", "终端", "闸机", "收银", "机具", "硬件"]): return "硬件设备/终端" if any(token in text for token in ["平台", "系统", "软件", "监管", "一卡通", "信息化", "数字化", "升级"]): return "软件系统/平台" if any(token in text for token in ["建设", "改造", "装修", "工程", "采购项目"]): return "综合建设/集成" return "智慧食堂综合项目" def classify_client(name: str) -> str: if not name: return "未识别" if any(token in name for token in ["银行", "建行", "农行", "工行", "中行", "交行", "邮储"]): return "银行" if any(token in name for token in ["医院", "卫生院", "卫生健康", "疾控"]): return "医院" if any(token in name for token in ["大学", "学院", "学校", "中学", "小学", "幼儿园", "教育", "职校"]): return "学校" if any(token in name for token in ["园区", "产业园", "开发区", "高新区", "城投", "产投", "科创"]): return "园区/产业平台" if any(token in name for token in ["烟草", "能源", "电力", "煤", "石油", "铁路", "机场", "集团", "国投", "国网", "中煤", "中建", "中铁", "中交"]): return "国企" if any(token in name for token in ["局", "委员会", "人民政府", "街道", "法院", "检察", "公安", "司法", "税务", "监狱", "中心", "机关", "管理处"]): return "政府/事业单位" if "公司" in name or "有限" in name: return "企业/其他公司" return "其他" def group_stats(rows: list[dict[str, Any]], keys: list[str]) -> list[dict[str, Any]]: groups: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list) for row in rows: key = tuple(display_text(row.get(k)) for k in keys) groups[key].append(row) result = [] for key, items in groups.items(): amounts = [r["金额数值"] for r in items if r.get("金额数值") is not None] total = sum(amounts) result.append( { **{keys[i]: key[i] for i in range(len(keys))}, "项目数": len(items), "披露金额项目数": len(amounts), "金额合计": round(total, 2), "平均金额": round(total / len(amounts), 2) if amounts else "", "中位数金额": round(statistics.median(amounts), 2) if amounts else "", "P90金额": round(quantile(amounts, 0.9), 2) if amounts else "", "最大金额": round(max(amounts), 2) if amounts else "", } ) return sorted(result, key=lambda item: (item["项目数"], item["金额合计"]), reverse=True) def top_counter(rows: list[dict[str, Any]], column: str, limit: int) -> list[tuple[str, int]]: return Counter(display_text(row.get(column)) for row in rows).most_common(limit) def detect_duplicates(rows: list[dict[str, Any]]) -> dict[str, set[int]]: buckets: dict[str, dict[str, list[int]]] = { "项目编号重复": defaultdict(list), "公告链接重复": defaultdict(list), "项目名称重复": defaultdict(list), "项目-客户-供应商-金额重复": defaultdict(list), } for idx, row in enumerate(rows, start=1): if row.get("项目编号"): buckets["项目编号重复"][row["项目编号"]].append(idx) if row.get("官网查看地址"): buckets["公告链接重复"][row["官网查看地址"]].append(idx) if row.get("项目名称"): buckets["项目名称重复"][row["项目名称"]].append(idx) exact = "|".join( [ row.get("项目名称", ""), row.get("招标单位", ""), row.get("中标单位", ""), str(row.get("金额数值") or ""), ] ) if exact.strip("|"): buckets["项目-客户-供应商-金额重复"][exact].append(idx) return { name: {item for values in bucket.values() if len(values) > 1 for item in values} for name, bucket in buckets.items() } def detect_risks(rows: list[dict[str, Any]], p99: float | None, median: float | None) -> list[dict[str, Any]]: risks: list[dict[str, Any]] = [] duplicate_indexes = detect_duplicates(rows) supplier_counts = Counter(display_text(r.get("中标单位")) for r in rows if display_text(r.get("中标单位")) != "未填") supplier_region_counts = Counter( (display_text(r.get("中标单位")), display_text(r.get("发布省份"))) for r in rows if display_text(r.get("中标单位")) != "未填" ) client_supplier_counts = Counter( (display_text(r.get("招标单位")), display_text(r.get("中标单位"))) for r in rows if display_text(r.get("招标单位")) != "未填" and display_text(r.get("中标单位")) != "未填" ) def add(row_no: int, row: dict[str, Any], level: str, rule: str, reason: str) -> None: risks.append( { "记录号": row_no, "风险等级": level, "触发规则": rule, "复核原因": reason, "项目名称": row.get("项目名称", ""), "发布时间": row.get("发布时间", ""), "省份": row.get("发布省份", ""), "城市": row.get("发布市级", ""), "金额": row.get("金额数值") if row.get("金额数值") is not None else "", "招标单位": row.get("招标单位", ""), "中标单位": row.get("中标单位", ""), "官网查看地址": row.get("官网查看地址", ""), } ) for row_no, row in enumerate(rows, start=1): amount = row.get("金额数值") supplier = display_text(row.get("中标单位")) province = display_text(row.get("发布省份")) client = display_text(row.get("招标单位")) if amount is None: add(row_no, row, "黄", "金额缺失", "中标金额未披露，影响规模和商机评分。") elif amount <= 0: add(row_no, row, "红", "金额非正", "金额小于等于 0，需要核对原公告。") elif (p99 is not None and amount >= p99) or (median and amount > median * 50): add(row_no, row, "红", "异常大额", "金额高于 P99 或超过中位数 50 倍，可能拉高区域/供应商规模。") elif amount <= 1000: add(row_no, row, "黄", "异常小额", "金额低于 1000 元，可能为单价、零星采购或录入问题。") if not row.get("项目编号"): add(row_no, row, "黄", "项目编号缺失", "项目编号为空或为占位符，追溯和去重可信度下降。") if not row.get("招标单位"): add(row_no, row, "黄", "招标单位缺失", "缺少客户主体，无法进入客户画像。") if not row.get("中标单位"): add(row_no, row, "黄", "中标单位缺失", "缺少供应商主体，无法进入竞争格局。") if not row.get("官网查看地址"): add(row_no, row, "黄", "公告链接缺失", "缺少原始追溯链接。") days = row.get("合同工期天数") if isinstance(days, int) and days > 1825: add(row_no, row, "黄", "合同工期过长", "合同工期超过 5 年，需确认是服务期还是录入值。") elif isinstance(days, int) and 0 < days <= 7: add(row_no, row, "黄", "合同工期过短", "合同工期小于等于 7 天，需确认口径。") for rule, indexes in duplicate_indexes.items(): if row_no in indexes: add(row_no, row, "黄", rule, "存在同编号、同链接、同项目名或同组合重复记录，需要确认是否多包件/多公告。") if supplier != "未填" and supplier_counts[supplier] >= 5: add(row_no, row, "黄", "同一供应商频繁中标", f"{supplier} 在样本中中标 {supplier_counts[supplier]} 次。") if supplier != "未填" and supplier_region_counts[(supplier, province)] >= 3: add(row_no, row, "黄", "同一供应商区域集中", f"{supplier} 在 {province} 中标 {supplier_region_counts[(supplier, province)]} 次。") if supplier != "未填" and client != "未填" and client_supplier_counts[(client, supplier)] >= 2: add(row_no, row, "黄", "客户-供应商高频绑定", f"{client} 与 {supplier} 在样本中出现 {client_supplier_counts[(client, supplier)]} 次。") priority = {"红": 0, "黄": 1, "绿": 2} return sorted(risks, key=lambda item: (priority.get(item["风险等级"], 9), item["记录号"], item["触发规则"])) def score_opportunities(rows: list[dict[str, Any]], p75: float | None, p90: float | None, p99: float | None) -> list[dict[str, Any]]: province_counts = Counter(display_text(r.get("发布省份")) for r in rows) client_counts = Counter(display_text(r.get("招标单位")) for r in rows) client_amounts = defaultdict(float) supplier_counts = Counter(display_text(r.get("中标单位")) for r in rows) client_supplier_counts = Counter((display_text(r.get("招标单位")), display_text(r.get("中标单位"))) for r in rows) max_province = max(province_counts.values() or [1]) max_client = max(client_counts.values() or [1]) max_client_amount = 1.0 for row in rows: if row.get("金额数值") is not None: client_amounts[display_text(row.get("招标单位"))] += float(row["金额数值"]) if client_amounts: max_client_amount = max(client_amounts.values()) scored = [] for idx, row in enumerate(rows, start=1): province = display_text(row.get("发布省份")) client = display_text(row.get("招标单位")) supplier = display_text(row.get("中标单位")) amount = row.get("金额数值") heat = 6 + 8 * province_counts[province] / max_province + 6 * client_counts[client] / max_client if client == "未填": heat = 6 + 8 * province_counts[province] / max_province if amount is None: amount_score = 6 elif p99 and amount >= p99: amount_score = 16 elif p90 and amount >= p90: amount_score = 20 elif p75 and amount >= p75: amount_score = 17 else: amount_score = 10 + min(5, math.log10(max(amount, 1)) / 2) if client == "未填": client_value = 2 else: client_value = 6 + 6 * client_counts[client] / max_client + 6 * (client_amounts.get(client, 0) / max_client_amount) if client != "未填" and row["客户类型"] in {"银行", "学校", "医院", "政府/事业单位", "国企", "园区/产业平台"}: client_value += 2 supplier_freq = supplier_counts[supplier] binding = client_supplier_counts[(client, supplier)] competition = 20 if supplier_freq >= 5: competition -= 6 elif supplier_freq >= 3: competition -= 3 if binding >= 2: competition -= 5 if supplier == "未填": competition -= 3 competition = max(4, competition) actionability = 6 if row.get("官网查看地址"): actionability += 4 if row.get("项目编号"): actionability += 3 if row.get("招标联系人可用"): actionability += 4 if row.get("需求类型") != "智慧食堂综合项目": actionability += 3 if client == "未填": actionability -= 5 actionability = min(20, actionability) actionability = max(4, actionability) total = round(heat + amount_score + client_value + competition + actionability, 1) scored.append( { "排名依据记录号": idx, "总分": total, "市场热度": round(heat, 1), "金额吸引力": round(amount_score, 1), "客户价值": round(client_value, 1), "竞争可进入性": round(competition, 1), "可行动性": round(actionability, 1), "项目名称": row["项目名称"], "发布时间": row["发布时间"], "省份": row["发布省份"], "城市": row["发布市级"], "金额": amount if amount is not None else "", "客户类型": row["客户类型"], "需求类型": row["需求类型"], "招标单位": row["招标单位"], "中标单位": row["中标单位"], "行动建议": recommend_action(row, total), "官网查看地址": row["官网查看地址"], } ) return sorted(scored, key=lambda item: item["总分"], reverse=True) def recommend_action(row: dict[str, Any], score: float) -> str: demand = row["需求类型"] client_type = row["客户类型"] if score >= 80: priority = "本周优先" elif score >= 70: priority = "两周内跟进" else: priority = "纳入线索池" if row.get("招标联系人可用"): contact = "可先用公告联系人核验需求和后续采购计划" else: contact = "需补齐客户联系人或通过官网/企查渠道找关键人" return f"{priority}：围绕{client_type}的{demand}复盘中标方案，{contact}。" def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) if not fieldnames: seen = [] for row in rows: for key in row: if key not in seen: seen.append(key) fieldnames = seen with path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def markdown_table(headers: list[str], rows: list[list[Any]], limit: int | None = None) -> str: if limit is not None: rows = rows[:limit] lines = [ "| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |", ] for row in rows: safe = [str(cell).replace("\n", " ").replace("|", "/") for cell in row] lines.append("| " + " | ".join(safe) + " |") return "\n".join(lines) def summarize(records: list[dict[str, Any]], key: str, value_key: str = "金额合计", limit: int = 10) -> list[list[Any]]: rows = [] for item in records[:limit]: rows.append( [ item.get(key, ""), item.get("项目数", 0), money(item.get(value_key) or 0), money(item.get("中位数金额") if item.get("中位数金额") != "" else None), ] ) return rows def generate_report( *, source: Path, rows: list[dict[str, Any]], monthly: list[dict[str, Any]], province: list[dict[str, Any]], city: list[dict[str, Any]], client: list[dict[str, Any]], supplier: list[dict[str, Any]], demand: list[dict[str, Any]], opportunities: list[dict[str, Any]], risks: list[dict[str, Any]], output_dir: Path, ) -> str: amounts = [r["金额数值"] for r in rows if r.get("金额数值") is not None] total = sum(amounts) median = statistics.median(amounts) if amounts else None avg = total / len(amounts) if amounts else None p90 = quantile(amounts, 0.9) p99 = quantile(amounts, 0.99) max_amount = max(amounts) if amounts else None max_share = max_amount / total if amounts and total else 0 amount_missing = len(rows) - len(amounts) project_id_missing = sum(1 for r in rows if not r.get("项目编号")) tenderer_missing = sum(1 for r in rows if not r.get("招标单位")) supplier_missing = sum(1 for r in rows if not r.get("中标单位")) contact_available = sum(1 for r in rows if r.get("招标联系人可用")) supplier_total = [item for item in group_stats(rows, ["中标单位"]) if item["中标单位"] != "未填"] supplier_by_count = sorted(supplier_total, key=lambda item: (item["项目数"], item["金额合计"]), reverse=True) supplier_by_amount = sorted(supplier_total, key=lambda item: (item["金额合计"], item["项目数"]), reverse=True) top5_amount = sum(item["金额合计"] for item in supplier_by_amount[:5]) top10_amount = sum(item["金额合计"] for item in supplier_by_amount[:10]) top5_count = sum(item["项目数"] for item in supplier_by_count[:5]) top10_count = sum(item["项目数"] for item in supplier_by_count[:10]) red_risks = sum(1 for r in risks if r["风险等级"] == "红") yellow_risks = sum(1 for r in risks if r["风险等级"] == "黄") quality_score = 100 quality_score -= min(20, amount_missing / max(len(rows), 1) * 20) quality_score -= min(15, project_id_missing / max(len(rows), 1) * 15) quality_score -= min(10, (tenderer_missing + supplier_missing) / max(len(rows), 1) * 10) quality_score -= min(10, len([r for r in risks if "重复" in r["触发规则"]]) / max(len(rows), 1) * 10) quality_score -= 5 if max_share > 0.3 else 0 quality_score = round(max(0, quality_score), 1) best_regions = sorted(province, key=lambda item: (item["项目数"], item["金额合计"]), reverse=True)[:5] high_amount_regions = sorted(province, key=lambda item: item["金额合计"], reverse=True)[:5] low_amount_high_freq = sorted( [item for item in province if item.get("中位数金额") != "" and item["项目数"] >= 10], key=lambda item: (item["中位数金额"], -item["项目数"]), )[:5] client_rows = [] for item in [record for record in client if record["招标单位"] != "未填"][:20]: sample = next((r for r in rows if display_text(r.get("招标单位")) == item["招标单位"]), {}) client_rows.append( [ item["招标单位"], sample.get("客户类型", "未识别"), item["项目数"], money(item["金额合计"]), money(item["中位数金额"] if item["中位数金额"] != "" else None), "高频/高金额/有公告链路" if item["项目数"] >= 2 or item["金额合计"] >= (p90 or 0) else "单点线索，先核验后续计划", ] ) supplier_focus = [] seen_suppliers = set() for item in supplier_by_count[:12] + supplier_by_amount[:12]: if item["中标单位"] in seen_suppliers: continue seen_suppliers.add(item["中标单位"]) supplier_focus.append(item) supplier_rows = [] for item in supplier_focus[:20]: sample = next((r for r in rows if display_text(r.get("中标单位")) == item["中标单位"]), {}) supplier_rows.append( [ item["中标单位"], item["项目数"], money(item["金额合计"]), money(item["中位数金额"] if item["中位数金额"] != "" else None), sample.get("发布省份", ""), "标杆/竞品" if item["项目数"] >= 3 or item["金额合计"] >= (p90 or 0) else "区域型供应商", ] ) opp_rows = [ [ idx + 1, item["总分"], item["项目名称"], f"{item['省份']}/{item['城市']}", money(item["金额"] if item["金额"] != "" else None), item["客户类型"], item["需求类型"], item["招标单位"], item["行动建议"], ] for idx, item in enumerate(opportunities[:30]) ] risk_rows = [ [ item["风险等级"], item["触发规则"], item["项目名称"], f"{item['省份']}/{item['城市']}", money(item["金额"] if item["金额"] != "" else None), item["复核原因"], ] for item in risks[:40] ] lines = [ "# 2026年1-6月乙方宝智慧食堂中标通知分析报告", "", f"- 原始文件: `{source}`", f"- 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", f"- 数据口径: 共 {len(rows)} 条中标通知，{len(amounts)} 条披露金额；联系人和电话仅用于可行动性判断，主报告不展开敏感明细。", f"- 产物目录: `{output_dir}`", "", "## A. 一页高管摘要", "", f"本批数据显示，智慧食堂相关中标通知在 2026 年 1-6 月共有 {len(rows)} 条，披露金额合计 {money(total)}，披露金额中位数 {money(median)}，P90 为 {money(p90)}。最大单项为 {money(max_amount)}，占披露金额合计 {pct(max_share)}，因此金额合计和均值会被超大项目显著拉高，正式经营判断应优先看中位数、P90、项目数和客户复购迹象。", "", f"数据质量评分为 {quality_score}/100。主要短板是金额缺失 {amount_missing} 条、项目编号缺失 {project_id_missing} 条、招标单位缺失 {tenderer_missing} 条、中标单位缺失 {supplier_missing} 条。招标侧联系人或电话可用记录 {contact_available} 条，足以支撑第一轮销售线索核验，但仍需要补齐客户关键人、预算来源、采购阶段和合同包件信息。", "", f"供应商集中度方面，按金额排序的 Top 5 供应商占披露金额 {pct(top5_amount / total if total else 0)}、Top 10 占 {pct(top10_amount / total if total else 0)}；按中标次数排序的 Top 5 供应商占项目数 {pct(top5_count / len(rows) if rows else 0)}、Top 10 占 {pct(top10_count / len(rows) if rows else 0)}。若剔除异常超大项目，需要重新评估真实竞争格局。", "", "## B. 关键数据表", "", "### 月度趋势", markdown_table( ["月份", "项目数", "披露金额项目数", "金额合计", "平均金额", "中位数金额", "P90"], [ [ item["月份"], item["项目数"], item["披露金额项目数"], money(item["金额合计"]), money(item["平均金额"] if item["平均金额"] != "" else None), money(item["中位数金额"] if item["中位数金额"] != "" else None), money(item["P90金额"] if item["P90金额"] != "" else None), ] for item in sorted(monthly, key=lambda x: x["月份"]) ], ), "", "### 省份 Top 10", markdown_table(["省份", "项目数", "金额合计", "中位数金额"], summarize(province, "发布省份", limit=10)), "", "### 城市 Top 10", markdown_table(["城市", "项目数", "金额合计", "中位数金额"], summarize(city, "发布市级", limit=10)), "", "### 需求类型", markdown_table(["需求类型", "项目数", "金额合计", "中位数金额"], summarize(demand, "需求类型", limit=10)), "", "### 高增长/高价值/低金额高频区域", markdown_table(["区域类型", "省份", "项目数", "金额合计", "中位数金额"], [["高频区域", r["发布省份"], r["项目数"], money(r["金额合计"]), money(r["中位数金额"] if r["中位数金额"] != "" else None)] for r in best_regions]), "", markdown_table(["区域类型", "省份", "项目数", "金额合计", "中位数金额"], [["高金额区域", r["发布省份"], r["项目数"], money(r["金额合计"]), money(r["中位数金额"] if r["中位数金额"] != "" else None)] for r in high_amount_regions]), "", markdown_table(["区域类型", "省份", "项目数", "金额合计", "中位数金额"], [["低金额高频", r["发布省份"], r["项目数"], money(r["金额合计"]), money(r["中位数金额"] if r["中位数金额"] != "" else None)] for r in low_amount_high_freq]), "", "## C. 重点客户清单 Top 20", "", markdown_table(["招标单位", "客户类型", "项目数", "金额合计", "中位数金额", "跟进理由"], client_rows, limit=20), "", "## D. 重点供应商/竞品清单 Top 20", "", markdown_table(["中标单位", "中标次数", "金额合计", "中位数金额", "样本省份", "判断"], supplier_rows, limit=20), "", "## E. 商机 Top 30", "", markdown_table(["排名", "总分", "项目名称", "区域", "金额", "客户类型", "需求类型", "招标单位", "下一步动作"], opp_rows, limit=30), "", "## F. 风险红旗清单", "", f"共生成 {len(risks)} 条红旗触发项，其中红色 {red_risks} 条、黄色 {yellow_risks} 条。红旗只代表需要复核，不代表违规。", "", markdown_table(["等级", "触发规则", "项目名称", "区域", "金额", "复核原因"], risk_rows, limit=40), "", "## G. 下周可执行行动计划", "", markdown_table( ["动作", "责任人", "完成时间", "交付物", "验收标准"], [ ["核验 Top 30 商机客户关键人和预算来源", "销售/市场", "下周三", "客户联系人补全表", "Top 30 至少补齐 80% 客户关键人或替代触达路径"], ["复盘高分项目的中标方案和供应商打法", "售前/产品", "下周四", "竞品方案拆解", "完成 Top 10 项目的一页式竞品复盘"], ["剔除异常大额后重算区域和供应商集中度", "数据分析", "下周二", "二次口径仪表盘", "给出含/不含异常项目两套结论"], ["建立客户类型和单位别名表", "运营/数据", "下周五", "unit-aliases.csv", "覆盖重点客户 Top 50 和供应商 Top 50"], ["对低金额高频区域设计轻量产品包", "产品/销售", "下周五", "区域打法清单", "明确目标区域、标准报价区间和首批触达名单"], ], ), "", "## H. 需要补充的数据字段", "", "- 采购预算来源、采购方式、包件号、是否续约/维保、是否框架协议。", "- 招标公告发布日期、成交公告发布日期、合同签订日期，用于判断采购周期。", "- 客户所属行业、客户规模、食堂规模、用餐人数、现有系统厂商。", "- 中标方案范围：软件模块、硬件清单、服务内容、部署模式、付款条款。", "- 失败/未中标供应商名单，用于判断真实竞争强度。", "- 人工确认的单位别名和集团归属字段。", "", "## 数据质量与口径说明", "", markdown_table( ["指标", "数值"], [ ["记录数", len(rows)], ["披露金额记录数", len(amounts)], ["金额缺失", amount_missing], ["项目编号缺失", project_id_missing], ["招标单位缺失", tenderer_missing], ["中标单位缺失", supplier_missing], ["金额合计", money(total)], ["平均金额", money(avg)], ["中位数金额", money(median)], ["P90", money(p90)], ["P99", money(p99)], ["最大金额占比", pct(max_share)], ], ), "", "## 经营分析闭环", "", "### 仪表盘摘要", "", f"- 项目数: {len(rows)}；披露金额: {money(total)}；中位数: {money(median)}；P90: {money(p90)}。", f"- 最高频省份: {', '.join([r['发布省份'] for r in best_regions[:5]])}。", f"- 主要需求: {', '.join([r['需求类型'] for r in demand[:5]])}。", f"- 供应商 Top 5 金额集中度: {pct(top5_amount / total if total else 0)}；Top 5 次数集中度: {pct(top5_count / len(rows) if rows else 0)}。", "", "### 红黄绿预警", "", "- 红: 超大额项目会显著扭曲金额合计，需要单独复核和剔除重算。", "- 黄: 金额缺失和项目编号缺失比例较高，影响正式市场规模测算。", "- 绿: 公告链接完整，具备追溯复核基础；多数记录有明确省市和项目名称，可支撑区域/需求画像。", "", "### 关键差距与根因假设", "", "- 目标差距: 当前只能证明中标事实，不能直接证明潜在客户未来预算。", "- 资源差距: 客户联系人和关键决策链不足，销售需要二次补全。", "- 能力差距: 缺少单位别名、集团归属和包件拆分规则，导致供应商集中度和客户复购判断偏粗。", "- 根因假设: 平台抓取字段偏公告结果而非采购全生命周期，且原始表未维护组织标准化主数据。", "", "### 机会清单", "", "- 优先跟进高分商机 Top 30；对学校、医院、银行、国企客户建立行业案例话术。", "- 对低金额高频区域，准备标准化轻量方案和维保替换方案。", "- 对异常高金额项目，复核是否包含土建、餐饮运营、硬件大包或长期框架合同，避免误判软件预算。", "", "### 改进行动", "", "- 建立 `unit-aliases.csv`，维护招标单位和中标单位标准名、集团名、行业标签。", "- 形成“含异常/不含异常”双口径仪表盘。", "- 将商机评分模型固化为月度复盘脚本，每月追加新公告后自动输出 Top 商机和红旗。", ] return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser(description="Analyze bid-award data and generate report artifacts.") parser.add_argument("input", type=Path, help="Input .csv/.xls/.xlsx file") parser.add_argument("--output-dir", type=Path, default=Path("data/processed/20260612-yifangbao")) parser.add_argument("--report", type=Path, default=Path("docs/reports/2026-h1-yifangbao-bid-analysis.md")) args = parser.parse_args() csv_path = input_to_csv(args.input) rows = read_rows(csv_path) if not rows: raise RuntimeError("No data rows found.") amounts = [r["金额数值"] for r in rows if r.get("金额数值") is not None] p75 = quantile(amounts, 0.75) p90 = quantile(amounts, 0.9) p99 = quantile(amounts, 0.99) median = statistics.median(amounts) if amounts else None monthly = group_stats(rows, ["月份"]) province = group_stats(rows, ["发布省份"]) city = group_stats(rows, ["发布市级"]) client = group_stats(rows, ["招标单位"]) supplier = group_stats(rows, ["中标单位"]) demand = group_stats(rows, ["需求类型"]) client_type = group_stats(rows, ["客户类型"]) risks = detect_risks(rows, p99, median) opportunities = score_opportunities(rows, p75, p90, p99) output_dir = args.output_dir output_dir.mkdir(parents=True, exist_ok=True) write_csv(output_dir / "cleaned_awards.csv", rows) write_csv(output_dir / "monthly_summary.csv", sorted(monthly, key=lambda x: x["月份"])) write_csv(output_dir / "province_summary.csv", province) write_csv(output_dir / "city_summary.csv", city) write_csv(output_dir / "client_top20.csv", client[:20]) supplier_by_count = [item for item in supplier if item["中标单位"] != "未填"] supplier_by_amount = sorted(supplier_by_count, key=lambda item: (item["金额合计"], item["项目数"]), reverse=True) write_csv(output_dir / "supplier_top20.csv", supplier_by_count[:20]) write_csv(output_dir / "supplier_amount_top20.csv", supplier_by_amount[:20]) write_csv(output_dir / "demand_summary.csv", demand) write_csv(output_dir / "client_type_summary.csv", client_type) write_csv(output_dir / "opportunity_top30.csv", opportunities[:30]) write_csv(output_dir / "risk_flags.csv", risks) summary = { "source": str(args.input), "row_count": len(rows), "amount_count": len(amounts), "amount_total": round(sum(amounts), 2), "amount_median": round(median, 2) if median is not None else None, "amount_p75": round(p75, 2) if p75 is not None else None, "amount_p90": round(p90, 2) if p90 is not None else None, "amount_p99": round(p99, 2) if p99 is not None else None, "max_amount": round(max(amounts), 2) if amounts else None, "risk_count": len(risks), "opportunity_count": len(opportunities), "generated_at": datetime.now().isoformat(timespec="seconds"), } (output_dir / "analysis_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") report = generate_report( source=args.input, rows=rows, monthly=monthly, province=province, city=city, client=client, supplier=supplier, demand=demand, opportunities=opportunities, risks=risks, output_dir=output_dir, ) args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(report, encoding="utf-8") print(json.dumps({"report": str(args.report), "output_dir": str(output_dir), **summary}, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())