#!/usr/bin/env python3 """Build SKU-level product financial statement from strategy and attribution sources.""" from __future__ import annotations import csv import html import json import re from collections import defaultdict from datetime import date from pathlib import Path from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter ROOT = Path(__file__).resolve().parents[2] OUT = Path(__file__).resolve().parent SOURCES = { "3spu9sku": ROOT / "work/2026-09-02-smart-canteen-3-spu-9-sku-strategy/strategy-model.json", "12sku": ROOT / "work/2026-09-02-smart-canteen-12-sku-strategy/strategy-model.json", "nc_clusters": ROOT / "work/2026-08-31-thirty-project-nutrition-sku-clustering/output/sku-clustering.csv", "sku_mapping": ROOT / "work/2026-08-31-thirty-project-nutrition-sku-clustering/output/project-sku-mapping.csv", "product_finance": ROOT / "standards-stack/product-strategy/smart-canteen/PRODUCT_LINE_FINANCE.csv", "pricing": ROOT / "standards-stack/product-strategy/smart-canteen/PRICING_ARCHITECTURE_V0.4.csv", "investor_metrics": ROOT / "standards-stack/product-strategy/smart-canteen/INVESTOR_METRIC_DASHBOARD_V0.1.csv", "attribution_md": ROOT / "work/2026-09-02-group-meal-project-sku-attribution-review/团餐项目SKU归因方案_待Review.md", } REPORT_DATE = date(2026, 9, 2) def read_json(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def read_csv(path: Path) -> list[dict[str, str]]: with path.open(newline="", encoding="utf-8-sig") as f: return list(csv.DictReader(f)) def parse_wan(text: str) -> float: text = text.strip() if text.endswith("万"): return round(float(text[:-1]), 2) if text.endswith("元"): return round(float(text[:-1]) / 10000, 2) if not text: return 0.0 return round(float(text), 2) SKU_NAME_TO_CODE = { "单食堂消费结算": "SPU01-SKU01", "多档口消费结算": "SPU01-SKU02", "多档口多食堂消费结算": "SPU01-SKU02", "集团版消费结算": "SPU01-SKU03", "补贴与多策略结算": "SPU01-SKU04", "绑盘称重结算": "SPU01-SKU05", "免绑盘称重结算": "SPU01-SKU06", "AI视觉识别结算": "SPU01-SKU07", "移动订餐与取餐": "SPU01-SKU08", "K12家校阳光消费": "SPU01-SKU09", "营养菜单与电子菜牌": "SPU01-SKU10", "集中保障供餐": "SPU01-SKU11", "营养健康智厨专业版": "SPU01-SKU12", "单食堂智慧食安": "SPU02-SKU01", "多食堂智慧食安": "SPU02-SKU02", "集团智慧食安": "SPU02-SKU03", "单食堂进销存": "SPU03-SKU01", "多食堂进销存": "SPU03-SKU02", "集团采购与库存": "SPU03-SKU03", } def extract_project_rows(path: Path) -> list[dict[str, str | float | list[str]]]: text = path.read_text(encoding="utf-8") rows = [] in_table = False for raw_line in text.splitlines(): line = raw_line.strip() if line.startswith("| 序号 | 开票客户/项目线索 |"): in_table = True continue if not in_table: continue if line.startswith("| ---"): continue if not line.startswith("|"): if rows: break continue cells = [cell.strip() for cell in line.strip("|").split("|")] if len(cells) < 9: continue amount_wan = parse_wan(cells[2]) evidence_status = cells[8] sku_codes: list[str] = [] for source_cell in cells[4:7]: if "不归因" in source_cell or "不判定" in source_cell or "不独立" in source_cell: continue for name, code in SKU_NAME_TO_CODE.items(): if name in source_cell and code not in sku_codes: sku_codes.append(code) if "必须拆单" in evidence_status or "暂挂" in " ".join(cells[4:9]) or len(sku_codes) > 1: bucket = "组合待拆/暂挂" elif len(sku_codes) == 1 and amount_wan > 0: bucket = "已确认单SKU归因" elif amount_wan > 0: bucket = "零散待挂母项目" else: bucket = "零金额证据" rows.append( { "序号": cells[0], "开票客户/项目线索": cells[1], "基表金额_万元": amount_wan, "销售母版本": cells[3], "前厅主SKU": cells[4], "食安主SKU": cells[5], "进销存主SKU": cells[6], "关键依据与处理意见": cells[7], "证据状态": evidence_status, "归因桶": bucket, "命中SKU编码": "；".join(sku_codes), "_sku_codes": sku_codes, } ) return rows def build_18_sku_catalog(model_12: dict, model_3: dict, cluster_rows: list[dict[str, str]]) -> list[dict[str, str | float | int]]: catalog: list[dict[str, str | float | int]] = [] nutrition_target = next( float(spu["target_wan"]) for spu in model_3["spus"] if spu["name"] == "营养结算" ) nutrition_raw_total = sum(float(item["target_wan"]) for item in model_12["skus"]) nutrition_scale = nutrition_target / nutrition_raw_total for item in model_12["skus"]: code = item["code"] target_wan = round(float(item["target_wan"]) * nutrition_scale, 2) target_units = int(item["target_units"]) catalog.append( { "SPU编码": "SPU01", "SPU名称": "营养结算", "SKU编码": code, "SKU名称": item["name"], "销售/战略角色": item["role"], "目标SKU成交/增购行数": target_units, "规划客单价_万元": round(target_wan / target_units, 2) if target_units else 0, "未来12个月目标收入_万元": target_wan, "原12SKU目标收入_万元": item["target_wan"], "目标校准说明": "按3SPU9SKU最新营养结算720万总盘子等比例校准", "战略动作": item["action"], } ) suffix_map = { "单食堂智慧食安": "SPU02-SKU01", "多食堂智慧食安": "SPU02-SKU02", "集团智慧食安": "SPU02-SKU03", "单食堂进销存": "SPU03-SKU01", "多食堂进销存": "SPU03-SKU02", "集团采购与库存": "SPU03-SKU03", } for item in model_3["skus"]: if item["name"] not in suffix_map: continue code = suffix_map[item["name"]] catalog.append( { "SPU编码": code.split("-")[0], "SPU名称": item["spu"], "SKU编码": code, "SKU名称": item["name"], "销售/战略角色": item["role"], "目标SKU成交/增购行数": item["units"], "规划客单价_万元": item["asp_wan"], "未来12个月目标收入_万元": item["target_wan"], "原12SKU目标收入_万元": "", "目标校准说明": "采用3SPU9SKU目标", "战略动作": item["action"], } ) project_counts = defaultdict(int) nc_to_spu_sku = { "NC-01": "SPU01-SKU01", "NC-02": "SPU01-SKU02", "NC-03": "SPU01-SKU02", "NC-04": "SPU01-SKU04", "NC-05": "SPU01-SKU04", "NC-06": "SPU01-SKU04", "NC-07": "SPU01-SKU05", "NC-08": "SPU01-SKU05", "NC-09": "SPU01-SKU05", "NC-10": "SPU01-SKU06", "NC-11": "SPU01-SKU07", "NC-12": "SPU01-SKU05", "NC-13": "SPU01-SKU08", "NC-14": "SPU01-SKU08", "NC-15": "SPU01-SKU08", "NC-16": "SPU01-SKU10", "NC-17": "SPU01-SKU10", "NC-18": "SPU01-SKU10", "NC-19": "SPU01-SKU12", "NC-20": "SPU01-SKU11", } cluster_meta = defaultdict(list) for row in cluster_rows: sku = nc_to_spu_sku.get(row["SKU编码"]) if not sku: continue projects = [p for p in re.split(r"[；/]", row["覆盖项目单元"]) if p] project_counts[sku] += len(projects) cluster_meta[sku].append(f"{row['SKU编码']} {row['SKU名称']}({row['当前复刻等级']})") for item in catalog: sku = str(item["SKU编码"]) item["历史项目证据覆盖数"] = project_counts.get(sku, 0) item["30项目聚类映射"] = "；".join(cluster_meta.get(sku, [])) or "待补证" return catalog def enrich_finance(catalog: list[dict], project_rows: list[dict]) -> list[dict]: actual = defaultdict(float) pending = defaultdict(float) related = defaultdict(float) lead_count = defaultdict(int) clean_projects = defaultdict(list) pending_projects = defaultdict(list) for row in project_rows: sku_codes = row["_sku_codes"] amount = float(row["基表金额_万元"]) for sku in sku_codes: related[sku] += amount lead_count[sku] += 1 if row["归因桶"] == "已确认单SKU归因" and len(sku_codes) == 1: actual[sku_codes[0]] += amount clean_projects[sku_codes[0]].append(str(row["开票客户/项目线索"])) elif sku_codes: for sku in sku_codes: pending[sku] += amount pending_projects[sku].append(str(row["开票客户/项目线索"])) for item in catalog: sku = item["SKU编码"] target = float(item["未来12个月目标收入_万元"]) item["已确认单SKU样本收入_万元"] = round(actual[sku], 2) item["组合/暂挂项目待拆金额_万元"] = round(pending[sku], 2) item["相关项目线索金额_万元_可重复计"] = round(related[sku], 2) item["相关项目线索数"] = lead_count[sku] item["目标覆盖率_按已确认样本"] = round(actual[sku] / target, 4) if target else 0 item["已确认样本项目"] = "；".join(clean_projects[sku]) or "无" item["待拆/暂挂项目"] = "；".join(dict.fromkeys(pending_projects[sku])) or "无" item["财务科目待补"] = "回款、应收、预收、硬件成本、实施人天、维保/续费、毛利" if pending[sku] > 0: item["经营调整建议"] = "先拆终端项目/订单簇，再拆软件、硬件、接口、实施和维保收入" elif actual[sku] > 0: item["经营调整建议"] = "可先进入SKU收入看板，同时补回款和成本形成毛利口径" elif target > 0: item["经营调整建议"] = "保留为目标SKU，必须补真实商机、报价包和交付成本后再确认预算" else: item["经营调整建议"] = "待补证" return catalog def rollup(rows: list[dict], keys: list[str]) -> list[dict]: acc = {} for row in rows: key = tuple(row[k] for k in keys) if key not in acc: acc[key] = {k: row[k] for k in keys} for field in [ "未来12个月目标收入_万元", "目标SKU成交/增购行数", "已确认单SKU样本收入_万元", "组合/暂挂项目待拆金额_万元", "相关项目线索金额_万元_可重复计", "历史项目证据覆盖数", ]: acc[key][field] = 0 for field in [ "未来12个月目标收入_万元", "目标SKU成交/增购行数", "已确认单SKU样本收入_万元", "组合/暂挂项目待拆金额_万元", "相关项目线索金额_万元_可重复计", "历史项目证据覆盖数", ]: acc[key][field] += float(row[field]) out = [] for item in acc.values(): for field in list(item): if isinstance(item[field], float): item[field] = round(item[field], 2) out.append(item) return out def write_csv(path: Path, rows: list[dict], fieldnames: list[str]) -> None: with path.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def style_sheet(ws) -> None: header_fill = PatternFill("solid", fgColor="1F4E78") header_font = Font(color="FFFFFF", bold=True) thin = Side(style="thin", color="D9E2F3") border = Border(left=thin, right=thin, top=thin, bottom=thin) for cell in ws[1]: cell.fill = header_fill cell.font = header_font cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) for row in ws.iter_rows(): for cell in row: cell.border = border cell.alignment = Alignment(vertical="top", wrap_text=True) ws.freeze_panes = "A2" for col in ws.columns: values = [str(c.value or "") for c in col[:80]] width = min(max(len(v) for v in values) + 2, 42) ws.column_dimensions[get_column_letter(col[0].column)].width = width def add_sheet(wb: Workbook, name: str, rows: list[dict], fieldnames: list[str]) -> None: ws = wb.create_sheet(name) ws.append(fieldnames) for row in rows: ws.append([row.get(f, "") for f in fieldnames]) style_sheet(ws) def write_xlsx(path: Path, sheets: list[tuple[str, list[dict], list[str]]]) -> None: wb = Workbook() wb.remove(wb.active) for name, rows, fields in sheets: add_sheet(wb, name, rows, fields) wb.save(path) def write_markdown( path: Path, sku_rows: list[dict], spu_rows: list[dict], project_rows: list[dict], finance_rows: list[dict], ) -> None: total_target = sum(float(r["未来12个月目标收入_万元"]) for r in sku_rows) total_actual = sum(float(r["已确认单SKU样本收入_万元"]) for r in sku_rows) total_sample = sum(float(r["基表金额_万元"]) for r in project_rows) split_amount = sum( float(r["基表金额_万元"]) for r in project_rows if r["归因桶"] == "组合待拆/暂挂" ) lines = [ "# 智慧食堂 SKU 产品财报表（补齐版）", "", f"> 生成日期：{REPORT_DATE.isoformat()}", "> 口径：以 18 个经营 SKU 为主账本，20 个营养结算聚类 SKU 作为证据映射，不把 Review 草案当作财务定版。", "", "## 结论", "", f"- 未来 12 个月 18SKU 经营目标合计 **{total_target:.0f} 万元**，其中营养结算 12SKU 按最新 3SPU9SKU 口径校准为 720 万元，食安 3SKU 目标 300 万元，进销存 3SKU 目标 180 万元。", f"- 团餐基表样本金额 **{total_sample:.2f} 万元**；当前可直接进入单 SKU 样本收入的金额为 **{total_actual:.2f} 万元**，其余大额项目主要属于跨 SPU 组合或渠道客户订单簇，必须先拆终端项目和收入构成。", f"- 已确认单 SKU 收入集中在绑盘称重、营养健康专业、K12、单食堂消费和 AI 视觉结算；组合待拆金额约 **{split_amount:.2f} 万元**，是下一步财务补表的主要工作量。", "", "## 3SPU 汇总", "", "| SPU | 目标收入(万元) | 已确认单SKU样本(万元) | 组合/暂挂待拆(万元) | 历史证据覆盖数 |", "| --- | ---: | ---: | ---: | ---: |", ] for r in spu_rows: lines.append( f"| {r['SPU名称']} | {r['未来12个月目标收入_万元']:.2f} | {r['已确认单SKU样本收入_万元']:.2f} | {r['组合/暂挂项目待拆金额_万元']:.2f} | {int(r['历史项目证据覆盖数'])} |" ) lines += [ "", "## 下一步调整", "", "1. 财务表主键从“开票客户”调整为“终端项目 + SKU + 收入科目”，渠道客户如江苏苏膳、河南聚铂源必须拆订单簇。", "2. 每个 SKU 补齐收入、应收、预收、回款、硬件成本、实施人天、维保/续费、毛利；没有成本和回款时只允许称为收入样本，不称为利润结论。", "3. 前厅 12SKU 继续细化销售入口，经营汇总对外收敛为 3 个销售版本：智能结算基础版、营养结算标准版、营养健康专业版。", "4. 食安和进销存不塞进前厅版本里，单独统计渗透率和交叉销售收入，才能看出增购能力。", "", "## 来源", "", ] for name, src in SOURCES.items(): lines.append(f"- {name}: `{src.relative_to(ROOT)}`") lines += [ "", "## 已用原产品线财务口径", "", "| 财务项 | 金额 | 证据 | 用途 |", "| --- | ---: | --- | --- |", ] for r in finance_rows: lines.append( f"| {r['product_line']} | {float(r['known_sales_amount'])/10000:.2f} 万元 | {r['evidence_level']} | {r['strategy_use']} |" ) path.write_text("\n".join(lines) + "\n", encoding="utf-8") def write_html(path: Path, sku_rows: list[dict], spu_rows: list[dict], project_rows: list[dict]) -> None: def table(rows: list[dict], fields: list[str]) -> str: body = [""] body += [f"" for f in fields] body += [""] for row in rows: body.append("") for f in fields: body.append(f"") body.append("") body.append("
{html.escape(f)}
{html.escape(str(row.get(f, '')))}
") return "".join(body) fields = [ "SPU名称", "SKU编码", "SKU名称", "未来12个月目标收入_万元", "原12SKU目标收入_万元", "目标校准说明", "已确认单SKU样本收入_万元", "组合/暂挂项目待拆金额_万元", "相关项目线索金额_万元_可重复计", "经营调整建议", ] project_fields = [ "开票客户/项目线索", "基表金额_万元", "销售母版本", "命中SKU编码", "归因桶", "证据状态", ] html_text = f""" 智慧食堂 SKU 产品财报表

