#!/usr/bin/env python3 """Build an auditable smart-canteen software/hardware portfolio snapshot. The source workbook mixes software, services, durable hardware, bulk materials, linear measures and area measures. This script keeps those grains separate and uses SQLite for the final aggregations so the natural-language-to-SQL route is inspectable and reproducible. """ from __future__ import annotations import csv import json import re import sqlite3 from pathlib import Path from openpyxl import load_workbook HERE = Path(__file__).resolve().parent REPO = HERE.parents[1] SOURCE_DIR = ( REPO / "modules/product/products/smart-canteen/review/" "2026-05-29-smart-canteen-15-project-product-evidence-database" ) WORKBOOK = SOURCE_DIR / "sources/2025-2026年项目收入与支出情况汇总表-团餐-无成本.xlsx" NORMALIZED = SOURCE_DIR / "product-line-item-normalized.csv" PROJECT_DB = SOURCE_DIR / "project-product-evidence-db.csv" UNIT_COLUMN_BY_SHEET = { "江苏省委办公厅西康路招待所职工智慧营养健康食堂项目": 5, "海开智慧（北京）科技服务有限公司": 7, "无锡市金城环保炊具设备有限公司": 5, "北京华谊汇加科技发展有限公司": 6, "北京华谊汇加科技发展有限公司 (安装)": 5, "中债金石资产管理有限公司": 6, "北京新合宜商用设备有限公司": None, "江西省人民政府对外联络办公室后勤服务中心": None, "中央网信办": 5, "北京九韶信科科技有限公司": 5, "江苏国信": 5, } # Commercial rows 4 and 5 are one customer project plus a separate installation # revenue row. All other rows remain distinct project contexts. PROJECT_CONTEXT = { "1": "CTX-01", "2": "CTX-02", "3": "CTX-03", "4": "CTX-04", "5": "CTX-04", "6": "CTX-06", "7": "CTX-07", "8": "CTX-08", "9": "CTX-09", "10": "CTX-10", "11": "CTX-11", } SOFTWARE_MODULES = { "MOD-01": "智慧餐饮运营与交易结算", "MOD-02": "食品安全监管、采购与进销存", "MOD-03": "营养健康分析与菜品展示", "MOD-04": "移动端、小程序与商城用户服务", "MOD-05": "系统集成、数据同步与消费策略", } EXTRA_SOFTWARE_EVIDENCE = [ # Bundled sub-lines in the workbook are not present in the normalized CSV # because they have no separate price. They still prove functional scope. ("6", "中债金石资产管理有限公司", 4, "技术服务费", "系统集成、数据同步与消费策略"), ("6", "中债金石资产管理有限公司", 5, "AI算法+前端软件", "智慧餐饮运营与交易结算"), ("6", "中债金石资产管理有限公司", 6, "科脉智海鲸商业管理软件", "智慧餐饮运营与交易结算"), ("6", "中债金石资产管理有限公司", 7, "科脉·店务通移动管理软件", "移动端、小程序与商城用户服务"), ("6", "中债金石资产管理有限公司", 9, "私有云软件对接定制", "系统集成、数据同步与消费策略"), ("6", "中债金石资产管理有限公司", 10, "小程序定制", "移动端、小程序与商城用户服务"), ] EXTRA_HARDWARE_EVIDENCE = [ # Bundled physical components with no separate sales price were omitted by # the earlier normalized commercial table. Quantities remain explicit in # the source workbook and are added as lower-bound hardware evidence. ("4", "北京华谊汇加科技发展有限公司", 4, "服务器", "联想", "ST45V3", 3, "套"), ("4", "北京华谊汇加科技发展有限公司", 6, "智能视频管理平台服务器", "海康威视", "iVMS-9000N-S5/C300", 3, "套"), ("6", "中债金石资产管理有限公司", 8, "读卡器", "浩顺", "F00U", 1, "台"), ] HARDWARE_RULES = [ ("前厅交易与身份设备", "称重、收银与消费终端", r"绑盘机|称重结算|扫码秤|收款机|安卓POS|视觉结算台|打餐机|消费机|收银称|收银机"), ("前厅交易与身份设备", "读卡、发卡与考勤设备", r"读卡器|发卡器|人脸识别考勤机"), ("后厨食安与物联网", "验收、收货与食品检测设备", r"验货机|验收秤|收货秤|食品安全检测|农残检测"), ("后厨食安与物联网", "晨检设备", r"晨检"), ("后厨食安与物联网", "留样设备", r"留样"), ("后厨食安与物联网", "温湿度与中心温度设备", r"温湿度|温度计|温度监测"), ("后厨食安与物联网", "库管手持设备", r"库管手持机|库存手持机"), ("视频与AI分析", "摄像与视频采集设备", r"摄像头|摄像机|枪机|半球|旋转摄像机"), ("视频与AI分析", "AI边缘计算与分析设备", r"边缘计算主机|边缘计算设备|抓拍盒子|AI分析服务器"), ("营养与信息展示", "营养与健康终端", r"膳食营养分析指导一体机|智能体测仪"), ("营养与信息展示", "电子价签与显示终端", r"营养价签屏|液晶显示屏|菜品电子签|可视化大屏|信息发布屏|电视盒子"), ("计算、网络与存储", "服务器与控制主机", r"管理平台服务器|智能视频管理平台服务器|视觉服务器|门禁消费服务器|控制系统电脑|^服务器$|^电脑$"), ("计算、网络与存储", "交换机与路由设备", r"交换机|路由"), ("计算、网络与存储", "硬盘与录像存储设备", r"硬盘录像机|NVR|16路双盘|硬盘4TB|8T硬盘|^硬盘$|硬盘（"), ("计算、网络与存储", "机柜、配线与电源附件", r"(?:^|\dU)机柜|机柜（|配线架|理线架|机柜插排|信息插座|监控支架|视频处理器"), ("餐线与配套设备", "保温餐炉", r"布菲炉|保温餐炉"), ("餐线与配套设备", "打印设备", r"打印机"), ("餐线与配套设备", "其他可计数配套设备", r"网线钳|VGA接口转接器|^插排$|音响系统|^控制系统$"), ] BULK_RULES = [ ("托盘与餐具", r"托盘|餐托"), ("线缆与光纤", r"网线|电源线|光缆|光纤|跳线|尾纤"), ("LED屏体面积", r"LED显示屏体"), ("GOB镀膜面积", r"GOB镀膜"), ("LED配件计价面积", r"配件包"), ("LED结构装饰计价面积", r"结构及边框装饰"), ("耗材与辅材", r"耗材|辅材|网头|耦合器"), ] SERVICE_PATTERNS = re.compile( r"实施费|技术服务费|硬件维保|设备对接|云服务器服务|施工费|安装实施|安装调式|调试|预埋安装费|外部接口|总价|^\.\.\.$" ) SOFTWARE_PATTERNS = re.compile( r"管理系统|管理平台|进销存管理|监管平台|监控平台|采购比价系统软件|软件系统|食安展示系统|移动端应用|第三方对接|对接线上商城|消费策略定制开发|智慧餐饮管理|消费结算软件|软件优化升级服务" ) 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, fieldnames: list[str], rows: list[dict]) -> None: with path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) def numeric(value: str | None) -> float | None: if value is None or str(value).strip() == "": return None try: return float(value) except ValueError: return None def software_matches(row: dict[str, str]) -> list[tuple[str, str]]: text = " | ".join( [row.get("item_name", ""), row.get("sub_item", ""), row.get("spec", ""), row.get("product_category", "")] ) matches: list[tuple[str, str]] = [] category = row.get("product_category", "") if category == "前厅交易结算系统" or re.search(r"智慧食堂管理|智慧餐厅管理|智慧餐饮管理|团餐管理|消费结算软件|软件优化升级", text): matches.append(("MOD-01", "explicit_software" if SOFTWARE_PATTERNS.search(text) else "hardware_implied")) if category == "后厨食安进销存系统" or re.search(r"食安|进销存|采购比价|安全监管", text): matches.append(("MOD-02", "explicit_software" if SOFTWARE_PATTERNS.search(text) else "hardware_implied")) if category == "营养健康与菜品展示系统" or re.search(r"膳食|营养|菜品展示|电子签", text): matches.append(("MOD-03", "explicit_software" if SOFTWARE_PATTERNS.search(text) else "hardware_implied")) if re.search(r"手机端|H5|移动端|小程序|线上商城", text, re.I): matches.append(("MOD-04", "explicit_software")) if re.search(r"对接|数据同步|消费策略", text): matches.append(("MOD-05", "explicit_software")) return list(dict.fromkeys(matches)) def classify_physical(row: dict[str, str]) -> tuple[str, str, str]: name = row.get("item_name", "").strip() sub_item = row.get("sub_item", "").strip() combined = " | ".join(value for value in (name, sub_item) if value) unit = row.get("unit", "").strip() if SERVICE_PATTERNS.search(name): return "service", "", "" # A priced system row can contain a bundled physical component in sub_item, # and names such as "management platform server" are hardware despite the # word "platform". Prefer explicit physical matches before software. for group, category, pattern in HARDWARE_RULES: if re.search(pattern, combined): return "hardware", group, category if SOFTWARE_PATTERNS.search(name): return "software", "", "" for category, pattern in BULK_RULES: if re.search(pattern, combined): return "bulk", "批量材料与餐具", category if name == "PVC": return "bulk", "批量材料与餐具", "耗材与辅材" if unit in {"台", "个", "套", "块", "条", "把", "盒", "根", "对"}: return "hardware", "其他硬件", "待人工归类硬件" if unit in {"米", "平米", "箱", "卷", "批"}: return "bulk", "批量材料与餐具", "待人工归类材料" return "unknown", "", "" def main() -> None: workbook = load_workbook(WORKBOOK, data_only=True, read_only=True) normalized_rows = read_csv(NORMALIZED) project_rows = read_csv(PROJECT_DB) enriched: list[dict] = [] software_evidence: list[dict] = [] for row in normalized_rows: sheet = row["detail_sheet"] unit_column = UNIT_COLUMN_BY_SHEET[sheet] unit = "" if unit_column: unit = workbook[sheet].cell(int(row["source_row"]), unit_column).value or "" quantity = numeric(row.get("quantity")) enriched_row = dict(row) enriched_row["project_context_id"] = PROJECT_CONTEXT[row["project_seq"]] enriched_row["unit"] = str(unit).strip() enriched_row["quantity_numeric"] = quantity record_type, hardware_group, hardware_category = classify_physical(enriched_row) enriched_row["record_type"] = record_type enriched_row["hardware_group"] = hardware_group enriched_row["hardware_category"] = hardware_category enriched.append(enriched_row) for module_id, evidence_kind in software_matches(enriched_row): software_evidence.append( { "module_id": module_id, "module_name": SOFTWARE_MODULES[module_id], "project_context_id": enriched_row["project_context_id"], "project_seq": row["project_seq"], "project": row["project"], "evidence_item": row["item_name"], "evidence_kind": evidence_kind, "source_sheet": sheet, "source_row": row["source_row"], } ) for project_seq, sheet, source_row, item, module_name in EXTRA_SOFTWARE_EVIDENCE: module_id = next(mid for mid, name in SOFTWARE_MODULES.items() if name == module_name) project = next(row["project"] for row in normalized_rows if row["project_seq"] == project_seq) software_evidence.append( { "module_id": module_id, "module_name": module_name, "project_context_id": PROJECT_CONTEXT[project_seq], "project_seq": project_seq, "project": project, "evidence_item": item, "evidence_kind": "bundled_software", "source_sheet": sheet, "source_row": source_row, } ) for project_seq, sheet, source_row, item, brand, spec, quantity, unit in EXTRA_HARDWARE_EVIDENCE: project_row = next(row for row in normalized_rows if row["project_seq"] == project_seq) synthetic = { "project_context_id": PROJECT_CONTEXT[project_seq], "project_seq": project_seq, "customer": project_row["customer"], "project": project_row["project"], "detail_sheet": sheet, "source_row": str(source_row), "item_name": item, "sub_item": "", "brand": brand, "spec": spec, "quantity": str(quantity), "quantity_numeric": float(quantity), "unit": unit, "unit_price": "", "sales_total": "", "product_category": "硬件集成与基础设施", "record_type": "hardware", "hardware_group": "", "hardware_category": "", } _, synthetic["hardware_group"], synthetic["hardware_category"] = classify_physical(synthetic) enriched.append(synthetic) # Deduplicate repeated matches while preserving separate source rows. dedup = {} for row in software_evidence: key = (row["module_id"], row["project_context_id"], row["evidence_item"], row["source_sheet"], row["source_row"]) dedup[key] = row software_evidence = list(dedup.values()) con = sqlite3.connect(":memory:") con.row_factory = sqlite3.Row con.execute( """ CREATE TABLE line_items ( project_context_id TEXT, project_seq INTEGER, customer TEXT, project TEXT, item_name TEXT, sub_item TEXT, brand TEXT, spec TEXT, quantity REAL, unit TEXT, sales_total REAL, product_category TEXT, record_type TEXT, hardware_group TEXT, hardware_category TEXT ) """ ) con.executemany( "INSERT INTO line_items VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", [ ( row["project_context_id"], int(row["project_seq"]), row["customer"], row["project"], row["item_name"], row["sub_item"], row["brand"], row["spec"], row["quantity_numeric"], row["unit"], numeric(row["sales_total"]), row["product_category"], row["record_type"], row["hardware_group"], row["hardware_category"], ) for row in enriched ], ) con.execute( """ CREATE TABLE software_module_evidence ( module_id TEXT, module_name TEXT, project_context_id TEXT, project_seq INTEGER, project TEXT, evidence_item TEXT, evidence_kind TEXT, source_sheet TEXT, source_row INTEGER ) """ ) con.executemany( "INSERT INTO software_module_evidence VALUES (?,?,?,?,?,?,?,?,?)", [tuple(row[key] for key in ["module_id", "module_name", "project_context_id", "project_seq", "project", "evidence_item", "evidence_kind", "source_sheet", "source_row"]) for row in software_evidence], ) queries = { "software_modules": """ SELECT module_id, module_name, COUNT(DISTINCT project_context_id) AS project_coverage_count, ROUND(100.0 * COUNT(DISTINCT project_context_id) / 10, 1) AS project_coverage_pct, COUNT(DISTINCT evidence_item) AS evidence_item_count, SUM(CASE WHEN evidence_kind = 'explicit_software' OR evidence_kind = 'bundled_software' THEN 1 ELSE 0 END) AS explicit_evidence_rows, SUM(CASE WHEN evidence_kind = 'hardware_implied' THEN 1 ELSE 0 END) AS hardware_implied_rows FROM software_module_evidence GROUP BY module_id, module_name ORDER BY project_coverage_count DESC, module_id """, "hardware_categories": """ SELECT hardware_group, hardware_category, COUNT(DISTINCT project_context_id) AS project_coverage_count, COUNT(*) AS source_line_count, ROUND(SUM(quantity), 2) AS observed_quantity, SUM(CASE WHEN quantity IS NULL THEN 1 ELSE 0 END) AS missing_quantity_lines, GROUP_CONCAT(DISTINCT CASE WHEN unit = '' THEN '原表未填' ELSE unit END) AS units FROM line_items WHERE record_type = 'hardware' GROUP BY hardware_group, hardware_category ORDER BY observed_quantity DESC, hardware_group, hardware_category """, "hardware_groups": """ SELECT hardware_group, COUNT(DISTINCT project_context_id) AS project_coverage_count, COUNT(DISTINCT hardware_category) AS category_count, COUNT(*) AS source_line_count, ROUND(SUM(quantity), 2) AS observed_quantity, SUM(CASE WHEN quantity IS NULL THEN 1 ELSE 0 END) AS missing_quantity_lines FROM line_items WHERE record_type = 'hardware' GROUP BY hardware_group ORDER BY observed_quantity DESC """, "bulk_quantities": """ SELECT hardware_category AS bulk_category, unit, COUNT(DISTINCT project_context_id) AS project_coverage_count, COUNT(*) AS source_line_count, ROUND(SUM(quantity), 2) AS observed_quantity, SUM(CASE WHEN quantity IS NULL THEN 1 ELSE 0 END) AS missing_quantity_lines FROM line_items WHERE record_type = 'bulk' GROUP BY hardware_category, unit ORDER BY hardware_category, unit """, "hardware_items": """ SELECT hardware_group, hardware_category, item_name, COUNT(DISTINCT project_context_id) AS project_coverage_count, ROUND(SUM(quantity), 2) AS observed_quantity, GROUP_CONCAT(DISTINCT CASE WHEN unit = '' THEN '原表未填' ELSE unit END) AS units, GROUP_CONCAT(DISTINCT brand) AS brands FROM line_items WHERE record_type = 'hardware' GROUP BY hardware_group, hardware_category, item_name ORDER BY hardware_group, hardware_category, observed_quantity DESC """, "record_profile": """ SELECT record_type, COUNT(*) AS line_count, COUNT(DISTINCT project_context_id) AS project_coverage_count, SUM(CASE WHEN quantity IS NULL THEN 1 ELSE 0 END) AS missing_quantity_lines, ROUND(SUM(COALESCE(quantity, 0)), 2) AS raw_quantity_sum FROM line_items GROUP BY record_type ORDER BY line_count DESC """, } query_outputs: dict[str, list[dict]] = {} for name, sql in queries.items(): query_outputs[name] = [dict(row) for row in con.execute(sql)] if query_outputs[name]: write_csv(HERE / f"{name}.csv", list(query_outputs[name][0]), query_outputs[name]) write_csv( HERE / "software_module_evidence.csv", list(software_evidence[0]), sorted(software_evidence, key=lambda row: (row["module_id"], row["project_context_id"], int(row["source_row"]))), ) enriched_fields = [ "project_context_id", "project_seq", "customer", "project", "detail_sheet", "source_row", "item_name", "sub_item", "brand", "spec", "quantity", "unit", "unit_price", "sales_total", "product_category", "record_type", "hardware_group", "hardware_category", ] write_csv(HERE / "line_items_enriched.csv", enriched_fields, [{key: row.get(key, "") for key in enriched_fields} for row in enriched]) complete_projects = [row for row in project_rows if row["review_id"] in {f"PR-{i:02d}" for i in range(1, 12)}] reconciled = sum(1 for row in complete_projects if row["amount_reconciliation"] == "一致") missing_units = sum(1 for row in enriched if not row["unit"] and row["record_type"] in {"hardware", "bulk"}) missing_quantities = sum(1 for row in enriched if row["quantity_numeric"] is None and row["record_type"] in {"hardware", "bulk"}) hardware_lines = sum(1 for row in enriched if row["record_type"] == "hardware") unknown_lines = sum(1 for row in enriched if row["record_type"] == "unknown") profile = { "source_commercial_records": 11, "consolidated_project_contexts": 10, "planned_project_slots": 15, "unfilled_project_slots": 4, "source_normalized_line_items": len(normalized_rows), "recovered_bundled_hardware_lines": len(EXTRA_HARDWARE_EVIDENCE), "analysis_line_items": len(enriched), "hardware_source_lines": hardware_lines, "unknown_source_lines": unknown_lines, "hardware_or_bulk_lines_missing_unit": missing_units, "hardware_or_bulk_lines_missing_quantity": missing_quantities, "amount_reconciled_records": reconciled, "amount_exception_records": 11 - reconciled, "analysis_window": "2025-01-22 to 2026-02-02", "quantity_interpretation": "Observed source quantities; hardware totals exclude software/services and keep bulk/linear/area measures separate.", } (HERE / "analysis_profile.json").write_text(json.dumps(profile, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") query_sections = [ "-- SQLite dialect. Tables are materialized in-memory by analyze_portfolio.py.", *[f"-- {name}\n{sql.strip()};" for name, sql in queries.items()], ] (HERE / "analysis_queries.sql").write_text("\n\n".join(query_sections) + "\n", encoding="utf-8") print(json.dumps({"profile": profile, "outputs": sorted(path.name for path in HERE.glob("*.csv"))}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()