#!/usr/bin/env python3 """Export projects above a large-amount threshold for manual review.""" from __future__ import annotations import argparse import html import re import zipfile from collections import defaultdict from datetime import datetime from pathlib import Path from typing import Any from xml.etree import ElementTree as ET NS_MAIN = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" NS_REL = "http://schemas.openxmlformats.org/officeDocument/2006/relationships" def cell_text(cell: ET.Element) -> str: if cell.attrib.get("t") == "inlineStr": return "".join(node.text or "" for node in cell.findall(f".//{{{NS_MAIN}}}t")) value = cell.find(f"{{{NS_MAIN}}}v") return value.text if value is not None and value.text is not None else "" def read_sheet_rows(xlsx_path: Path, sheet_name: str) -> list[dict[str, str]]: with zipfile.ZipFile(xlsx_path) as zf: workbook = ET.fromstring(zf.read("xl/workbook.xml")) rels_root = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels")) rels = {item.attrib["Id"]: item.attrib["Target"] for item in rels_root} target = None sheets_node = workbook.find(f"{{{NS_MAIN}}}sheets") if sheets_node is None: raise ValueError("Workbook has no sheets node.") for sheet in sheets_node: if sheet.attrib.get("name") == sheet_name: rel_id = sheet.attrib[f"{{{NS_REL}}}id"] target = rels[rel_id] break if target is None: raise ValueError(f"Sheet not found: {sheet_name}") worksheet = ET.fromstring(zf.read(f"xl/{target}")) raw_rows: list[list[str]] = [] for row in worksheet.findall(f".//{{{NS_MAIN}}}sheetData/{{{NS_MAIN}}}row"): raw_rows.append([cell_text(cell) for cell in row.findall(f"{{{NS_MAIN}}}c")]) if not raw_rows: return [] headers = raw_rows[0] rows = [] for raw in raw_rows[1:]: if not any(value.strip() for value in raw): continue rows.append({header: raw[idx] if idx < len(raw) else "" for idx, header in enumerate(headers)}) return rows def parse_amount(value: Any) -> float | None: text = str(value or "").strip().replace(",", "") for token in ("元", "￥", "¥"): text = text.replace(token, "") if not text: return None try: return float(text) except ValueError: return None def money_yuan(value: float | 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 normalize_title(title: str) -> str: return re.sub(r"\d+", "#", title or "") def build_review_rows(rows: list[dict[str, str]], threshold: float) -> list[dict[str, str]]: selected = [] for row in rows: amount = parse_amount(row.get("金额数值") or row.get("中标金额（元）")) if amount is None or amount <= threshold: continue review = dict(row) review["审核金额"] = f"{amount:.2f}" review["金额展示"] = money_yuan(amount) review["审核要点"] = "超过 1 亿元；核对公告金额单位、合同范围与是否为总包/食材劳务合并口径。" selected.append(review) groups: dict[tuple[str, str, str, str], list[dict[str, str]]] = defaultdict(list) for row in selected: groups[ ( row.get("发布时间") or row.get("信息发布时间", ""), row.get("招标单位", ""), row.get("中标单位", ""), row.get("审核金额", ""), ) ].append(row) for group in groups.values(): if len(group) < 2: continue normalized_titles = {normalize_title(item.get("项目名称", "")) for item in group} if len(normalized_titles) <= 2: for item in group: item["审核要点"] += " 同日同采购方/供应商/金额出现多条相似记录，重点排查重复公告、标段拆分或金额录入问题。" return sorted(selected, key=lambda item: float(item["审核金额"]), reverse=True) def esc(value: Any) -> str: return html.escape(str(value or ""), quote=True) def render_html(rows: list[dict[str, str]], source: Path, threshold: float) -> str: total_amount = sum(float(row["审核金额"]) for row in rows) generated_at = datetime.now().strftime("%Y-%m-%d %H:%M") threshold_label = money_yuan(threshold) table_rows = [] for idx, row in enumerate(rows, start=1): url = row.get("官网查看地址", "") table_rows.append( f""" {idx} {esc(row.get("项目名称"))}{esc(row.get("项目编号")) or "项目编号未披露"} {esc(row.get("发布时间") or row.get("信息发布时间"))} {esc(row.get("发布省份"))}/{esc(row.get("发布市级"))} {esc(row.get("金额展示"))}{esc(row.get("审核金额"))} 元 {esc(row.get("招标单位"))} {esc(row.get("中标单位"))} {esc(row.get("审核要点"))} {f'查看公告' if url else "未披露"} """ ) empty_state = "" if not rows: empty_state = '

