#!/usr/bin/env python3 """Build portfolio cards, Markdown, a portable HTML report, and artifact JSON.""" from __future__ import annotations import csv import datetime as dt import html import json from collections import Counter from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parent AS_OF = "2026-08-03" CURRENT_PERIOD = "2026-07-05 至 2026-08-03" PREVIOUS_PERIOD = "2026-06-05 至 2026-07-04" TITLE = "28个智慧食堂项目：哪些功能在真正使用" def read_csv(name: str) -> list[dict[str, str]]: with (ROOT / name).open(encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def write_csv(name: str, rows: list[dict[str, Any]]) -> None: fields: list[str] = [] for row in rows: for key in row: if key not in fields: fields.append(key) with (ROOT / name).open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows(rows) def integer(value: Any) -> int: return int(float(value or 0)) def number(value: Any) -> float: return float(value or 0) def pct(value: float) -> str: return f"{value:.1%}" def delta(current: int, previous: int) -> str: return f"{current / previous - 1:+.1%}" if previous else "—" def money(value: float) -> str: return f"¥{value:,.2f}" def derive() -> dict[str, Any]: master = read_csv("project-master.csv") core = read_csv("db-project-status.csv") orders = read_csv("db-order-periods.csv") channels = read_csv("db-channel-mix.csv") payments = read_csv("db-payment-mix.csv") meals = read_csv("db-meal-mix.csv") features = read_csv("db-feature-activity.csv") central = read_csv("central-project-summary.csv") core_by_key = {row["project_key"]: row for row in core} central_by_key = {row["project_code"]: row for row in central} current_order = {(row["project_key"], row["period"]): row for row in orders} channel_rows = [row for row in channels if row["period"] == "近30天"] channel_by_key: dict[str, list[dict[str, str]]] = {} for row in channel_rows: channel_by_key.setdefault(row["project_key"], []).append(row) feature_by_key: dict[str, list[dict[str, str]]] = {} for row in features: feature_by_key.setdefault(row["project_key"], []).append(row) cards: list[dict[str, Any]] = [] for row in master: key = row["central_project_code"] db = core_by_key.get(key, {}) central_row = central_by_key.get(key, {}) db_state = row["database_state"] orders30 = integer(db.get("orders30")) if db_state == "live" else 0 users30 = integer(db.get("mau30")) if db_state == "live" else 0 amount30 = number(db.get("amount30")) if db_state == "live" else 0 prev = current_order.get((key, "前30天"), {}) current = current_order.get((key, "近30天"), {}) previous_orders = integer(prev.get("orders")) current_orders = integer(current.get("orders")) if current else orders30 top_channels = sorted(channel_by_key.get(key, []), key=lambda item: integer(item["orders"]), reverse=True) queried_features = [ item for item in feature_by_key.get(key, []) if item["table_state"] == "queried" and integer(item["current_30d_records"]) > 0 ] queried_features.sort(key=lambda item: integer(item["current_30d_records"]), reverse=True) if db_state == "live" and orders30 >= 100_000: tier, state = "L1 高频运行", "生产库有大规模稳定交易" elif db_state == "live" and orders30 >= 10_000: tier, state = "L2 稳定运行", "生产库有稳定交易" elif db_state == "live" and orders30 > 0: tier, state = "L3 低量/波动", "生产库有交易但规模较低" elif db_state == "live": tier, state = "L4 已连接无近30天交易", "可连接不等于功能已激活" elif db_state == "unavailable": tier, state = "L5 不可核验", "数据库不可达，不能记为零使用" elif row["central_registered"] == "yes": tier, state = "L6 中央登记未直连", "只有中央项目维度，缺直接数据库映射" elif row["inclusion_source"] == "近期项目线程": tier, state = "L7 项目推进证据", "有项目证据，暂无线上使用证据" else: tier, state = "L8 规划/售前", "不得解读为已上线" top_entry = ( f"{top_channels[0]['channel']} {pct(number(top_channels[0]['order_share']))}" if top_channels else "—" ) feature_text = "；".join( f"{item['feature']} {integer(item['current_30d_records']):,}" for item in queried_features[:3] ) or "无可见近30天功能记录" if tier in ("L1 高频运行", "L2 稳定运行"): action = "优先保稳定性、对账、自动报表，并补齐关键链路埋点" elif tier == "L3 低量/波动": action = "核查下降/低量原因，区分客群规模、停运和产品激活问题" elif tier == "L4 已连接无近30天交易": action = "确认项目是否上线、停运或仅完成部署，补激活验收" elif tier == "L5 不可核验": action = "先恢复只读数据链路；未恢复前不评价使用率" else: action = "建立唯一项目ID、中央项目码与数据库映射后再做使用分析" cards.append({ "project_id": row["project_id"], "project_name": row["project_name"], "inclusion_source": row["inclusion_source"], "usage_tier": tier, "data_state": state, "orders30": orders30 if db_state == "live" else "", "active_users30": users30 if db_state == "live" else "", "amount30": round(amount30, 2) if db_state == "live" else "", "order_change_vs_previous30": delta(current_orders, previous_orders) if current else "", "top_business_entry": top_entry, "top_feature_signals": feature_text if db_state == "live" else "—", "central_current30_orders": integer(central_row.get("current30_orders")) if central_row else "", "central_vs_direct_gap": ( integer(central_row.get("current30_orders")) - orders30 if central_row and db_state == "live" else "" ), "project_phase_or_health": " / ".join(filter(None, [row["current_phase"], row["project_health"]])), "next_action": action, "evidence_level": row["evidence_level"], "boundary": row["usage_analysis_boundary"], }) live = [row for row in core if row["data_status"] == "live"] active = [row for row in live if integer(row["orders30"]) > 0] active_sorted = sorted(active, key=lambda row: integer(row["orders30"]), reverse=True) direct_orders = sum(integer(row["orders30"]) for row in active) direct_amount = sum(number(row["amount30"]) for row in active) direct_users_sum = sum(integer(row["mau30"]) for row in active) central_orders = sum(integer(row["current30_orders"]) for row in central) channel_totals: Counter[str] = Counter() for row in channel_rows: channel_totals[row["channel"]] += integer(row["orders"]) meal_totals: Counter[str] = Counter() for row in meals: if row["period"] == "近30天": meal_totals[row["meal"]] += integer(row["orders"]) feature_totals: Counter[str] = Counter() for row in features: if row["table_state"] == "queried": feature_totals[row["feature"]] += integer(row["current_30d_records"]) project_rank = [ { "project": row["project_name"], "orders": integer(row["orders30"]), "active_users": integer(row["mau30"]), "amount": number(row["amount30"]), "order_share": integer(row["orders30"]) / direct_orders if direct_orders else 0, } for row in active_sorted ] channel_rank = [ {"channel": label, "orders": count, "order_share": count / direct_orders if direct_orders else 0} for label, count in channel_totals.most_common() ] feature_rank = [ {"feature": label, "records": count} for label, count in feature_totals.most_common() if count > 0 ] meal_rank = [ {"meal": label, "orders": count, "order_share": count / direct_orders if direct_orders else 0} for label, count in meal_totals.most_common() ] reconciliation = [ {"scope": "历史线程里的28", "count": 28, "meaning": "2026-07-21时点的在途事项，不是客户项目数", "can_be_master": "no"}, {"scope": "历史业务仓库", "count": 11, "meaning": "代码仓库数量，不是客户项目数", "can_be_master": "no"}, {"scope": "数据库直连注册表", "count": 17, "meaning": "有数据库连接配置的项目清单", "can_be_master": "partial"}, {"scope": "当前中央统计项目维度", "count": 20, "meaning": "中央统计登记项目，当前最接近线上项目总表", "can_be_master": "partial"}, {"scope": "7月22项目同时推进", "count": 22, "meaning": "项目文档中的时点说法，缺唯一原始名单", "can_be_master": "no"}, {"scope": "本报告28候选项目", "count": 28, "meaning": "20中央项目+5近期独立项目+3规划项目的可追溯重建", "can_be_master": "candidate_only"}, ] portfolio = [ {"metric": "重建候选项目", "value": 28, "note": "不是当前权威主数据"}, {"metric": "中央统计已登记", "value": 20, "note": "其中部分同步失败"}, {"metric": "数据库注册", "value": 17, "note": "有连接配置"}, {"metric": "本次数据库可连接", "value": len(live), "note": "生产库只读"}, {"metric": "近30天有有效订单", "value": len(active), "note": CURRENT_PERIOD}, {"metric": "数据库不可达", "value": len([row for row in core if row["data_status"] == "unavailable"]), "note": "不可解释为零使用"}, {"metric": "近30天有效订单", "value": direct_orders, "note": "7个有单项目合计"}, {"metric": "近30天项目活跃用户之和", "value": direct_users_sum, "note": "跨项目未去重"}, {"metric": "近30天订单流水", "value": round(direct_amount, 2), "note": "不等于收入/回款"}, {"metric": "中央统计同期订单", "value": central_orders, "note": "比直连少，存在同步覆盖差异"}, ] return { "master": master, "cards": cards, "reconciliation": reconciliation, "portfolio": portfolio, "project_rank": project_rank, "channel_rank": channel_rank, "feature_rank": feature_rank, "meal_rank": meal_rank, "direct_orders": direct_orders, "direct_amount": direct_amount, "direct_users_sum": direct_users_sum, "central_orders": central_orders, } def bar_rows(rows: list[dict[str, Any]], label: str, value: str, formatter, limit: int = 10) -> str: selected = rows[:limit] maximum = max((number(row[value]) for row in selected), default=1) parts = [] for row in selected: width = number(row[value]) / maximum * 100 if maximum else 0 parts.append( f'
{html.escape(str(row[label]))}
' f'
' f'
{html.escape(formatter(row[value]))}
' ) return "".join(parts) def table_html(headers: list[tuple[str, str]], rows: list[dict[str, Any]], classes: str = "") -> str: head = "".join(f"{html.escape(label)}" for _, label in headers) body = [] for row in rows: cells = [] for key, _ in headers: value = row.get(key, "") cells.append(f"{html.escape(str(value))}") body.append("" + "".join(cells) + "") return f'
{head}{"".join(body)}
' def build_markdown(data: dict[str, Any]) -> str: projects = data["project_rank"] channels = data["channel_rank"] features = data["feature_rank"] top3_share = sum(row["orders"] for row in projects[:3]) / data["direct_orders"] lines = [ f"# {TITLE}", "", f"报告日：{AS_OF}｜当前窗口：{CURRENT_PERIOD}｜前一窗口：{PREVIOUS_PERIOD}", "", "## Executive Summary｜执行摘要", "", "- **28不是当前权威项目总数。** 历史线程里的28指当时的在途事项；本报告重建了28个可追溯候选项目，用来把其他线程的项目统一放到一张表里。当前线上中央统计登记20个，数据库注册17个。", f"- **本次可核验12个数据库，其中7个近30天有有效订单。** 7个项目合计 {data['direct_orders']:,} 单、项目活跃用户之和 {data['direct_users_sum']:,}、订单流水 {money(data['direct_amount'])}；用户数跨项目未去重，流水不等于收入。", f"- **交易高度集中。** 金斯瑞、机场、赛迪合计贡献 {pct(top3_share)} 的直连有效订单；消费机贡献 {channels[0]['orders']:,} 单，占 {pct(channels[0]['order_share'])}，是跨项目使用最多的功能入口。", f"- **运营功能呈明显分层。** 订餐取餐流程 {features[0]['records']:,} 条、取餐柜接口 {features[1]['records']:,} 条、安全审计 {features[2]['records']:,} 条；健康数据、健康测量、AI营养对话的可见业务记录很少或为零。", f"- **中央统计暂不能替代生产库直连。** 同期中央统计只有 {data['central_orders']:,} 单，生产库直连为 {data['direct_orders']:,} 单，少计 {data['direct_orders']-data['central_orders']:,} 单；同时前30天中央活跃用户返回0，需先修复同步与指标质量。", "", "## 哪些功能使用最多", "", "### 1. 交易入口：消费机是绝对主入口", "", "| 功能入口 | 近30天订单 | 占直连有效订单 |", "|---|---:|---:|", ] lines += [f"| {row['channel']} | {row['orders']:,} | {pct(row['order_share'])} |" for row in channels] lines += ["", "### 2. 项目规模：前三个项目决定了绝大多数真实用量", "", "| 项目 | 近30天订单 | 项目活跃用户 | 订单流水 | 订单占比 |", "|---|---:|---:|---:|---:|"] lines += [f"| {row['project']} | {row['orders']:,} | {row['active_users']:,} | {money(row['amount'])} | {pct(row['order_share'])} |" for row in projects] lines += ["", "### 3. 非交易功能：取餐与运营管理真实使用，健康创新功能仍弱", "", "| 功能信号 | 近30天记录 | 解释边界 |", "|---|---:|---|"] boundary = { "订餐取餐流程": "流程记录，不等于独立用户数", "取餐柜接口": "接口请求/业务记录", "后台安全审计操作": "后台操作轨迹", "实体卡建档": "配置/建档，不等于日常使用", "退款申请": "主动业务动作", "后台报表导出": "主动业务动作", "人脸建档": "配置/建档，不等于刷脸支付", "消费限额拦截": "规则触发记录", } lines += [f"| {row['feature']} | {row['records']:,} | {boundary.get(row['feature'], '业务表记录')} |" for row in features] lines += ["", "## 28个项目逐项卡片", "", "| 项目 | 使用层级 | 近30天订单 | 主入口 | 可见功能信号 | 下一步 |", "|---|---|---:|---|---|---|"] for row in data["cards"]: orders = f"{row['orders30']:,}" if isinstance(row["orders30"], int) else "—" lines.append(f"| {row['project_name']} | {row['usage_tier']} | {orders} | {row['top_business_entry']} | {row['top_feature_signals']} | {row['next_action']} |") lines += [ "", "## 建议动作", "", "1. **建立唯一项目主数据。** 把客户项目、代码仓库、部署环境、中央项目码、数据库名和线程任务ID分开管理；当前不存在一份可证明的权威28项目清单。", "2. **修复中央统计同步与口径。** 优先解释同期少计241,174单、前30天活跃用户为0，以及多个直连正常项目在中央被标为failed的问题。", "3. **按项目画像运营。** 金斯瑞围绕消费机+取餐柜；机场围绕消费机+外部订单；赛迪围绕消费机+闸机+限额；产业园围绕绑盘+虚拟订单；西康围绕绑盘。", "4. **给健康与AI功能补真实埋点和激活目标。** 业务表零记录只说明没有可见业务动作，不能证明页面从未访问；必须补页面/小程序事件流才可评价功能渗透率。", "5. **把不可达和零使用彻底分开。** 滨州、城市副中心、江西206、莱蒂森、新吴区当前只能标为不可核验，不能写成没有用户或没有订单。", "", "## 仍需回答的问题", "", "- 20个中央项目之外，国康、四方达、网信办、大兴、安徽部队是否已有独立生产项目码或共用数据库？", "- 中煤、保康三中、C08是否仍处于售前/规划，还是已经进入正式交付？", "- 是否将“功能使用”升级为点击流、页面访问、设备在线、接口成功率和业务结果的统一遥测，而不只依赖业务表记录？", "", "## 口径、假设与限制", "", "- 有效订单：支付完成、订单完成、未删除、有效用户；字段不存在时按项目实际字段降级，按meal_date归属。", "- 当前窗口与前一窗口均为30个完整自然日；报告日排除2026-08-04当日未完整数据。", "- 数据库只保存项目级聚合，不保存凭据、人员、手机号、订单明细或原始生产payload。", "- 记录数不是页面访问人数；配置/建档、流程记录、规则触发和主动动作已分开解释。", "- 28项目候选清单是多源重建，不是正式项目主数据。", "", "## 来源", "", "- 中央统计API：20项目维度及两个30天窗口聚合；接口明确只查中央统计表，不直接连接项目业务库。", "- 生产数据库：17项目注册表，本次12个项目只读可连接；5个不可达。", "- 项目证据：近期17项统一分析中的独立项目，以及未来项目分层图。", "- 历史线程：2026-07-21答复明确区分11个业务仓库与28个在途事项。", ] return "\n".join(lines) + "\n" def build_html(data: dict[str, Any]) -> str: top3_share = sum(row["orders"] for row in data["project_rank"][:3]) / data["direct_orders"] cards = data["cards"] card_headers = [ ("project_name", "项目"), ("usage_tier", "使用层级"), ("orders30", "近30天订单"), ("top_business_entry", "主入口"), ("top_feature_signals", "可见功能信号"), ("next_action", "下一步"), ] reconciliation_table = table_html( [("scope", "口径"), ("count", "数量"), ("meaning", "真实含义"), ("can_be_master", "能否作主表")], data["reconciliation"], ) project_table = table_html(card_headers, cards, "project-table") project_bars = bar_rows(data["project_rank"], "project", "orders", lambda value: f"{integer(value):,}") channel_bars = bar_rows(data["channel_rank"], "channel", "orders", lambda value: f"{integer(value):,}") feature_bars = bar_rows(data["feature_rank"], "feature", "records", lambda value: f"{integer(value):,}", 8) generated = dt.datetime.now().astimezone().isoformat(timespec="minutes") return f""" {TITLE}
SMART CANTEEN PORTFOLIO · 2026-08-03

