#!/usr/bin/env python3 """Prepare a privacy-safe smart-canteen capability base from frozen local evidence.""" from __future__ import annotations import csv import json import re import subprocess import tempfile from collections import defaultdict from pathlib import Path from openpyxl import load_workbook ROOT = Path(__file__).resolve().parents[2] TASK_DIR = Path(__file__).resolve().parent SOURCE_XLS = Path( "/Users/jack/Library/Containers/com.tencent.WeWorkMac/Data/Documents/Profiles/" "A74DF84611CC2ED79C4C206E40B5D80E/Caches/Files/2026-08/" "16ae54553ed6e6afff698b9c98bb52d8/更新-2026年数科事业部-0701-数据分析.xls" ) OPS_DIR = ROOT / "work_store/data-queries/smart-canteen-28-project-usage-20260804" PORTFOLIO_DIR = ROOT / "work/2026-08-04-smart-canteen-project-portfolio-analysis" SOFFICE = Path( "/Users/jack/.cache/codex-runtimes/codex-primary-runtime/" "dependencies/bin/override/soffice" ) SALES_PROJECT_PATTERNS = [ ("江苏省国信集团", "P010", "A-名称直接映射"), ("滨州健康科技职业学院", "P014", "A-名称直接映射"), ("江西省人民政府对外联络办公室", "P011", "B-项目别名映射"), ("南京莱迪森", "P015", "B-项目别名映射"), ("样品出库（北京首通慧城", "P020", "B-样品归属映射"), ("北京首通慧城", "P020", "B-名称近似映射"), ("中央网信办秘书局", "P023", "B-项目别名映射"), ("北京九韶信科", "P009", "B-既有项目清单证明为首都机场"), ] CONTEXT_PROJECT_MAP = { "CTX-01": "P016", "CTX-02": "P012", "CTX-03": "P008", "CTX-04": "P024", "CTX-08": "P011", "CTX-09": "P023", "CTX-10": "P009", "CTX-11": "P010", } def read_csv(path: Path) -> list[dict[str, str]]: with path.open(encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def write_csv(path: Path, rows: list[dict], headers: list[str] | None = None) -> None: if not rows and not headers: raise ValueError(f"Cannot infer headers for empty CSV: {path}") fieldnames = headers or list(rows[0]) with path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def number(value): if value in (None, ""): return None try: return float(value) except (TypeError, ValueError): return None def joined(values) -> str: return "、".join(sorted({str(v).strip() for v in values if str(v).strip()})) def hardware_group(item: str) -> str: text = item or "" rules = [ ("前厅交易与身份设备", r"称重|收银|消费机|结算台|绑盘|发卡|读卡|IC卡|取餐柜|售货机|自助终端"), ("视频与AI分析", r"摄像|录像|NVR|AI分析盒|分析主机|硬盘录像"), ("后厨食安与物联网", r"晨检|留样|检测仪|中心温度|温湿度|水浸|烟感|燃气|验收秤|验货|手持机|溯源"), ("计算、网络与存储", r"服务器|交换机|路由|网关|机柜|硬盘|控制主机|工作站|UPS|电源"), ("营养与信息展示", r"电子价签|显示屏|广告机|信息盒|LED|电视|触摸屏|音响|营养健康一体机"), ("餐线与配套设备", r"保温|餐炉|餐台|打印机|叉车秤|餐盘|托盘"), ] for group, pattern in rules: if re.search(pattern, text, re.I): return group return "其他硬件与工程材料" def quantity_grain(item: str, qty=None) -> str: text = item or "" numeric_qty = number(qty) if numeric_qty is not None and not float(numeric_qty).is_integer(): return "工程量/材料（不并入设备台数）" if re.search(r"餐盘|托盘|IC卡", text, re.I): return "批量载体（不并入设备台数）" if re.search(r"光缆|光纤|网线|电源线|线缆|屏体|镀膜|结构|边框|辅材|耗材|施工|安装", text, re.I): return "工程量/材料（不并入设备台数）" return "可计数设备观察值" def map_sales_customer(customer: str) -> tuple[str, str]: for pattern, project_id, status in SALES_PROJECT_PATTERNS: if pattern in customer: return project_id, status return "", "未映射：销售对象不等同于已登记项目" def convert_source() -> Path: if not SOURCE_XLS.exists(): raise FileNotFoundError(SOURCE_XLS) temp_dir = Path(tempfile.mkdtemp(prefix="smart-canteen-base-")) subprocess.run( [str(SOFFICE), "--headless", "--convert-to", "xlsx", "--outdir", str(temp_dir), str(SOURCE_XLS)], check=True, capture_output=True, text=True, ) converted = temp_dir / f"{SOURCE_XLS.stem}.xlsx" if not converted.exists(): raise RuntimeError("XLS conversion did not produce an XLSX file") return converted def load_sales_rows() -> list[dict]: converted = convert_source() wb = load_workbook(converted, data_only=True, read_only=True) ws = wb["基表"] raw_headers = [cell.value for cell in next(ws.iter_rows(min_row=1, max_row=1))] headers = [str(h).strip() if h is not None else f"_col_{i}" for i, h in enumerate(raw_headers, 1)] rows = [] for values in ws.iter_rows(min_row=2, values_only=True): row = dict(zip(headers, values)) if str(row.get("所属业务中心") or "").strip() != "团餐业务中心": continue customer = str(row.get("客户名称") or "").strip() item = str(row.get("存货名称") or "").strip() primary = str(row.get("数科产品一级分类") or "").strip() project_id, mapping = map_sales_customer(customer) is_hardware = primary.startswith("1智能硬件") group = hardware_group(item) if is_hardware else "" grain = quantity_grain(item, row.get("数量")) if is_hardware else "" rows.append( { "销售对象": customer, "映射项目ID": project_id, "映射状态": mapping, "开票月份": str(row.get("月份") or ""), "省份": str(row.get("所属省份（单据头）") or ""), "存货编码": str(row.get("存货编码") or ""), "存货名称": item, "规格型号": str(row.get("规格型号") or ""), "一级分类": primary, "二级分类": str(row.get("数科产品二级分类") or ""), "功能细分": str(row.get("数科产品功能细分") or ""), "数量": number(row.get("数量")), "价税合计": number(row.get("价税合计")) or 0, "成本总价": number(row.get("成本总价")) or 0, "供应商名称": str(row.get("供应商名称") or ""), "记录类型": "硬件" if is_hardware else ("软件/服务" if primary else "未分类"), "硬件大类": group, "数量口径": grain, "接口关键词": item if re.search(r"接口|对接", item) else "", } ) wb.close() return rows def main() -> None: projects = read_csv(OPS_DIR / "project-master.csv") ops = {r["project_key"]: r for r in read_csv(OPS_DIR / "db-core-refresh/all-project-dau.csv")} sales = load_sales_rows() portfolio_items = read_csv(PORTFOLIO_DIR / "line_items_enriched.csv") module_evidence = read_csv(PORTFOLIO_DIR / "software_module_evidence.csv") project_sales = defaultdict(list) for row in sales: if row["映射项目ID"]: project_sales[row["映射项目ID"]].append(row) project_portfolio = defaultdict(list) for row in portfolio_items: pid = CONTEXT_PROJECT_MAP.get(row["project_context_id"]) if pid: project_portfolio[pid].append(row) project_modules = defaultdict(set) for row in module_evidence: pid = CONTEXT_PROJECT_MAP.get(row["project_context_id"]) if pid: project_modules[pid].add(row["module_name"]) project_rows = [] for project in projects: pid = project["project_id"] op = ops.get(project.get("central_project_code", ""), {}) srows = project_sales[pid] prows = project_portfolio[pid] hw_rows = [r for r in prows if r["record_type"] == "hardware"] current_hw = [r for r in srows if r["记录类型"] == "硬件"] device_qty = sum((number(r["quantity"]) or 0) for r in hw_rows) current_device_qty = sum( (r["数量"] or 0) for r in current_hw if r["数量口径"] == "可计数设备观察值" ) weighing_qty = sum( (number(r["quantity"]) or 0) for r in hw_rows if re.search(r"称重|收银秤|验收秤|叉车秤", r["item_name"], re.I) ) interface_items = [ r["接口关键词"] for r in srows if r["接口关键词"] and r["接口关键词"] != "外部接口" ] one_card = "" finance = "" other_interface = joined(interface_items) interface_status = "未在本轮证据中发现" consumption = "待形成项目启用/验收矩阵" if pid == "P014": one_card = "人员/组织/卡/余额同步；订单与退款回传（B级项目/代码证据，待验收）" finance = "进销存财务系统对接费（2026销售明细）" interface_status = "A/B-销售明细 + 项目代码证据" elif pid == "P010": interface_status = "B-存在第三方/软件对接与代码口径，待生产验收" consumption = "本人餐/亲友餐/招待餐/外卖/普通消费机五类口径（代码与项目规则，待生产验收）" elif pid == "P009": consumption = "线上商城对接、消费策略定制开发（B级项目清单证据）" interface_status = "B-既有项目清单证据" high_volume = "" if op.get("data_status") == "live" and number(op.get("valid_orders")): high_volume = ( f"生产只读：累计有效订单 {int(float(op['valid_orders'])):,}；" f"30日峰值DAU {int(float(op['dau30_peak'])):,}" ) revenue = sum(r["价税合计"] for r in srows) cost = sum(r["成本总价"] for r in srows) project_rows.append( { "项目ID": pid, "项目名称": project["project_name"], "项目别名": project["aliases"], "纳入来源": project["inclusion_source"], "项目状态": project["inclusion_state"], "数据库状态": project["database_state"], "运营数据边界": project["usage_analysis_boundary"], "运营快照时间": op.get("snapshot_at", ""), "注册用户数": number(op.get("registered_users")), "最近有单日日活": number(op.get("latest_day_dau")), "7日日均DAU": number(op.get("dau7_avg")), "30日日均DAU": number(op.get("dau30_avg")), "30日峰值DAU": number(op.get("dau30_peak")), "30日活跃用户": number(op.get("mau30")), "累计订单量": number(op.get("all_orders")), "累计有效订单量": number(op.get("valid_orders")), "累计订单金额": number(op.get("all_amount")), "累计有效订单金额": number(op.get("valid_amount")), "近30日订单量": number(op.get("orders30")), "近30日订单金额": number(op.get("amount30")), "规模运行证据": high_volume, "弹性伸缩": "待补项目级部署证据", "负载均衡": "待补项目级部署证据", "动静分离": "待补项目级部署证据", "冷热分离": "待补项目级部署证据", "国产CPU": "待核验", "国产操作系统": "待核验", "国产数据库": "待核验", "国产中间件": "待核验", "国产化证据状态": "当前来源未提供项目级适配清单/验收报告", "消费策略": consumption, "一卡通接口": one_card, "财务系统接口": finance, "其他接口": other_interface, "接口证据状态": interface_status, "软件功能模块": joined(project_modules[pid]), "历史项目清单硬件大类": joined(r["hardware_group"] for r in hw_rows), "历史项目清单硬件观察数量": device_qty if hw_rows else None, "历史项目清单称重设备数量": weighing_qty if hw_rows else None, "2026销售明细硬件大类": joined(r["硬件大类"] for r in current_hw), "2026销售明细可计数设备数量": current_device_qty if current_hw else None, "2026销售价税合计": revenue if srows else None, "2026销售成本总价": cost if srows else None, "销售源映射状态": joined(r["映射状态"] for r in srows), "综合证据等级": project["evidence_level"], "主要证据缺口": ( "架构部署清单、国产化适配/验收、消费策略启用矩阵、最终硬件签收台账" ), } ) sales_objects = [] grouped_sales = defaultdict(list) for row in sales: grouped_sales[row["销售对象"]].append(row) for customer, rows in sorted(grouped_sales.items()): hardware = [r for r in rows if r["记录类型"] == "硬件"] sales_objects.append( { "销售对象": customer, "映射项目ID": rows[0]["映射项目ID"], "映射状态": rows[0]["映射状态"], "明细行数": len(rows), "价税合计": sum(r["价税合计"] for r in rows), "成本总价": sum(r["成本总价"] for r in rows), "硬件行数": len(hardware), "可计数设备观察数量": sum( (r["数量"] or 0) for r in hardware if r["数量口径"] == "可计数设备观察值" ), "硬件大类": joined(r["硬件大类"] for r in hardware), "接口/对接条目": joined(r["接口关键词"] for r in rows), "软件/功能细分": joined(r["功能细分"] for r in rows if r["功能细分"]), } ) current_hardware = [r for r in sales if r["记录类型"] == "硬件"] current_hardware_summary = [] for group in sorted({r["硬件大类"] for r in current_hardware}): rows = [r for r in current_hardware if r["硬件大类"] == group] current_hardware_summary.append( { "证据来源": "用户提供的2026销售明细", "硬件大类": group, "覆盖销售对象数": len({r["销售对象"] for r in rows}), "明细行数": len(rows), "可计数设备观察数量": sum( (r["数量"] or 0) for r in rows if r["数量口径"] == "可计数设备观察值" ), "批量载体数量": sum( (r["数量"] or 0) for r in rows if r["数量口径"].startswith("批量载体") ), "工程量/材料原表数量": sum( (r["数量"] or 0) for r in rows if r["数量口径"].startswith("工程量") ), "口径说明": "不同数量口径分列，不可直接相加", } ) historical_hardware_summary = [] for r in read_csv(PORTFOLIO_DIR / "hardware_groups.csv"): historical_hardware_summary.append( { "证据来源": "上一轮10个项目场景归一化清单", "硬件大类": r["hardware_group"], "覆盖销售对象数": r["project_coverage_count"], "明细行数": r["source_line_count"], "可计数设备观察数量": r["observed_quantity"], "批量载体数量": "", "工程量/材料原表数量": "", "口径说明": "观察值，需最终签收/验收台账去重确认", } ) hardware_summary = historical_hardware_summary + current_hardware_summary source_index = [ { "source_id": "SRC-01", "source": str(SOURCE_XLS), "type": "用户提供XLS｜团餐业务中心销售明细", "snapshot": "文件更新口径：2026-07-01附近", "evidence_level": "A-原始业务明细", "usage": "销售对象、软件/接口条目、硬件物料、数量、收入与成本", "boundary": "不是运行监控；客户标签不一定等于独立项目", }, { "source_id": "SRC-02", "source": str(OPS_DIR), "type": "28项目主档 + 生产库只读聚合快照", "snapshot": "2026-08-04 18:45 +0800", "evidence_level": "A/B-主档与只读聚合", "usage": "注册用户、DAU、累计/近30日订单与金额", "boundary": "不可达数据库留空，不解释为0", }, { "source_id": "SRC-03", "source": str(PORTFOLIO_DIR), "type": "上一轮项目软硬件归一化证据", "snapshot": "2026-08-04", "evidence_level": "B/D-项目清单与销售证据", "usage": "软件模块、硬件大类和观察数量、机场消费策略定制", "boundary": "10个客户项目场景样本，不是全量签收台账", }, ] quality = [ {"严重度": "高", "问题": "高可用架构字段缺少项目级部署证据", "影响": "交易量只能证明真实规模，不能证明弹性伸缩/负载均衡/动静或冷热分离", "建议": "补每项目部署拓扑、云资源清单、压测与故障演练记录"}, {"严重度": "高", "问题": "国产化适配缺少项目矩阵", "影响": "无法列出已验收CPU/OS/数据库/中间件组合", "建议": "按产品版本×项目补兼容报告、安装记录、验收单"}, {"严重度": "高", "问题": "数据库不可达项目不能按0使用解释", "影响": "滨州、江西206等项目运营指标为空", "建议": "恢复只读连接或导出脱敏聚合快照"}, {"严重度": "中", "问题": "销售对象与项目主档并非一一对应", "影响": "4个销售对象无法安全映射到28项目主档", "建议": "由销售/交付补项目ID与最终客户"}, {"严重度": "中", "问题": "硬件数量存在设备/餐具/工程量混合口径", "影响": "直接相加会夸大设备台数", "建议": "保留单位与数量口径，最终以签收/验收BOM为准"}, {"严重度": "中", "问题": "消费策略只有局部项目证据", "影响": "不能回答每项目实际启用了哪些策略", "建议": "导出消费策略配置并形成项目启用/验收矩阵"}, ] write_csv(TASK_DIR / "project-capability-base.csv", project_rows) write_csv(TASK_DIR / "sales-object-mapping.csv", sales_objects) write_csv(TASK_DIR / "sales-product-detail.csv", sales) write_csv(TASK_DIR / "hardware-category-summary.csv", hardware_summary) write_csv(TASK_DIR / "data-quality-findings.csv", quality) write_csv(TASK_DIR / "source-index.csv", source_index) airport = next(r for r in project_rows if r["项目ID"] == "P009") jsr = next(r for r in project_rows if r["项目ID"] == "P019") summary = { "project_count": len(project_rows), "queryable_project_count": sum(1 for r in project_rows if r["数据库状态"] == "live"), "source_sales_object_count": len(sales_objects), "confident_sales_project_mapping_count": len({r["映射项目ID"] for r in sales_objects if r["映射项目ID"]}), "historical_hardware_observed": sum(float(r["可计数设备观察数量"]) for r in historical_hardware_summary), "historical_hardware_group_count": len(historical_hardware_summary), "airport": airport, "jsr": jsr, } (TASK_DIR / "analysis-summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8" ) (TASK_DIR / "workbook-data.json").write_text( json.dumps( { "project_rows": project_rows, "sales_objects": sales_objects, "sales_detail": sales, "hardware_summary": hardware_summary, "quality": quality, "sources": source_index, "summary": summary, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) if __name__ == "__main__": main()