#!/usr/bin/env python3 """Build a CRM lead-import CSV from the July Yifangbao workbook template.""" from __future__ import annotations import argparse import csv from pathlib import Path from typing import Any from openpyxl import load_workbook ROOT = Path(__file__).resolve().parents[1] DEFAULT_TEMPLATE = Path("/Users/zhangling/Desktop/lead_import_template (4).csv") DEFAULT_INPUT = ROOT / "docs/final-results/20260813-yifangbao-phone-legal-enrichment/7月线索导入模板-0715-0812-电话法人已补全-中标单位全量补查.xlsx" DEFAULT_OUTPUT_DIR = ROOT / "docs/final-results/20260820-yifangbao-july-crm-import" DEFAULT_OUTPUT = DEFAULT_OUTPUT_DIR / "7月线索导入模板-按上月CRM模板-剔除无中标单位.csv" CONSTANTS = { "线索来源": "乙方宝-数科", "所属部门": "数科业务部", "所属区域公海": "数字体育科技事业部公海池", } OWNER_BY_PROVINCE = { "广东": "刘劲玮", "广西": "刘劲玮", "湖南": "刘劲玮", "四川": "刘劲玮", "重庆": "刘劲玮", "云南": "刘劲玮", "贵州": "刘劲玮", "福建": "刘劲玮", "江西": "刘劲玮", "海南": "刘劲玮", "上海": "孔令颖", "河南": "潘辉", "湖北": "潘辉", "陕西": "申海宇", "青海": "申海宇", "新疆": "申海宇", "西藏": "申海宇", "黑龙江": "申海宇", "吉林": "申海宇", "辽宁": "申海宇", "山西": "申海宇", "山东": "边豪", "浙江": "边豪", "宁夏": "边豪", "甘肃": "边豪", "天津": "杜佳", "河北": "薛桢", "内蒙古": "薛桢", "内蒙": "薛桢", "江苏": "高成君", "安徽": "高成君", } BEIJING_OWNERS = ["薛桢", "王一茗", "杜佳"] PROVINCE_FULL = { "北京": "北京市", "天津": "天津市", "上海": "上海市", "重庆": "重庆市", "河北": "河北省", "山西": "山西省", "辽宁": "辽宁省", "吉林": "吉林省", "黑龙江": "黑龙江省", "江苏": "江苏省", "浙江": "浙江省", "安徽": "安徽省", "福建": "福建省", "江西": "江西省", "山东": "山东省", "河南": "河南省", "湖北": "湖北省", "湖南": "湖南省", "广东": "广东省", "海南": "海南省", "四川": "四川省", "贵州": "贵州省", "云南": "云南省", "陕西": "陕西省", "甘肃": "甘肃省", "青海": "青海省", "台湾": "台湾省", "内蒙古": "内蒙古自治区", "内蒙": "内蒙古自治区", "广西": "广西壮族自治区", "西藏": "西藏自治区", "宁夏": "宁夏回族自治区", "新疆": "新疆维吾尔自治区", "香港": "香港特别行政区", "澳门": "澳门特别行政区", } def clean(value: Any) -> str: if value is None: return "" text = str(value).strip() if text in {"--", "暂未公布", "None", "nan"}: return "" if text.endswith(".0") and text[:-2].isdigit(): return text[:-2] return text def normalize_province(value: str) -> str: value = clean(value).replace("省", "").replace("市", "") value = value.replace("自治区", "").replace("壮族", "").replace("回族", "").replace("维吾尔", "") value = value.replace("特别行政区", "") if value == "内蒙古": return "内蒙古" return value def full_province(value: str) -> str: key = normalize_province(value) return PROVINCE_FULL.get(key, clean(value)) def city_name(value: str, province: str) -> str: value = clean(value) if not value: return "" if province in {"北京", "天津", "上海", "重庆"} and value in {province, f"{province}市"}: return f"{province}市" if value.endswith(("市", "自治州", "地区", "盟", "县", "区")): return value return f"{value}市" def owner_for(province: str, beijing_index: int) -> tuple[str, int]: key = normalize_province(province) if key == "北京": owner = BEIJING_OWNERS[beijing_index % len(BEIJING_OWNERS)] return owner, beijing_index + 1 return OWNER_BY_PROVINCE.get(key, ""), beijing_index def load_template(path: Path) -> list[str]: with path.open("r", encoding="gb18030", newline="") as fh: reader = csv.reader(fh) return next(reader) def build_rows(input_path: Path, headers: list[str]) -> tuple[list[list[str]], dict[str, Any]]: wb = load_workbook(input_path, data_only=True) ws = wb["已补电话-新增标黄"] source_headers = [clean(ws.cell(2, col).value) for col in range(1, ws.max_column + 1)] col = {header: idx + 1 for idx, header in enumerate(source_headers) if header} rows: list[list[str]] = [] excluded = 0 beijing_index = 0 owner_counts: dict[str, int] = {} required = [ "项目名称", "信息发布时间", "发布省份", "发布市级", "中标金额（元）", "单位名称", "招标单位联系人", "招标单位联系人电话", "中标单位联系人", "中标单位联系人电话", "官网查看地址", "最终可用电话（原表优先）", "法定代表人（登记参考，不等同联系人）", ] missing = [name for name in required if name not in col] if missing: raise RuntimeError(f"missing source columns: {missing}") for row_idx in range(3, ws.max_row + 1): winbid_unit = clean(ws.cell(row_idx, 13).value) if not winbid_unit: excluded += 1 continue province_short = normalize_province(ws.cell(row_idx, col["发布省份"]).value) owner, beijing_index = owner_for(province_short, beijing_index) if owner: owner_counts[owner] = owner_counts.get(owner, 0) + 1 out = {header: "" for header in headers} out.update(CONSTANTS) out["销售线索所有人"] = owner out["联系人"] = owner out["省"] = full_province(province_short) out["市"] = city_name(ws.cell(row_idx, col["发布市级"]).value, province_short) out["项目名称"] = clean(ws.cell(row_idx, col["项目名称"]).value) out["招标单位"] = clean(ws.cell(row_idx, 10).value) out["中标金额"] = clean(ws.cell(row_idx, col["中标金额（元）"]).value) out["信息发布时间"] = clean(ws.cell(row_idx, col["信息发布时间"]).value) out["招标单位联系人"] = clean(ws.cell(row_idx, col["招标单位联系人"]).value) out["招标单位联系人电话"] = clean(ws.cell(row_idx, col["招标单位联系人电话"]).value) out["中标单位名称"] = winbid_unit out["中标单位联系人"] = clean(ws.cell(row_idx, col["中标单位联系人"]).value) final_phone = clean(ws.cell(row_idx, col["最终可用电话（原表优先）"]).value) original_winbid_phone = clean(ws.cell(row_idx, col["中标单位联系人电话"]).value) out["中标单位联系人电话"] = original_winbid_phone or final_phone out["官网查看地址"] = clean(ws.cell(row_idx, col["官网查看地址"]).value) out["天眼查法定代表人"] = clean(ws.cell(row_idx, col["法定代表人（登记参考，不等同联系人）"]).value) out["天眼查电话"] = final_phone rows.append([out.get(header, "") for header in headers]) return rows, {"included": len(rows), "excluded": excluded, "owner_counts": owner_counts} def write_csv(path: Path, headers: list[str], rows: list[list[str]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="gb18030", newline="") as fh: writer = csv.writer(fh, lineterminator="\n") writer.writerow(headers) writer.writerows(rows) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Build Yifangbao CRM import CSV from last month's template.") parser.add_argument("--template", type=Path, default=DEFAULT_TEMPLATE) parser.add_argument("--input", type=Path, default=DEFAULT_INPUT) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) return parser.parse_args() def main() -> int: args = parse_args() headers = load_template(args.template) rows, stats = build_rows(args.input, headers) write_csv(args.output, headers, rows) print({"output": str(args.output), **stats}) return 0 if __name__ == "__main__": raise SystemExit(main())