{TITLE}

从其他线程、中央统计、17项目数据库注册表和生产库只读聚合重建全局视角。28是候选组合，不是当前权威主数据。

Executive Summary｜执行摘要

28的历史口径不是项目数旧线程的28指在途事项。本报告用20个中央项目、5个近期独立项目、3个规划项目重建28个候选。
12个库可核验，7个近30天有交易合计 {data['direct_orders']:,} 单、项目活跃用户之和 {data['direct_users_sum']:,}、流水 {money(data['direct_amount'])}。
消费机是使用最多的功能入口{data['channel_rank'][0]['orders']:,} 单，占直连有效订单 {pct(data['channel_rank'][0]['order_share'])}；前三项目贡献 {pct(top3_share)}。
中央统计存在同步缺口同期中央 {data['central_orders']:,} 单，直连 {data['direct_orders']:,} 单，少计 {data['direct_orders']-data['central_orders']:,} 单。
28重建候选项目
20中央统计已登记
12 / 17数据库可连接 / 注册
7近30天有有效订单

先把“28”说清楚

项目数、在途事项、代码仓库和数据库环境是四种不同对象。当前没有一份能证明“正好28个客户项目”的权威主表。

{reconciliation_table}

真实使用集中在交易核心

近30天项目订单

{project_bars}

