#!/usr/bin/env python3 from __future__ import annotations import csv import hashlib import json import re import shutil from collections import defaultdict from pathlib import Path from typing import Iterable from openpyxl import Workbook, load_workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from openpyxl.worksheet.datavalidation import DataValidation REPO = Path(__file__).resolve().parents[3] TASK_DIR = REPO / "work/2026-08-31-thirty-project-nutrition-sku-clustering" ATLAS_DIR = REPO / "modules/product/products/smart-canteen/product-solutions/2026-08-06-28-project-3d-case-atlas/data" OUTPUT_DIR = TASK_DIR / "output" USER_OUTPUT = Path("/Users/jack/同步空间/cpt/05_经营管理与会议/005_日常管理/周例会（事业部层面）/2026年/项目汇总_30项目营养结算SKU聚类与复刻版_20260831.xlsx") WORKBOOK_NAME = "智慧食堂30项目营养结算SKU聚类与1to1复刻_20260831.xlsx" SKU_ROWS = [ { "SKU编码": "NC-01", "SKU名称": "轻量消费结算版", "主餐线模式": "消费机结算", "适用场景": "单食堂、少档口、基础刷卡/刷脸/扫码结算", "档口方式": "单档口/少档口", "规模档": "轻量（以实际峰值校准）", "账户/接口": "基础账户；可选访客/发卡", "部署方式": "云端或本地单机", "标准软件功能": "组织人员；账户余额；菜品价格；消费结算；订单查询；基础对账；设备管理", "8分钟演示路径": "建人员和菜品→配置支付媒介→完成一笔消费→查订单→看日结对账", "销售必问": "人数；档口；高峰；支付媒介；访客；是否已有消费终端", "标准边界": "不含称重、AI识别、订餐、食安、进销存和新增第三方接口", "母项目ID": "P015/P020/P027", }, { "SKU编码": "NC-02", "SKU名称": "多档口消费结算版", "主餐线模式": "消费机结算", "适用场景": "园区、写字楼、多入口多档口食堂", "档口方式": "多档口", "规模档": "标准（以实际峰值校准）", "账户/接口": "员工+访客；可选银行/聚合支付", "部署方式": "云端/本地化", "标准软件功能": "多档口；组织账户；多媒介支付；访客；异常补单；档口对账；经营汇总；设备状态", "8分钟演示路径": "建档口→配置员工/访客→多入口消费→异常补单→档口对账→经营汇总", "销售必问": "食堂/档口/入口数；峰值人数；访客比例；支付与财务口径", "标准边界": "第三方支付和闸机按接口白名单另行确认", "母项目ID": "P002/P017/P020", }, { "SKU编码": "NC-03", "SKU名称": "高并发消费结算版", "主餐线模式": "消费机结算", "适用场景": "机场、大园区、大型员工食堂高峰结算", "档口方式": "多档口/多入口", "规模档": "大型（需压力测试）", "账户/接口": "多身份、多支付、外部接口", "部署方式": "本地化/集群候选", "标准软件功能": "高峰交易；多档口；账户策略；支付路由；异常补偿；实时监控；经营汇总；接口管理", "8分钟演示路径": "展示组织和档口→连续交易→异常恢复→实时监控→汇总对账→接口状态", "销售必问": "峰值TPS；高峰时长；终端数；接口；可用性；灾备；性能验收", "标准边界": "性能、高可用与灾备必须按项目压测和部署设计确认", "母项目ID": "P009", }, { "SKU编码": "NC-04", "SKU名称": "企业补贴结算版", "主餐线模式": "消费机/称重可选", "适用场景": "企业、集团、工业园、宾馆职工食堂", "档口方式": "单/多档口", "规模档": "标准/大型", "账户/接口": "现金账户+补贴账户+班次餐标", "部署方式": "云端/本地化", "标准软件功能": "组织人员；补贴发放；班次餐标；消费限额；账户优先级；订单；退款；财务对账", "8分钟演示路径": "导入员工→发补贴→配班次餐标→消费扣款→退款/异常→财务汇总", "销售必问": "补贴周期；清零/结转；班次；限额；退款；财务科目；访客", "标准边界": "资金规则和财务口径冻结后才能报价与验收", "母项目ID": "P007/P012/P016/P018/P026", }, { "SKU编码": "NC-05", "SKU名称": "央企多策略结算版", "主餐线模式": "消费机结算", "适用场景": "央国企多人群、多消费场景食堂", "档口方式": "多档口", "规模档": "标准/大型", "账户/接口": "本人/亲友/招待/现场消费/外卖策略", "部署方式": "本地化", "标准软件功能": "组织人员；多账户；本人/亲友/招待；餐次限额；消费场景；移动服务；订单；对账；权限审计", "8分钟演示路径": "建身份→配五类消费场景→本人/亲友/招待演示→订单归属→分类对账", "销售必问": "身份类别；消费场景；补贴归属；亲友/招待审批；对账；接口", "标准边界": "新增策略与接口不能从历史项目自动承诺", "母项目ID": "P010", }, { "SKU编码": "NC-06", "SKU名称": "政务本地化结算版", "主餐线模式": "消费机/称重可选", "适用场景": "政府机关、事业单位、政务园区", "档口方式": "单/多档口", "规模档": "标准/大型", "账户/接口": "组织权限；可选政务接口", "部署方式": "政务云/纯本地", "标准软件功能": "组织权限；身份账户；交易结算；消费策略；订单对账；设备监控；审计日志；备份恢复", "8分钟演示路径": "展示组织权限→消费→对账→审计→设备监控→部署和备份边界", "销售必问": "政务云/内网；账号体系；接口；等保/国产化；备份；SLA；验收", "标准边界": "等保、国产化、政务云和高可用均需项目级证据", "母项目ID": "P004/P005/P008", }, { "SKU编码": "NC-07", "SKU名称": "高校绑盘称重版", "主餐线模式": "绑盘称重自选", "适用场景": "高校、大型机构自助称重餐线", "档口方式": "多菜位/多餐线", "规模档": "大型", "账户/接口": "校园账户；接口可选", "部署方式": "纯本地", "标准软件功能": "人员账户；托盘绑定；按克称重；菜品价格；金额/营养显示；结算；订单；设备监控；餐线分析", "8分钟演示路径": "身份识别→绑盘→多菜位称重→金额/营养→结算→订单/设备状态", "销售必问": "服务人数；餐线；菜位；托盘；称重型号；峰值；营养展示；接口", "标准边界": "型号、协议、吞吐和营养展示以真机测试为准", "母项目ID": "P011/P014", }, { "SKU编码": "NC-08", "SKU名称": "医院双餐线称重版", "主餐线模式": "绑盘称重自选", "适用场景": "医院职工双餐线、高峰称重", "档口方式": "双餐线/多菜位", "规模档": "标准", "账户/接口": "福利/补贴账户；异常报警", "部署方式": "纯本地", "标准软件功能": "福利账户；绑盘；称重取餐；营养金额显示；余额/逃单报警；补餐收银；订单；对账", "8分钟演示路径": "福利账户→绑盘→双侧称重→金额/营养→报警处理→收银兜底→对账", "销售必问": "福利规则；双餐线；菜位；托盘；报警阈值；访客；补餐收银", "标准边界": "报警策略完成不等于全链路上线，需真机回归和客户验收", "母项目ID": "P021", }, { "SKU编码": "NC-09", "SKU名称": "小型双秤称重版", "主餐线模式": "绑盘称重自选", "适用场景": "宾馆、企业小型职工食堂", "档口方式": "单餐线/少菜位", "规模档": "轻量", "账户/接口": "员工补贴账户", "部署方式": "云端/本地化", "标准软件功能": "员工账户；补贴；绑盘；单/双秤称重；计价；营养显示；订单；基础对账", "8分钟演示路径": "发补贴→绑盘→单秤/双秤取餐→结算→订单→日结", "销售必问": "人数；菜位；单/双秤；托盘；补贴；峰值；部署", "标准边界": "单秤与双秤硬件协议必须按型号冻结", "母项目ID": "P016", }, { "SKU编码": "NC-10", "SKU名称": "无绑盘大规模称重版", "主餐线模式": "免绑盘称重自选", "适用场景": "大规模称重餐线、减少绑盘步骤", "档口方式": "多餐线/多菜位", "规模档": "大型", "账户/接口": "身份直连；接口待定", "部署方式": "纯本地", "标准软件功能": "身份识别；免绑盘称重；金额/营养显示；自动结算；异常兜底；设备监控；余量看板；数据大屏", "8分钟演示路径": "身份识别→免绑盘取餐→金额/营养→结算→异常兜底→余量/数据大屏", "销售必问": "身份方式；56台级规模；菜位；并发；设备协议；异常扣款；看板", "标准边界": "免绑盘、人脸兜底和设备兼容必须现场验证", "母项目ID": "P028", }, { "SKU编码": "NC-11", "SKU名称": "AI单通道识别结算版", "主餐线模式": "AI视觉识别结算", "适用场景": "标准菜品、小碗菜、制造企业餐线", "档口方式": "单通道", "规模档": "轻量/标准", "账户/接口": "员工账户；支付可选", "部署方式": "纯本地", "标准软件功能": "菜品库；AI识别；人工纠错；计价；支付；订单；识别日志；设备状态；经营分析", "8分钟演示路径": "建菜品→摆餐→AI识别→人工纠错→结算→识别/经营复盘", "销售必问": "菜品标准化；餐具；准确率目标；峰值；人工兜底；模型服务；设备", "标准边界": "识别准确率和适配范围必须以真实菜品测试为准", "母项目ID": "P022", }, { "SKU编码": "NC-12", "SKU名称": "一卡通称重版", "主餐线模式": "绑盘称重自选", "适用场景": "高校/机构已有一卡通体系", "档口方式": "多菜位/多餐线", "规模档": "大型", "账户/接口": "一卡通+访客账户", "部署方式": "纯本地", "标准软件功能": "一卡通身份；访客账户；绑盘；称重计价；余额/异常；订单；退款；接口对账；设备监控", "8分钟演示路径": "一卡通/访客识别→绑盘→称重→异常策略→结算→接口对账", "销售必问": "一卡通协议；测试环境；账户/退款；访客；餐线；设备；联调责任人", "标准边界": "接口文档、测试环境、联调和退款规则未冻结前不得承诺上线", "母项目ID": "P014", }, { "SKU编码": "NC-13", "SKU名称": "移动订餐基础版", "主餐线模式": "移动订餐取餐", "适用场景": "企业员工预订餐、自提", "档口方式": "单/多档口", "规模档": "标准", "账户/接口": "小程序账户+支付", "部署方式": "云端/本地化", "标准软件功能": "小程序登录；菜单；库存；下单；支付；订单；退款；取餐码；运营统计", "8分钟演示路径": "登录→选餐下单→支付→后台接单→出餐→取餐核销→退款/统计", "销售必问": "堂食/自提；支付；菜单库存；退款；峰值订单；消息通知", "标准边界": "不默认包含打印机、取餐柜或配送接口", "母项目ID": "P019", }, { "SKU编码": "NC-14", "SKU名称": "订餐厨房打印版", "主餐线模式": "移动订餐取餐", "适用场景": "预订餐+后厨分单打印", "档口方式": "多档口", "规模档": "标准/大型", "账户/接口": "小程序账户+支付+打印", "部署方式": "云端/本地化", "标准软件功能": "移动订餐；支付；厨房分单；档口打印；补打；出餐状态；核销；退款；订单分析", "8分钟演示路径": "下单→支付→厨房分单打印→补打→出餐→核销→退款/分析", "销售必问": "厨房/档口；打印机数量；分单规则；网络；补打；高峰订单", "标准边界": "打印机型号、驱动、网络与补打规则按白名单验收", "母项目ID": "P019", }, { "SKU编码": "NC-15", "SKU名称": "订餐智能取餐柜版", "主餐线模式": "移动订餐取餐", "适用场景": "预订餐+智能取餐柜自提", "档口方式": "集中出餐", "规模档": "标准", "账户/接口": "小程序账户+支付+柜机接口", "部署方式": "云端/本地化", "标准软件功能": "移动订餐；支付；出餐；装柜；格口分配；取餐码；开柜核销；超时/异常；退款；柜机状态", "8分钟演示路径": "下单→厨房出餐→装柜→通知→扫码开柜→异常格口→订单复盘", "销售必问": "柜机型号；格口；接口；网络；超时；温控；异常开柜；维保", "标准边界": "柜机协议和异常恢复必须按真实型号联调", "母项目ID": "P019", }, { "SKU编码": "NC-16", "SKU名称": "校园电子菜牌版", "主餐线模式": "信息发布辅助结算", "适用场景": "中小学菜单、价格、营养公示", "档口方式": "单/多档口", "规模档": "按屏幕数", "账户/接口": "不含支付；可接菜谱数据", "部署方式": "云端", "标准软件功能": "菜谱；餐次；图片；价格；营养标签；内容审核；模板布局；多屏编组；发布；屏幕状态；开机自启", "8分钟演示路径": "建菜谱→选布局→PC预览→审核→发布到屏→查看状态→异常重发", "销售必问": "校区；档口；屏幕数/型号/分辨率；布局；审核；更新频率；网络", "标准边界": "本SKU不包含账户、支付、称重或AI结算", "母项目ID": "P003/P015/P027", }, { "SKU编码": "NC-17", "SKU名称": "营养信息发布版", "主餐线模式": "信息发布辅助结算", "适用场景": "政企/校园菜品营养公示", "档口方式": "单/多档口", "规模档": "按屏幕/电子价签数", "账户/接口": "菜谱/营养数据接口", "部署方式": "云端/本地化", "标准软件功能": "菜品营养；价格；过敏原/标签；菜单编排；审核；多终端发布；电子价签；发布日志；设备状态", "8分钟演示路径": "维护菜品营养→编排菜单→审核→发布到价签/屏→查看日志和状态", "销售必问": "营养字段；数据来源；屏/价签；审核；更新频率；接口；专业审核", "标准边界": "营养信息展示不等于医学诊断或干预效果", "母项目ID": "P003/P008", }, { "SKU编码": "NC-18", "SKU名称": "营养数据大屏版", "主餐线模式": "数据展示辅助结算", "适用场景": "大型食堂餐线和管理驾驶舱", "档口方式": "多餐线", "规模档": "大型", "账户/接口": "交易/称重数据接入", "部署方式": "本地化", "标准软件功能": "餐线交易；菜品余量；营养汇总；设备在线；高峰趋势；经营指标；轮播展示；权限；数据刷新", "8分钟演示路径": "完成称重交易→查看余量→查看营养/经营→设备在线→高峰趋势→大屏轮播", "销售必问": "指标口径；数据源；刷新频率；屏幕数/尺寸；权限；展示场地", "标准边界": "指标口径和屏幕硬件必须随项目冻结", "母项目ID": "P028", }, { "SKU编码": "NC-19", "SKU名称": "营养健康管理增强版", "主餐线模式": "营养服务增强", "适用场景": "体育院校、运动队、企业健康干预", "档口方式": "与主结算SKU组合", "规模档": "按服务人数", "账户/接口": "健康档案+餐食数据", "部署方式": "云端/本地化", "标准软件功能": "健康档案；营养评估；人群分层；食谱/训练餐；餐食记录；营养建议；随访；个人/群体看板", "8分钟演示路径": "建档案→评估→分层→生成食谱/训练餐→关联餐食→随访→看板", "销售必问": "服务对象；评估量表；营养师；数据授权；周期；隐私；专业审核", "标准边界": "不做医疗诊断和疗效承诺；需专业人员与隐私授权", "母项目ID": "P006/P013", }, { "SKU编码": "NC-20", "SKU名称": "内网保障供餐版", "主餐线模式": "集中保障结算", "适用场景": "部队、封闭机构、内网/离线供餐", "档口方式": "集中供餐", "规模档": "按编制与保障批次", "账户/接口": "编制账户+餐标策略", "部署方式": "纯本地/内网", "标准软件功能": "编制人员；餐标；批次供餐；身份核验；内网/离线交易；回传对账；权限审计；保障看板", "8分钟演示路径": "导入编制→配餐标→离线核销→恢复回传→差异处理→保障看板→审计", "销售必问": "内网；离线时长；编制；餐标；回传；终端白名单；安全审计；验收", "标准边界": "当前为储备版；安全等级、离线一致性和终端白名单须专项验证", "母项目ID": "P025", }, ] PROJECT_SKU_MAP = { "P001": ["OUT-NC"], "P002": ["NC-02"], "P003": ["NC-16", "NC-17"], "P004": ["NC-06"], "P005": ["NC-06"], "P006": ["NC-19"], "P007": ["NC-04"], "P008": ["NC-06", "NC-17"], "P009": ["NC-03"], "P010": ["NC-05"], "P011": ["NC-07"], "P012": ["NC-04"], "P013": ["NC-19"], "P014": ["NC-07", "NC-12"], "P015": ["NC-01", "NC-16"], "P016": ["NC-09", "NC-04"], "P017": ["NC-02"], "P018": ["NC-04"], "P019": ["NC-13", "NC-14", "NC-15"], "P020": ["NC-01", "NC-02"], "P021": ["NC-08"], "P022": ["NC-11"], "P023": ["OUT-NC"], "P024-A": ["OUT-NC"], "P024-B": ["OUT-NC"], "P024-C": ["OUT-NC"], "P025": ["NC-20"], "P026": ["NC-04"], "P027": ["NC-01", "NC-16"], "P028": ["NC-10", "NC-18"], } SKU_META = { "NC-01": ("C-消费机结算", "POS-LITE", "管理端×1；消费终端×实际结算点；发卡器×1（可选）；网络按点位"), "NC-02": ("C-消费机结算", "POS-MULTI", "管理端×1；消费终端×档口/入口数；发卡器×1（可选）；支付设备/网络按点位"), "NC-03": ("C-消费机结算", "POS-HIGH", "消费终端×档口/入口数；应用/数据库服务器按并发方案；监控与网络按点位"), "NC-04": ("C-消费机结算", "POS-SUBSIDY", "管理端×1；消费终端×档口数；发卡器×1（可选）；补贴账户无专属硬件"), "NC-05": ("C-消费机结算", "POS-STRATEGY", "管理端×1；消费终端×档口数；移动端按账号；接口网关按实际系统"), "NC-06": ("C-消费机结算", "GOV-LOCAL", "消费/称重终端按餐线；本地服务器按规模；网络与备份按政务边界"), "NC-07": ("W-称重结算", "W-450-CAMPUS", "绑盘终端×餐线入口；单秤智能台×菜位；布菲炉×菜位；托盘×服务规模；收银兜底按餐线"), "NC-08": ("W-称重结算", "W-450-HOSP", "绑盘终端×入口；单秤智能台×双餐线菜位；双面收银机×收银点；称重收银机×兜底点"), "NC-09": ("W-称重结算", "W-450D-SMALL", "绑盘终端×1；单秤/双秤智能台按菜位；托盘按服务人数；管理端×1"), "NC-10": ("W-称重结算", "W-460-NOBIND", "免绑盘智能台×菜位；布菲炉×菜位；身份终端按入口；交换机/看板/大屏按点位"), "NC-11": ("A-AI识别结算", "AI-S30A-1CH", "AI单通道识别结算台×通道数；双面收银机×兜底点；应用/数据库服务器按部署"), "NC-12": ("W-称重结算", "W-450-CARD", "NC-07称重BOM + 一卡通读卡/接口环境；访客终端按入口"), "NC-13": ("M-移动订餐", "M-BASE", "移动端按账号；管理端×1；不默认新增打印机/柜机"), "NC-14": ("M-移动订餐", "M-PRINT", "NC-13软件 + 后厨小票打印机×厨房/档口；网络按打印点"), "NC-15": ("M-移动订餐", "M-LOCKER", "NC-13软件 + 智能取餐柜×取餐点；柜机控制与网络按点位"), "NC-16": ("D-信息发布", "DISP-TV", "管理PC×1；电子菜牌/电视×屏幕点位；服务器/网络按部署；PoE按终端类型"), "NC-17": ("D-信息发布", "DISP-ESL", "管理端×1；电子价签/营养屏×展示点；网关/网络按终端协议"), "NC-18": ("D-信息发布", "DISP-DASH", "数据大屏×展示点；余量看板×餐线；播放/控制终端和网络按点位"), "NC-19": ("H-营养健康", "HEALTH-SOFT", "纯软件增强包；移动端/PC按角色；测评设备不在默认BOM"), "NC-20": ("I-内网保障", "INTRA-OFFLINE", "内网服务器×方案；身份/核销终端×入口；本地网络、备份和离线介质按安全边界"), } P024_SITES = [ ("P024-A", "北京市大兴区旧宫中学", "旧宫中学"), ("P024-B", "北京市大兴区采育镇第三中心小学", "采育三小"), ("P024-C", "兴海学校", "兴海学校（历史材料亦见海兴/兴海中学，名称待项目负责人确认）"), ] FUNCTION_DESCRIPTIONS = { "组织人员": "维护组织、人员、角色和批量导入，为身份、账户和权限提供基础。", "账户余额": "维护现金、补贴等账户余额与流水。", "菜品价格": "维护菜品、菜单、档口、售价及基础营养字段。", "消费结算": "完成刷卡、刷脸、扫码或配置媒介的消费扣款。", "订单查询": "按人员、档口、时间和状态查询交易记录。", "基础对账": "汇总交易、退款、补贴和支付数据，支持日结核对。", "设备管理": "登记终端并查看基础在线、版本和故障状态。", "绑盘": "把人员身份与托盘/餐盘建立、解除和异常恢复关系。", "称重取餐": "按菜品单价和重量自动计算金额并记录取餐。", "营养金额显示": "在取餐过程中展示金额和可用的营养信息。", "AI识别": "识别标准菜品并输出候选菜品和价格。", "人工纠错": "在识别不确定或错误时由操作员修正。", "移动订餐": "在小程序完成菜单查看、下单和订单管理。", "厨房分单": "按厨房、档口或菜品把订单分发到制作端。", "取餐核销": "通过取餐码、扫码或柜机完成订单核销。", "营养评估": "基于授权档案和规则形成营养评估结果。", "营养建议": "在专业边界内输出餐食建议和随访信息。", "多屏编组": "按地点、档口和屏幕组管理发布对象。", "内容审核": "在菜单或营养内容发布前完成审批。", "设备监控": "查看餐线设备、屏幕或柜机的运行状态。", } def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def write_csv(path: Path, rows: list[dict[str, object]], fields: Iterable[str] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) if fields is None: fields = list(rows[0].keys()) if rows else [] with path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(fields), extrasaction="ignore") writer.writeheader() writer.writerows(rows) def is_number(value: str) -> bool: return bool(re.fullmatch(r"\d+(?:\.\d+)?", (value or "").strip())) def clean_quantity(value: str) -> str: if not value: return "待补证" if is_number(value): number = float(value) return str(int(number)) if number.is_integer() else str(number) return value def split_quantity(value: str, divisor: int = 3) -> str: if not is_number(value): return value or "待补证" number = float(value) / divisor return str(int(number)) if number.is_integer() else f"{number:g}（三校均分假设）" def base_projects() -> list[dict[str, str]]: rows = read_csv(ATLAS_DIR / "project-case-base.csv") result = [] for row in rows: if row["项目ID"] != "P024": item = dict(row) item.update({ "项目单元ID": row["项目ID"], "母项目ID": row["项目ID"], "项目单元类型": "独立项目", "名称校验备注": "", }) result.append(item) continue for unit_id, name, note in P024_SITES: item = dict(row) item.update({ "项目单元ID": unit_id, "母项目ID": "P024", "项目ID": unit_id, "项目名称": name, "简称": name.replace("北京市大兴区", ""), "项目单元类型": "大兴三校母项目下的项目点位", "名称校验备注": note, "一句话场景": "单校区食安/收货节点，归属大兴三校母项目", "项目特点": "由大兴三校合并项目拆出的点位单元；不是新增合同项目", "验证能力/证据边界": "三校总BOM可核验；按每校一套主链路拆分，具体点位仍需签收/SN台账确认", "设备观察口径": "大兴三校总BOM按点位拆分，不代表三份独立合同", }) result.append(item) return result def build_software_rows(projects: list[dict[str, str]]) -> list[dict[str, str]]: source = read_csv(ATLAS_DIR / "software-detail.csv") product_catalog = read_csv(ATLAS_DIR / "product-image-catalog-P020-P028.csv") supplements = [] for row in product_catalog: name = row["hardware_or_software"] if not any(token in name for token in ["软件", "系统", "平台", "小程序", "充值及扣款流程"]): continue nature = "项目事实" if not row["mapping_status"].startswith("D-") else "方案/缺口" supplements.append({ "项目ID": row["project_id"], "项目名称": row["project_name"], "软件模块": name, "事实/建议": nature, "数量": row["quantity"], "单位": row["unit"], "版本/型号": row["model_or_alias"], "映射状态": row["mapping_status"], "证据等级": row["mapping_status"][:1] if row["mapping_status"] else "D", "证据路径": row["project_quantity_source"], }) grouped: dict[str, list[dict[str, str]]] = defaultdict(list) for row in source + supplements: grouped[row["项目ID"]].append(row) result = [] for project in projects: unit_id = project["项目单元ID"] parent = project["母项目ID"] seen = set() for row in grouped[parent]: key = (row["软件模块"], row["版本/型号"], row["事实/建议"]) if key in seen: continue seen.add(key) item = dict(row) item.update({ "项目单元ID": unit_id, "母项目ID": parent, "项目名称": project["项目名称"], "性质": "场景建议" if row["事实/建议"] == "场景建议" or row["事实/建议"] == "方案/缺口" else "项目事实", "点位拆分说明": "大兴三校共享/均分假设，需点位台账确认" if parent == "P024" else "", }) result.append(item) return result def build_hardware_rows(projects: list[dict[str, str]]) -> list[dict[str, str]]: primary = read_csv(ATLAS_DIR / "hardware-detail.csv") catalog = read_csv(ATLAS_DIR / "product-image-catalog-P020-P028.csv") grouped_primary: dict[str, list[dict[str, str]]] = defaultdict(list) for row in primary: grouped_primary[row["项目ID"]].append(row) grouped_catalog: dict[str, list[dict[str, str]]] = defaultdict(list) for row in catalog: name = row["hardware_or_software"] if any(token in name for token in ["软件", "系统", "平台", "小程序", "充值及扣款流程"]): continue grouped_catalog[row["project_id"]].append(row) result = [] for project in projects: unit_id = project["项目单元ID"] parent = project["母项目ID"] rows = [] if parent == "P024": for row in grouped_catalog[parent]: rows.append({ "硬件类别": "项目BOM", "硬件名称": row["hardware_or_software"], "标准型号": row["model_or_alias"], "项目写法/型号": row["model_or_alias"], "数量原文": split_quantity(row["quantity"]), "单位": row["unit"], "事实/建议": "项目证据", "映射状态": row["mapping_status"], "证据等级": row["mapping_status"][:1] if row["mapping_status"] else "D", "证据路径": row["project_quantity_source"], "点位拆分说明": "大兴三校总量按3个点位拆分；具体点位须签收/SN台账确认", }) else: for row in grouped_primary[parent]: rows.append({ "硬件类别": row["硬件类别"], "硬件名称": row["硬件名称"], "标准型号": row["标准型号"], "项目写法/型号": row["项目写法/型号"], "数量原文": clean_quantity(row["数量原文"]), "单位": row["单位"], "事实/建议": row["事实/建议"], "映射状态": row["映射状态"], "证据等级": row["证据等级"], "证据路径": row["证据路径"], "点位拆分说明": "", }) existing = {(r["硬件名称"], r["标准型号"]) for r in rows} for row in grouped_catalog[parent]: key = (row["hardware_or_software"], row["model_or_alias"]) similar = any(row["hardware_or_software"].replace("产品族", "") in a or a in row["hardware_or_software"] for a, _ in existing) if key in existing or similar: continue rows.append({ "硬件类别": "补充项目BOM", "硬件名称": row["hardware_or_software"], "标准型号": row["model_or_alias"], "项目写法/型号": row["model_or_alias"], "数量原文": clean_quantity(row["quantity"]), "单位": row["unit"], "事实/建议": "项目证据" if row["quantity"] != "待补证" else "证据缺口", "映射状态": row["mapping_status"], "证据等级": row["mapping_status"][:1] if row["mapping_status"] else "D", "证据路径": row["project_quantity_source"], "点位拆分说明": "", }) for row in rows: item = dict(row) item.update({ "项目单元ID": unit_id, "母项目ID": parent, "项目名称": project["项目名称"], "是否可原样进入报价BOM": "是" if row["映射状态"].startswith("A-") and is_number(row["数量原文"]) else "否", }) result.append(item) return result def summarize_software(rows: list[dict[str, str]], project_id: str, actual_only: bool = True) -> str: selected = [r for r in rows if r["项目单元ID"] == project_id and (not actual_only or r["性质"] == "项目事实")] if not selected: return "待补证" return "；".join( f"{r['软件模块']}（{r['版本/型号'] or '版本待补'}，{r['证据等级']}）" for r in selected ) def summarize_hardware(rows: list[dict[str, str]], project_id: str, actual_only: bool = True) -> str: selected = [ r for r in rows if r["项目单元ID"] == project_id and (not actual_only or r["事实/建议"] in {"项目证据", "运行/清单观察"}) ] if not selected: return "待补证" return "；".join( f"{r['硬件名称']}（{r['标准型号'] or '型号待补'}，{r['数量原文']}{r['单位']}，{r['映射状态']}）" for r in selected ) def project_replication_level(project_id: str, sw_rows: list[dict[str, str]], hw_rows: list[dict[str, str]]) -> tuple[str, str]: actual_sw = [r for r in sw_rows if r["项目单元ID"] == project_id and r["性质"] == "项目事实"] actual_hw = [r for r in hw_rows if r["项目单元ID"] == project_id and r["事实/建议"] == "项目证据" and r["数量原文"] != "待补证"] if actual_sw and actual_hw: return "R2-项目校准复刻", "可用于销售演示和方案/BOM模板；签约前仍需冻结版本、点位、SN、接口和验收" if actual_sw or actual_hw: return "R1-流程/单侧复刻", "只能复刻已证实的软件流程或硬件BOM一侧，另一侧需补证" return "R0-场景建议", "只能用于需求沟通，不能作为历史交付或1:1复刻承诺" def make_project_rows(projects: list[dict[str, str]], sw_rows: list[dict[str, str]], hw_rows: list[dict[str, str]]) -> list[dict[str, str]]: sku_by_code = {r["SKU编码"]: r for r in SKU_ROWS} result = [] for project in projects: pid = project["项目单元ID"] codes = PROJECT_SKU_MAP[pid] primary = codes[0] if primary == "OUT-NC": sku_name = "不纳入本轮营养结算SKU（食安/进销存主导样本）" standard_scope = "仅保留项目事实；如需营养结算须重新确认餐线、身份、账户、结算和硬件" demo = "不安排营养结算标准演示；先完成需求资格判断" sales_ask = "是否存在前厅营养结算需求；服务人数；餐线；支付；账户；部署" boundary = "本轮只梳理营养结算，不把食安/进销存能力改写成营养结算交付事实" else: sku = sku_by_code[primary] sku_name = f"{primary} {sku['SKU名称']}" standard_scope = sku["标准软件功能"] demo = sku["8分钟演示路径"] sales_ask = sku["销售必问"] boundary = sku["标准边界"] level, level_note = project_replication_level(pid, sw_rows, hw_rows) if primary == "OUT-NC": level = "OUT-非营养结算范围" level_note = "保留原项目事实，不参与20个营养结算SKU的复刻等级计算" scale = "；".join( f"{label}{project[key]}" for label, key in [ ("注册账户", "已观察注册账户"), ("；日日活", "已观察日日活"), ("；30日活跃", "已观察30日活跃"), ("；累计订单", "累计有效订单"), ("；近30日订单", "近30日订单"), ] if project.get(key) ).lstrip("；") or "待补证" result.append({ "项目单元ID": pid, "母项目ID": project["母项目ID"], "项目单元类型": project["项目单元类型"], "项目名称": project["项目名称"], "名称校验备注": project["名称校验备注"], "场景类型": project["场景类型"], "服务人群": project["服务人群"], "规模事实": scale, "主要问题": project["主要问题"], "主SKU": sku_name, "关联SKU": "；".join(codes), "标准软件范围（产品定义）": standard_scope, "项目实际软件（事实）": summarize_software(sw_rows, pid, actual_only=True), "项目实际硬件/BOM（事实）": summarize_hardware(hw_rows, pid, actual_only=True), "前厅主流程": project["前厅主流程"], "后厨主流程": project["后厨主流程"], "8分钟演示主线": demo, "销售价值": project["项目特点"], "部署/规模/接口必配": sales_ask, "复刻等级": level, "销售可用范围": level_note, "明确边界": boundary, "证据等级": project["项目证据等级"], "证据边界": project["验证能力/证据边界"], "主要缺口": project["主要证据缺口"], "来源": "28项目证据册；项目软硬件明细；P020-P028项目BOM补充", }) return result def make_mapping_rows(projects: list[dict[str, str]]) -> list[dict[str, str]]: names = {p["项目单元ID"]: p["项目名称"] for p in projects} sku_names = {r["SKU编码"]: r["SKU名称"] for r in SKU_ROWS} result = [] for pid, codes in PROJECT_SKU_MAP.items(): for index, code in enumerate(codes, start=1): result.append({ "项目单元ID": pid, "项目名称": names[pid], "映射顺序": str(index), "SKU编码": code, "SKU名称": sku_names.get(code, "非营养结算范围"), "映射角色": "主SKU" if index == 1 else "关联/增购SKU", "说明": "项目事实不足时仅作聚类参考，不等于已交付", }) return result def make_sku_rows(project_rows: list[dict[str, str]], sw_rows: list[dict[str, str]], hw_rows: list[dict[str, str]]) -> list[dict[str, str]]: projects_by_id = {r["项目单元ID"]: r for r in project_rows} result = [] rank = {"R0-场景建议": 0, "R1-流程/单侧复刻": 1, "R2-项目校准复刻": 2} for sku in SKU_ROWS: anchor_ids = sku["母项目ID"].split("/") eligible = [projects_by_id[pid] for pid in anchor_ids if pid in projects_by_id] best = max(eligible, key=lambda r: rank.get(r["复刻等级"], -1)) if eligible else None current_level = best["复刻等级"] if best else "R0-场景建议" actual_sw = "；".join( f"{pid}:{summarize_software(sw_rows, pid, True)}" for pid in anchor_ids if pid in projects_by_id and summarize_software(sw_rows, pid, True) != "待补证" ) or "待补证" actual_hw = "；".join( f"{pid}:{summarize_hardware(hw_rows, pid, True)}" for pid in anchor_ids if pid in projects_by_id and summarize_hardware(hw_rows, pid, True) != "待补证" ) or "待补证" item = dict(sku) series, bom_version, standard_bom = SKU_META[sku["SKU编码"]] item.update({ "SKU系列（互斥主类）": series, "硬件BOM版本": bom_version, "标准硬件BOM（销售配置模板）": standard_bom, "SKU唯一规格键": "|".join([ series, sku["主餐线模式"], sku["档口方式"], sku["规模档"], sku["账户/接口"], sku["部署方式"], bom_version, ]), "覆盖项目单元": "；".join(r["项目单元ID"] for r in project_rows if sku["SKU编码"] in r["关联SKU"].split("；")), "当前复刻等级": current_level, "是否可直接1:1签约复制": "否", "项目实际软件证据": actual_sw, "项目实际硬件证据": actual_hw, "升级R3还缺什么": "冻结软件版本/授权；硬件型号、数量、点位、SN；部署与接口；真机全流程；性能/异常恢复；签收和验收", "销售使用说明": "可按当前等级演示和选型；只有R3才允许不经二次定标直接复制报价/BOM/验收", }) result.append(item) return result def make_replication_rows(sku_rows: list[dict[str, str]]) -> list[dict[str, str]]: result = [] for row in sku_rows: result.append({ "SKU编码": row["SKU编码"], "SKU名称": row["SKU名称"], "推荐母项目": row["母项目ID"], "当前复刻等级": row["当前复刻等级"], "销售现在能复制什么": "SKU定位、标准功能、8分钟演示路径、售前问题、已有项目软件/硬件证据", "软件1:1内容": row["项目实际软件证据"], "硬件1:1内容": row["项目实际硬件证据"], "部署与接口": f"{row['部署方式']}；{row['账户/接口']}", "演示流程": row["8分钟演示路径"], "验收要点": "身份/账户→主交易→异常恢复→订单/对账→设备状态→性能与连续运行→签收/验收", "不可直接复制的内容": row["升级R3还缺什么"], "直接1:1签约状态": "未达R3，不可直接原样签约", "下一责任人": "产品负责人+项目交付负责人+硬件产品+研发/测试+商务", }) return result def make_sku_function_rows() -> list[dict[str, str]]: result = [] for sku in SKU_ROWS: for order, raw in enumerate(sku["标准软件功能"].split("；"), start=1): name = raw.strip() result.append({ "SKU编码": sku["SKU编码"], "SKU名称": sku["SKU名称"], "功能序号": str(order), "软件功能": name, "销售解释": FUNCTION_DESCRIPTIONS.get(name, f"完成“{name}”相关业务操作，最终页面、字段和规则以产品验收基线为准。"), "功能性质": "SKU标准功能定义", "证据要求": "进入R3前必须关联真实页面/接口/数据/自动化测试和母项目验收证据", }) return result def make_gap_rows(project_rows: list[dict[str, str]], sku_rows: list[dict[str, str]]) -> list[dict[str, str]]: result = [] for row in project_rows: if row["复刻等级"].startswith("R3") or row["复刻等级"].startswith("OUT"): continue result.append({ "对象类型": "项目", "对象ID": row["项目单元ID"], "对象名称": row["项目名称"], "当前等级": row["复刻等级"], "缺口": row["主要缺口"], "影响": "不能作为合同级1:1复制承诺", "补证动作": "软件版本/页面/接口/测试与硬件BOM/SN/签收/验收逐项勾稽", }) for row in sku_rows: result.append({ "对象类型": "SKU", "对象ID": row["SKU编码"], "对象名称": row["SKU名称"], "当前等级": row["当前复刻等级"], "缺口": row["升级R3还缺什么"], "影响": "销售可演示/选型，但不可不经二次定标直接复制签约", "补证动作": "选择一个母项目完成R3六证闭环，再冻结为正式销售BOM", }) return result def source_rows() -> list[dict[str, str]]: sources = [ ("项目主档", ATLAS_DIR / "project-case-base.csv", "28个权威项目主档；P024为大兴三校合并项目"), ("项目软件", ATLAS_DIR / "software-detail.csv", "项目软件事实与场景建议分栏"), ("项目硬件", ATLAS_DIR / "hardware-detail.csv", "硬件型号、数量、映射状态和证据路径"), ("P020-P028补充BOM", ATLAS_DIR / "product-image-catalog-P020-P028.csv", "补充软件、硬件、型号、数量和映射状态"), ("P020-P028照片证据", ATLAS_DIR / "photo-evidence-P020-P028.csv", "1:1状态和不可复原原因"), ("四重点项目补充", Path("/Users/jack/同步空间/cpt/05_经营管理与会议/005_日常管理/周例会（事业部层面）/2026年/项目汇总_营养结算销售产品补充版_20260831.xlsx"), "滨州、国康、城市副中心、人大附中销售产品化补充"), ] result = [] for source_type, path, usage in sources: result.append({ "来源类型": source_type, "来源路径": str(path), "存在": "是" if path.exists() else "否", "SHA256": hashlib.sha256(path.read_bytes()).hexdigest() if path.exists() else "", "用途/边界": usage, }) return result HEADER_FILL = PatternFill("solid", fgColor="163A5F") SUB_FILL = PatternFill("solid", fgColor="DCEAF5") WARN_FILL = PatternFill("solid", fgColor="FFF1CC") BAD_FILL = PatternFill("solid", fgColor="FDE2E2") GOOD_FILL = PatternFill("solid", fgColor="E2F0D9") WHITE_FONT = Font(name="微软雅黑", size=11, color="FFFFFF", bold=True) BODY_FONT = Font(name="微软雅黑", size=11, color="1F2937") TITLE_FONT = Font(name="微软雅黑", size=18, color="163A5F", bold=True) THIN = Side(style="thin", color="CBD5E1") def add_table_sheet(wb: Workbook, name: str, rows: list[dict[str, object]], freeze: str = "A2") -> None: ws = wb.create_sheet(name) if not rows: ws.append(["无数据"]) return headers = list(rows[0].keys()) ws.append(headers) for row in rows: ws.append([row.get(h, "") for h in headers]) ws.freeze_panes = freeze ws.auto_filter.ref = ws.dimensions ws.sheet_view.showGridLines = False for cell in ws[1]: cell.fill = HEADER_FILL cell.font = WHITE_FONT cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) cell.border = Border(bottom=THIN) ws.row_dimensions[1].height = 34 for row in ws.iter_rows(min_row=2): for cell in row: cell.font = BODY_FONT cell.alignment = Alignment(vertical="top", wrap_text=True) cell.border = Border(bottom=THIN) for idx, header in enumerate(headers, start=1): values = [str(ws.cell(r, idx).value or "") for r in range(1, min(ws.max_row, 100) + 1)] width = min(max(len(header) * 2 + 2, max((min(len(v), 45) for v in values), default=8) + 2), 48) if header in {"项目单元ID", "母项目ID", "SKU编码", "映射顺序", "功能序号"}: width = 14 ws.column_dimensions[get_column_letter(idx)].width = width for row_idx in range(2, ws.max_row + 1): status_text = " ".join(str(ws.cell(row_idx, col).value or "") for col in range(1, min(ws.max_column, 8) + 1)) if "R2-" in status_text: for cell in ws[row_idx]: cell.fill = GOOD_FILL elif "R1-" in status_text: for cell in ws[row_idx]: cell.fill = WARN_FILL elif "R0-" in status_text or "未达R3" in status_text: for cell in ws[row_idx]: cell.fill = BAD_FILL def add_readme_sheet(wb: Workbook, project_rows: list[dict[str, str]], sku_rows: list[dict[str, str]]) -> None: ws = wb.active ws.title = "先看这里" ws.sheet_view.showGridLines = False ws.merge_cells("A1:H1") ws["A1"] = "30项目营养结算产品化与20个SKU聚类" ws["A1"].font = TITLE_FONT ws["A1"].alignment = Alignment(vertical="center") ws.row_dimensions[1].height = 34 content = [ ("本表用途", "销售先从“20SKU销售表”选版本，再去“1to1复刻包”查看能复制什么；产品/交付用后续明细补齐证据。"), ("30项目口径", "28项目权威主档中，大兴三校原为1个合并项目；本表拆为旧宫、采育三小、兴海3个项目点位，因此是30个项目单元，不代表30份独立合同。"), ("SPU定义", "本轮只梳理“营养结算”SPU，不深入进销存和食安。食安/进销存主导项目仍保留事实，但标记OUT-NC。"), ("SKU定义", "营养结算SPU × 主餐线模式 × 场景版型 × 账户/接口 × 部署方式 × 规模档 × 硬件BOM版。20个为常用参考配置，不穷举所有排列组合。"), ("R0", "场景建议：只能讨论需求，不能说历史项目已交付。"), ("R1", "流程/单侧复刻：软件流程或硬件BOM只有一侧有证据，可演示已证实部分。"), ("R2", "项目校准复刻：软件和硬件两侧都有项目证据，可作为演示、方案和BOM模板；签约前仍须二次定标。"), ("R3", "合同级1:1复刻：软件版本/授权、硬件型号/数量/点位/SN、部署、接口、测试、签收和验收全部闭环后，才允许直接复制报价/BOM/验收。"), ("当前结论", f"30个项目单元已补齐；20个SKU已聚类。R3数量：{sum(r['当前复刻等级'].startswith('R3') for r in sku_rows)}；因此本版不能宣称已有可不经定标直接签约的1:1 SKU。"), ("推荐动作", "优先把NC-07高校绑盘称重、NC-08医院双餐线称重、NC-12一卡通称重、NC-13/14/15移动订餐链路中的1—2个母项目补到R3。"), ("销售使用红线", "不得把场景建议写成已交付；不得把相似硬件图片写成项目精确型号；不得用设备初始化替代软件全流程、性能或客户验收。"), ] ws.append([]) ws.append(["项目", "说明"]) for cell in ws[3]: cell.fill = HEADER_FILL cell.font = WHITE_FONT cell.alignment = Alignment(horizontal="center") for label, text in content: ws.append([label, text]) for row in ws.iter_rows(min_row=4, max_row=ws.max_row): row[0].font = Font(name="微软雅黑", size=11, bold=True, color="163A5F") row[0].fill = SUB_FILL row[1].font = BODY_FONT for cell in row: cell.alignment = Alignment(vertical="top", wrap_text=True) cell.border = Border(bottom=THIN) ws.column_dimensions["A"].width = 22 ws.column_dimensions["B"].width = 105 ws.freeze_panes = "A3" def add_validations(wb: Workbook) -> None: for sheet_name in ["20SKU销售表", "1to1复刻包", "30项目产品化"]: ws = wb[sheet_name] headers = {cell.value: cell.column for cell in ws[1]} status_header = "当前复刻等级" if "当前复刻等级" in headers else "复刻等级" if status_header not in headers: continue col = get_column_letter(headers[status_header]) dv = DataValidation(type="list", formula1='"R0-场景建议,R1-流程/单侧复刻,R2-项目校准复刻,R3-合同级1:1复刻,OUT-非营养结算范围"', allow_blank=True) ws.add_data_validation(dv) dv.add(f"{col}2:{col}{max(ws.max_row, 2)}") def build_workbook(path: Path, project_rows: list[dict[str, str]], sku_rows: list[dict[str, str]], replication_rows: list[dict[str, str]], mapping_rows: list[dict[str, str]], sku_function_rows: list[dict[str, str]], sw_rows: list[dict[str, str]], hw_rows: list[dict[str, str]], gap_rows: list[dict[str, str]], sources: list[dict[str, str]]) -> None: wb = Workbook() add_readme_sheet(wb, project_rows, sku_rows) add_table_sheet(wb, "20SKU销售表", sku_rows) add_table_sheet(wb, "1to1复刻包", replication_rows) add_table_sheet(wb, "30项目产品化", project_rows) add_table_sheet(wb, "项目SKU映射", mapping_rows) add_table_sheet(wb, "SKU软件功能", sku_function_rows) add_table_sheet(wb, "项目软件证据", sw_rows) add_table_sheet(wb, "项目硬件BOM", hw_rows) add_table_sheet(wb, "证据缺口", gap_rows) add_table_sheet(wb, "来源索引", sources) add_validations(wb) for ws in wb.worksheets: ws.sheet_properties.pageSetUpPr.fitToPage = True ws.page_setup.fitToWidth = 1 ws.page_setup.fitToHeight = 0 ws.page_margins.left = 0.25 ws.page_margins.right = 0.25 ws.page_margins.top = 0.4 ws.page_margins.bottom = 0.4 path.parent.mkdir(parents=True, exist_ok=True) wb.save(path) def validate_deliverables(data: dict[str, list[dict[str, str]]], workbook: Path) -> dict[str, object]: checks = [] def check(name: str, condition: bool, detail: str) -> None: checks.append({"check": name, "passed": bool(condition), "detail": detail}) projects = data["projects"] sku_rows = data["skus"] mapping = data["mapping"] ids = [r["项目单元ID"] for r in projects] check("30个唯一项目单元", len(ids) == 30 and len(set(ids)) == 30, f"rows={len(ids)}, unique={len(set(ids))}") split_ids = {"P024-A", "P024-B", "P024-C"} check("大兴三校拆分且不保留聚合重复", split_ids.issubset(ids) and "P024" not in ids, str(sorted(split_ids.intersection(ids)))) check("20个唯一SKU", len(sku_rows) == 20 and len({r["SKU编码"] for r in sku_rows}) == 20, f"rows={len(sku_rows)}") check("20个SKU规格键不重复", len({r["SKU唯一规格键"] for r in sku_rows}) == 20, f"unique_keys={len({r['SKU唯一规格键'] for r in sku_rows})}") required_project_fields = ["项目名称", "场景类型", "服务人群", "主要问题", "主SKU", "标准软件范围（产品定义）", "前厅主流程", "后厨主流程", "8分钟演示主线", "销售价值", "明确边界", "证据等级"] incomplete = [r["项目单元ID"] for r in projects if any(not str(r.get(field, "")).strip() for field in required_project_fields)] check("30项目销售字段均补齐", not incomplete, f"incomplete={incomplete}") mapped_ids = {r["项目单元ID"] for r in mapping} check("项目无孤儿", set(ids) == mapped_ids, f"unmapped={sorted(set(ids)-mapped_ids)}") valid_levels = {"R0-场景建议", "R1-流程/单侧复刻", "R2-项目校准复刻", "R3-合同级1:1复刻"} check("SKU复刻等级合法", all(r["当前复刻等级"] in valid_levels for r in sku_rows), "all valid") r3_rows = [r for r in sku_rows if r["当前复刻等级"].startswith("R3")] check("无伪造R3", not r3_rows, f"R3={len(r3_rows)}") check("所有SKU有母项目", all(r["母项目ID"] for r in sku_rows), "20/20") check("工作簿已生成", workbook.exists() and workbook.stat().st_size > 0, str(workbook)) wb = load_workbook(workbook, data_only=False, read_only=True) expected = {"先看这里", "20SKU销售表", "1to1复刻包", "30项目产品化", "项目SKU映射", "SKU软件功能", "项目软件证据", "项目硬件BOM", "证据缺口", "来源索引"} check("Excel工作表完整", expected == set(wb.sheetnames), ",".join(wb.sheetnames)) formula_errors = [] for ws in wb.worksheets: for row in ws.iter_rows(): for cell in row: if isinstance(cell.value, str) and cell.value in {"#REF!", "#DIV/0!", "#VALUE!", "#NAME?", "#N/A"}: formula_errors.append(f"{ws.title}!{cell.coordinate}:{cell.value}") check("无公式错误", not formula_errors, str(formula_errors[:10])) wb.close() return {"passed": all(c["passed"] for c in checks), "checks": checks} def build_all(output_dir: Path = OUTPUT_DIR, user_copy: Path | None = USER_OUTPUT) -> dict[str, object]: output_dir.mkdir(parents=True, exist_ok=True) projects = base_projects() sw_rows = build_software_rows(projects) hw_rows = build_hardware_rows(projects) project_rows = make_project_rows(projects, sw_rows, hw_rows) mapping_rows = make_mapping_rows(projects) sku_rows = make_sku_rows(project_rows, sw_rows, hw_rows) replication_rows = make_replication_rows(sku_rows) sku_function_rows = make_sku_function_rows() gap_rows = make_gap_rows(project_rows, sku_rows) sources = source_rows() data = { "projects": project_rows, "skus": sku_rows, "replication": replication_rows, "mapping": mapping_rows, "sku_functions": sku_function_rows, "project_software": sw_rows, "project_hardware": hw_rows, "gaps": gap_rows, "sources": sources, } write_csv(output_dir / "30-project-master.csv", project_rows) write_csv(output_dir / "sku-clustering.csv", sku_rows) write_csv(output_dir / "sku-1to1-replication.csv", replication_rows) write_csv(output_dir / "project-sku-mapping.csv", mapping_rows) write_csv(output_dir / "software-function-detail.csv", sku_function_rows) write_csv(output_dir / "project-software-evidence.csv", sw_rows) write_csv(output_dir / "hardware-bom-detail.csv", hw_rows) write_csv(output_dir / "evidence-gaps.csv", gap_rows) write_csv(output_dir / "source-index.csv", sources) workbook = output_dir / WORKBOOK_NAME build_workbook(workbook, project_rows, sku_rows, replication_rows, mapping_rows, sku_function_rows, sw_rows, hw_rows, gap_rows, sources) validation = validate_deliverables(data, workbook) (output_dir / "validation.json").write_text(json.dumps(validation, ensure_ascii=False, indent=2), encoding="utf-8") if user_copy: user_copy.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(workbook, user_copy) return {"data": data, "workbook": workbook, "validation": validation, "user_copy": user_copy} if __name__ == "__main__": result = build_all() print(json.dumps({ "workbook": str(result["workbook"]), "user_copy": str(result["user_copy"]), "passed": result["validation"]["passed"], "checks": result["validation"]["checks"], }, ensure_ascii=False, indent=2))