智慧食堂 SKU 产品财报表

生成日期：{REPORT_DATE.isoformat()}；口径：18 个经营 SKU + 团餐项目归因 Review 草案 + 30 项目营养结算聚类证据
18SKU目标收入{sum(float(r['未来12个月目标收入_万元']) for r in sku_rows):.0f} 万
团餐样本金额{sum(float(r['基表金额_万元']) for r in project_rows):.2f} 万
可直接单SKU归因{sum(float(r['已确认单SKU样本收入_万元']) for r in sku_rows):.2f} 万
组合/暂挂待拆{sum(float(r['基表金额_万元']) for r in project_rows if r['归因桶']=='组合待拆/暂挂'):.2f} 万
财报补齐的核心不是把 486.10 万样本硬摊到 SKU，而是先把可确认的单 SKU 收入入账，把跨 SPU 组合项目列为待拆池，再补应收、预收、应付、硬件成本、实施人天和维保续费。

3SPU经营汇总

{table(spu_rows, ['SPU名称','未来12个月目标收入_万元','目标SKU成交/增购行数','已确认单SKU样本收入_万元','组合/暂挂项目待拆金额_万元','历史项目证据覆盖数'])}

18SKU产品财报表

{table(sku_rows, fields)}

项目金额归因明细

{table(project_rows, project_fields)}
""" path.write_text(html_text, encoding="utf-8") def main() -> None: OUT.mkdir(parents=True, exist_ok=True) model_3 = read_json(SOURCES["3spu9sku"]) model_12 = read_json(SOURCES["12sku"]) cluster_rows = read_csv(SOURCES["nc_clusters"]) project_rows = extract_project_rows(SOURCES["attribution_md"]) finance_rows = read_csv(SOURCES["product_finance"]) pricing_rows = read_csv(SOURCES["pricing"]) investor_rows = read_csv(SOURCES["investor_metrics"]) sku_rows = enrich_finance(build_18_sku_catalog(model_12, model_3, cluster_rows), project_rows) spu_rows = rollup(sku_rows, ["SPU编码", "SPU名称"]) finance_fields = [ "SPU编码", "SPU名称", "SKU编码", "SKU名称", "销售/战略角色", "目标SKU成交/增购行数", "规划客单价_万元", "未来12个月目标收入_万元", "原12SKU目标收入_万元", "目标校准说明", "已确认单SKU样本收入_万元", "组合/暂挂项目待拆金额_万元", "相关项目线索金额_万元_可重复计", "相关项目线索数", "目标覆盖率_按已确认样本", "历史项目证据覆盖数", "30项目聚类映射", "已确认样本项目", "待拆/暂挂项目", "财务科目待补", "经营调整建议", "战略动作", ] project_fields = [ "序号", "开票客户/项目线索", "基表金额_万元", "销售母版本", "前厅主SKU", "食安主SKU", "进销存主SKU", "命中SKU编码", "归因桶", "关键依据与处理意见", "证据状态", ] spu_fields = [ "SPU编码", "SPU名称", "未来12个月目标收入_万元", "目标SKU成交/增购行数", "已确认单SKU样本收入_万元", "组合/暂挂项目待拆金额_万元", "相关项目线索金额_万元_可重复计", "历史项目证据覆盖数", ] write_csv(OUT / "sku-product-financial-statement.csv", sku_rows, finance_fields) write_csv(OUT / "project-amount-attribution-ledger.csv", project_rows, project_fields) write_csv(OUT / "spu-financial-rollup.csv", spu_rows, spu_fields) write_markdown(OUT / "sku-product-financial-statement.md", sku_rows, spu_rows, project_rows, finance_rows) write_html(OUT / "sku-product-financial-statement.html", sku_rows, spu_rows, project_rows) sheets = [ ("00-来源与口径", [ {"项目": "生成日期", "内容": REPORT_DATE.isoformat()}, {"项目": "主账本", "内容": "18个经营SKU：营养结算12、智慧食安3、智慧进销存3"}, {"项目": "收入样本", "内容": "团餐基表SKU归因方案v0.1，295条明细、60个订单、22个开票客户、486.10万元"}, {"项目": "重要边界", "内容": "Review草案不替代财务定版；跨SPU组合项目不做强制摊派"}, ] + [{"项目": k, "内容": str(v.relative_to(ROOT))} for k, v in SOURCES.items()], ["项目", "内容"]), ("01-3SPU经营汇总", spu_rows, spu_fields), ("02-18SKU产品财报表", sku_rows, finance_fields), ("03-项目金额归因", project_rows, project_fields), ("04-产品线财务原表", finance_rows, list(finance_rows[0].keys())), ("05-定价科目", pricing_rows, list(pricing_rows[0].keys())), ("06-投资人指标缺口", investor_rows, list(investor_rows[0].keys())), ("07-NC20聚类证据", cluster_rows, list(cluster_rows[0].keys())), ] write_xlsx(OUT / "智慧食堂SKU产品财报表_20260902.xlsx", sheets) if __name__ == "__main__": main()