未发现超过阈值的项目。

' return f""" 超一亿元项目审核清单
人工复核清单

超一亿元项目审核清单

筛选口径：中标金额（元）/金额数值 > {esc(threshold_label)}；来源表：原始清洗数据；生成时间：{esc(generated_at)}

来源工作簿
{esc(source.name)}

待审核项目数{len(rows)} 条
超阈值金额合计{esc(money_yuan(total_amount))}
单项金额阈值{esc(threshold_label)}
{empty_state} {''.join(table_rows)}
# 项目 / 编号 发布时间 地区 中标金额 招标单位 中标单位 审核要点 证据链接

两条记录同日、同采购方、同供应商、同金额，且项目编号仅尾号不同；审核时优先确认是否为重复公告、两个标段，或金额单位录入为元/万元时发生误读。

本清单不删除原始数据，只作为人工复核入口；最终保留、合并或修正金额应以官方公告详情和合同口径为准。

""" def xml_escape(value: Any) -> str: return ( str(value or "") .replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) ) def col_letter(n: int) -> str: result = "" while n: n, rem = divmod(n - 1, 26) result = chr(65 + rem) + result return result def cell_xml(ref: str, value: Any, style: int = 0) -> str: if value is None or value == "": return f'' number = parse_amount(value) if number is not None and str(value).strip() == str(value).strip().replace(",", ""): return f'{number}' return f'{xml_escape(value)}' def row_xml(row_num: int, values: list[Any], styles: list[int] | None = None) -> str: styles = styles or [0] * len(values) return "" + "".join( cell_xml(f"{col_letter(idx)}{row_num}", value, styles[idx - 1] if idx - 1 < len(styles) else 0) for idx, value in enumerate(values, start=1) ) + "" def write_xlsx(rows: list[dict[str, str]], output: Path) -> None: headers = [ "序号", "项目名称", "项目编号", "发布时间", "省份", "城市", "中标金额（元）", "金额展示", "招标单位", "中标单位", "审核要点", "官网查看地址", ] data = [] for idx, row in enumerate(rows, start=1): data.append( [ idx, row.get("项目名称", ""), row.get("项目编号", ""), row.get("发布时间") or row.get("信息发布时间", ""), row.get("发布省份", ""), row.get("发布市级", ""), row.get("审核金额", ""), row.get("金额展示", ""), row.get("招标单位", ""), row.get("中标单位", ""), row.get("审核要点", ""), row.get("官网查看地址", ""), ] ) widths = [8, 56, 22, 14, 10, 12, 18, 14, 28, 30, 58, 64] cols = "".join( f'' for idx, width in enumerate(widths, start=1) ) sheet_rows = [row_xml(1, headers, [1] * len(headers))] sheet_rows.extend(row_xml(idx, row, [0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0]) for idx, row in enumerate(data, start=2)) last_row = max(len(data) + 1, 1) worksheet = f""" {cols} {''.join(sheet_rows)} """ styles = f""" """ content_types = """ """ workbook = f""" """ root_rels = """ """ workbook_rels = """ """ output.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as zf: zf.writestr("[Content_Types].xml", content_types) zf.writestr("_rels/.rels", root_rels) zf.writestr("xl/workbook.xml", workbook) zf.writestr("xl/_rels/workbook.xml.rels", workbook_rels) zf.writestr("xl/styles.xml", styles) zf.writestr("xl/worksheets/sheet1.xml", worksheet) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--source", default="docs/deliverables/2026-h1-yifangbao-bid-analysis-workbook.xlsx") parser.add_argument("--threshold", type=float, default=100000000) parser.add_argument("--html", default="docs/reports/2026-h1-yifangbao-large-amount-review.html") parser.add_argument("--xlsx", default="docs/deliverables/2026-h1-yifangbao-large-amount-review.xlsx") args = parser.parse_args() source = Path(args.source) rows = read_sheet_rows(source, "原始清洗数据") selected = build_review_rows(rows, args.threshold) Path(args.html).write_text(render_html(selected, source, args.threshold), encoding="utf-8") write_xlsx(selected, Path(args.xlsx)) print(f"source_rows={len(rows)} selected_rows={len(selected)} threshold={args.threshold:.0f}") print(f"html={args.html}") print(f"xlsx={args.xlsx}") if __name__ == "__main__": main()