仅12个可连接生产库中的7个有单项目；不可达项目未记为0。

功能入口订单

{channel_bars}

入口来自订单source枚举；消费机占绝对多数。

非交易功能业务记录

{feature_bars}

记录数不是用户数，不同表之间只能判断活跃信号，不能直接当同一转化漏斗。

项目画像不同，不能用一套运营动作

28个项目逐项卡片

空白不是0；不可达、无数据库映射、规划中分别标记。

{project_table}

建议动作

  1. 建立唯一项目主数据：客户项目、仓库、部署环境、中央项目码、数据库和线程任务ID分别建模。
  2. 修复中央统计：解释同期少计241,174单、前30天活跃用户为0，以及直连正常但中央sync_status=failed的问题。
  3. 按项目画像运营：优先优化各自高频链路，不用“所有项目都推同一套功能”。
  4. 补齐功能遥测：页面点击、小程序访问、设备在线、接口成功率和业务结果统一埋点，才能评价真正的功能渗透。
  5. 恢复5个不可达项目的数据链路：滨州、城市副中心、江西206、莱蒂森、新吴区未恢复前不得评价使用率。
决策边界：本报告可以回答“当前可核验项目里哪些功能最常用”，不能证明未映射或不可达项目没有使用，也不能证明28是正式项目总数。

口径、来源与限制

""" def main() -> None: data = derive() write_csv("project-usage-cards.csv", data["cards"]) write_csv("source-reconciliation.csv", data["reconciliation"]) write_csv("portfolio-summary.csv", data["portfolio"]) write_csv("project-order-ranking.csv", data["project_rank"]) write_csv("channel-ranking.csv", data["channel_rank"]) write_csv("feature-ranking.csv", data["feature_rank"]) write_csv("meal-ranking.csv", data["meal_rank"]) (ROOT / "summary.md").write_text(build_markdown(data), encoding="utf-8") (ROOT / "smart-canteen-28-project-usage.html").write_text(build_html(data), encoding="utf-8") artifact = { "manifest": { "title": TITLE, "audience": "product stakeholders", "report_date": AS_OF, "current_period": CURRENT_PERIOD, "previous_period": PREVIOUS_PERIOD, "privacy": "aggregate only; no credentials or person/order-level records", }, "datasets": { "portfolio_summary": data["portfolio"], "project_usage_cards": data["cards"], "project_order_ranking": data["project_rank"], "channel_ranking": data["channel_rank"], "feature_ranking": data["feature_rank"], "meal_ranking": data["meal_rank"], "source_reconciliation": data["reconciliation"], }, "sources": [ {"id": "central", "label": "中央统计API", "path": "central-current30.json", "boundary": "只查中央统计表"}, {"id": "db", "label": "17项目生产库只读聚合", "path": "db-project-status.csv", "boundary": "12可连接，5不可达；已移除本机访问字段"}, {"id": "threads", "label": "项目线程与项目分层证据", "path": "project-master.csv", "boundary": "B级项目证据不等于线上使用"}, ], } (ROOT / "artifact.json").write_text(json.dumps(artifact, ensure_ascii=False, indent=2), encoding="utf-8") write_csv("chart-map.csv", [ {"chart": "近30天项目订单", "dataset": "project-order-ranking.csv", "type": "horizontal bar", "question": "哪些项目贡献真实用量"}, {"chart": "功能入口订单", "dataset": "channel-ranking.csv", "type": "horizontal bar", "question": "哪些业务入口使用最多"}, {"chart": "非交易功能业务记录", "dataset": "feature-ranking.csv", "type": "horizontal bar", "question": "哪些运营功能有明确业务信号"}, ]) print(json.dumps({"projects": len(data["cards"]), "orders": data["direct_orders"], "central_orders": data["central_orders"]}, ensure_ascii=False)) if __name__ == "__main__": main()