#!/usr/bin/env python3 """Build a traceable 28-project portfolio and query reachable production DBs read-only. Only aggregate project-level metrics are written. Credentials, people, order rows, and production payloads are never persisted. """ from __future__ import annotations import csv import datetime as dt import importlib.util import json import subprocess from pathlib import Path from typing import Any from urllib.parse import urlencode from urllib.request import urlopen ROOT = Path(__file__).resolve().parents[3] OUT = Path(__file__).resolve().parent STORE_REPO = Path("/Users/jack/code/010-cpt/008-zhct/zhct/zhctproject/store") HELPER_PATH = ROOT / "work_store/data-queries/all-project-dau-dashboard/refresh_all_project_dau_dashboard.py" REGISTRY_PATH = ROOT / "work_store/data-queries/all-project-dau-dashboard/project-registry.csv" DB_CORE_PATH = OUT / "db-core-refresh/all-project-dau.csv" ACTIVE_PATH = ROOT / "work/2026-07-29-smart-canteen-three-agent-plan/recent-project-unified-analysis.csv" FUTURE_PATH = ROOT / "work/2026-06-10-smart-canteen-project-layered-report/future-project-layered-map.csv" CENTRAL_URL = "https://zhct.yyangpt.cn/api/dataCenterDashboard" AS_OF = dt.date(2026, 8, 3) CURRENT_START = AS_OF - dt.timedelta(days=29) PREVIOUS_END = CURRENT_START - dt.timedelta(days=1) PREVIOUS_START = PREVIOUS_END - dt.timedelta(days=29) ORDER_SOURCE_LABELS = { 1: "绑盘机", 2: "消费机", 3: "虚拟订单", 4: "线上订餐", 5: "外部订单接入", 6: "闸机", 10: "普通订单", } PAY_TYPE_LABELS = {10: "线上支付", 20: "刷卡", 30: "消费码", 40: "刷脸"} MEAL_LABELS = {1: "早餐", 2: "午餐", 3: "晚餐", 4: "夜宵"} FEATURE_TABLES = { "ydy_meal_order_refund": ("退款申请", "apply_date", "is_delete=0", "主动业务动作"), "ydy_order_evaluate": ("就餐评价", "create_time", "is_delete=0", "主动业务动作"), "ydy_meal_order_pickup": ("订餐取餐流程", "meal_date", "1=1", "业务流程记录"), "ydy_menus_subscribe": ("餐单预约", "menu_date", "1=1", "主动业务动作"), "ydy_menus_attendance": ("就餐考勤", "menu_date", "1=1", "业务流程记录"), "ydy_recharge_order": ("线上充值", "recharge_date", "is_delete=0", "主动业务动作"), "ydy_ai_conversation": ("AI营养对话", "create_time", "1=1", "主动业务动作"), "ydy_measurement_session": ("健康测量会话", "start_time", "1=1", "主动业务动作"), "ydy_health_record": ("健康数据记录", "start_time", "1=1", "业务流程记录"), "ydy_sign_daka": ("签到打卡", "daka_date", "status=1", "主动业务动作"), "ydy_export_log": ("后台报表导出", "create_time", "status=1", "主动业务动作"), "ydy_consume_limit_block_log": ("消费限额拦截", "consume_time", "1=1", "规则触发记录"), "ydy_security_audit_log": ("后台安全审计操作", "create_time", "is_delete=0", "业务流程记录"), "ydy_staff_card": ("实体卡建档", "create_time", "is_delete=0", "配置/建档"), "ydy_staff_face": ("人脸建档", "create_time", "1=1", "配置/建档"), "ydy_device_weight_event": ("称重设备事件", "event_time", "is_delete=0", "业务流程记录"), "ydy_purchase": ("采购单", "create_date", "is_delete=0", "主动业务动作"), "ydy_billing_record": ("账单缴费", "issue_time", "is_delete=0", "主动业务动作"), "ydy_locker_pickup_api_log": ("取餐柜接口", "request_time", "1=1", "业务流程记录"), "ydy_vending_shipment_result": ("售货机出货", "create_time", "1=1", "业务流程记录"), } # The first 20 are the live central project dimension on 2026-08-04. Five # additional active projects and three future projects rebuild a traceable # 28-candidate portfolio; this is explicitly not a canonical master-data claim. PROJECTS = [ ("P001", "三全智慧营养健康餐厅", "三全项目|三全集团", "zhct_sanquan", "中央统计项目维度", "中央已登记"), ("P002", "产业园智慧营养健康餐厅", "产业园", "zhct", "中央统计项目维度", "中央已登记"), ("P003", "人大附中", "人大附中项目", "zhct_rdfz", "中央统计项目维度", "中央已登记"), ("P004", "四川射洪", "射洪", "zhct_sh", "中央统计项目维度", "中央已登记"), ("P005", "城市副中心", "城市副中心项目", "zhct_csfzx", "中央统计项目维度", "中央已登记"), ("P006", "多样人群营养支持健康教育系统", "", "zhct_gfkj", "中央统计项目维度", "中央已登记"), ("P007", "山西焦煤", "山西焦煤项目", "zhct_sxjm", "中央统计项目维度", "中央已登记"), ("P008", "无锡新吴区区政府", "新吴区", "zhct_xwq", "中央统计项目维度", "中央已登记"), ("P009", "机场", "首都机场|机场项目", "zhct_jc", "中央统计项目维度", "中央已登记"), ("P010", "江苏国信", "国信项目", "zhct_guoxin", "中央统计项目维度", "中央已登记"), ("P011", "江西206", "江西206项目", "zhct_jx206", "中央统计项目维度", "中央已登记"), ("P012", "海开集团", "海开项目", "zhct_haikai", "中央统计项目维度", "中央已登记"), ("P013", "湖南体职院", "湖南体院", "zhct_hntzy", "中央统计项目维度", "中央已登记"), ("P014", "滨州健康科技职业学院", "滨州项目", "zhct_bzjk", "中央统计项目维度", "中央已登记"), ("P015", "莱蒂森", "莱迪森|南京建宁中学", "zhct_lds", "中央统计项目维度", "中央已登记"), ("P016", "西康宾馆", "西康项目", "zhct_xkbg", "中央统计项目维度", "中央已登记"), ("P017", "赛迪物业", "赛迪|赛迪项目", "zhct_saidi", "中央统计项目维度", "中央已登记"), ("P018", "重庆环卫", "重庆环卫项目", "zhct_cqhw", "中央统计项目维度", "中央已登记"), ("P019", "金斯瑞", "金斯瑞项目", "zhct_jsr", "中央统计项目维度", "中央已登记"), ("P020", "首通智城", "首通智城项目", "zhct_stzc", "中央统计项目维度", "中央已登记"), ("P021", "国康项目", "国康医院", "", "近期项目线程", "有项目证据，未映射中央项目码"), ("P022", "四方达项目", "四方达", "", "近期项目线程", "有项目证据，未映射中央项目码"), ("P023", "网信办项目", "网信办", "", "近期项目线程", "有项目证据，未映射中央项目码"), ("P024", "大兴项目", "大兴", "", "近期项目线程", "有项目证据，未映射中央项目码"), ("P025", "安徽部队项目", "安徽部队", "", "近期项目线程", "有项目证据，未映射中央项目码"), ("P026", "中煤", "中煤项目", "", "未来项目分层图", "规划/售前证据"), ("P027", "保康三中", "保康三中项目", "", "未来项目分层图", "规划/售前证据"), ("P028", "C08", "C08项目", "", "未来项目分层图", "规划/售前证据"), ] def load_helper(): spec = importlib.util.spec_from_file_location("all_project_helper", HELPER_PATH) if not spec or not spec.loader: raise RuntimeError("helper_load_failed") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module 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(name: str, rows: list[dict[str, Any]]) -> None: path = OUT / name fields: list[str] = [] for row in rows: for key in row: if key not in fields: fields.append(key) with path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows(rows) def sanitize_db_core(rows: list[dict[str, str]]) -> list[dict[str, str]]: """Remove local access surfaces before persisting the report snapshot.""" keep = [ "project_key", "project_name", "data_status", "status_detail", "snapshot_at", "database_name", "registered_users", "all_orders", "all_amount", "valid_orders", "valid_amount", "first_meal_date", "last_meal_date", "today_dau", "latest_valid_meal_date", "latest_day_dau", "dau7_avg", "dau7_peak", "mau7", "dau30_avg", "dau30_peak", "mau30", "orders30", "amount30", ] return [{field: row.get(field, "") for field in keep} for row in rows] def to_number(value: str) -> int | float | str: if value in ("", "NULL", None): return 0 try: number = float(value) except (ValueError, TypeError): return value return int(number) if number.is_integer() else number def parse_rows(raw: str, fields: list[str]) -> list[dict[str, Any]]: return [ {field: to_number(values[index]) if index < len(values) else "" for index, field in enumerate(fields)} for line in raw.splitlines() if line for values in [line.split("\t")] ] def fetch_central(start: dt.date, end: dt.date) -> dict[str, Any]: url = f"{CENTRAL_URL}?{urlencode({'date_from': start.isoformat(), 'date_to': end.isoformat()})}" with urlopen(url, timeout=30) as response: return json.load(response) def config_for(helper: Any, registry_row: dict[str, str]) -> tuple[dict[str, str] | None, str]: config, source = helper.load_config(STORE_REPO, registry_row["config_dir"]) if registry_row["project_key"] == "zhct_rdfz" and not config: config, fallback = helper.load_config(STORE_REPO, "zhct_guoxin") if config: config["database"] = "zhct_rdfz" source = f"same_rds_config_fallback:{fallback}" if registry_row["project_key"] == "zhct_jsr" and config: config = helper.jsr_readonly_config(config["database"]) source = "jsr_readonly_env" return config, source def safe_query(helper: Any, binary: str, config: dict[str, str], sql: str, timeout: int = 45) -> str: statement = ( "SET SESSION TRANSACTION READ ONLY; START TRANSACTION WITH CONSISTENT SNAPSHOT; " f"{sql.rstrip().rstrip(';')}; ROLLBACK;" ) return run_with_retry(helper, binary, config, statement, timeout) def run_with_retry( helper: Any, binary: str, config: dict[str, str], sql: str, timeout: int = 45, attempts: int = 2, ) -> str: last_error: Exception | None = None for _ in range(attempts): try: return helper.run_mysql(binary, config, sql, timeout) except (RuntimeError, subprocess.TimeoutExpired) as exc: last_error = exc assert last_error is not None raise last_error def build_master( central_current: dict[str, Any], db_core: list[dict[str, str]], registry: list[dict[str, str]], active: list[dict[str, str]], future: list[dict[str, str]], ) -> list[dict[str, Any]]: central_rows = {row["project_code"]: row for row in central_current["data"]["project_analysis"]} db_by_key = {row["project_key"]: row for row in db_core} registry_keys = {row["project_key"] for row in registry} active_by_name = {row["project_name"]: row for row in active} future_by_name = {row["project"]: row for row in future} active_alias = { "国康项目": "国康项目", "四方达项目": "四方达项目", "网信办项目": "网信办项目", "大兴项目": "大兴项目", "安徽部队项目": "安徽部队项目", } rows: list[dict[str, Any]] = [] for project_id, name, aliases, central_code, source, inclusion_state in PROJECTS: central = central_rows.get(central_code, {}) db = db_by_key.get(central_code, {}) active_row = active_by_name.get(active_alias.get(name, ""), {}) future_row = future_by_name.get(name, {}) if central_code and central_code in registry_keys: database_state = db.get("data_status", "未查询") elif central_code: database_state = "未进入17项目数据库注册表" else: database_state = "未建立数据库映射" evidence_level = active_row.get("evidence_level") or future_row.get("evidence_level") or ("A" if central else "B") rows.append({ "project_id": project_id, "project_name": name, "aliases": aliases, "inclusion_source": source, "inclusion_state": inclusion_state, "central_project_code": central_code, "central_registered": "yes" if central else "no", "central_operating_status": central.get("operating_status", ""), "central_sync_status": central.get("sync_status", ""), "db_registry_registered": "yes" if central_code in registry_keys else "no", "database_state": database_state, "database_detail": db.get("status_detail", ""), "current_phase": active_row.get("current_phase", ""), "project_health": active_row.get("health", ""), "primary_blocker": active_row.get("primary_blocker", ""), "future_position": future_row.get("project_position", ""), "evidence_level": evidence_level, "usage_analysis_boundary": ( "生产库可核验" if database_state == "live" else "数据库不可达，不得解释为零使用" if database_state == "unavailable" else "仅中央聚合可见" if central else "仅项目/规划证据，不判断线上使用" ), }) return rows def analyze_databases( helper: Any, registry: list[dict[str, str]], db_core: list[dict[str, str]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: binary = helper.mysql_binary() core_by_key = {row["project_key"]: row for row in db_core} order_periods: list[dict[str, Any]] = [] channel_mix: list[dict[str, Any]] = [] payment_mix: list[dict[str, Any]] = [] meal_mix: list[dict[str, Any]] = [] feature_activity: list[dict[str, Any]] = [] dimensions = [ ("source", ORDER_SOURCE_LABELS, "channel", channel_mix), ("pay_type", PAY_TYPE_LABELS, "payment", payment_mix), ("meal_times", MEAL_LABELS, "meal", meal_mix), ] for registry_row in registry: project_key = registry_row["project_key"] project_name = registry_row["project_name"] if core_by_key.get(project_key, {}).get("data_status") != "live": continue config, source = config_for(helper, registry_row) if not config: continue try: tables = set(run_with_retry(helper, binary, config, "SHOW TABLES;", 30).splitlines()) columns = { line.split("\t", 1)[0] for line in run_with_retry(helper, binary, config, "SHOW COLUMNS FROM ydy_meal_order;", 30).splitlines() if line } except (RuntimeError, subprocess.TimeoutExpired): continue valid_parts = ["user_id>0"] for field, condition in (("pay_status", "pay_status=20"), ("order_status", "order_status=30"), ("is_delete", "is_delete=0")): if field in columns: valid_parts.append(condition) valid = " AND ".join(valid_parts) periods = [("近30天", CURRENT_START, AS_OF), ("前30天", PREVIOUS_START, PREVIOUS_END)] for period, start, end in periods: try: raw = safe_query(helper, binary, config, f""" SELECT COUNT(*), ROUND(COALESCE(SUM(total_price),0),2), COUNT(DISTINCT user_id), COUNT(DISTINCT meal_date), ROUND(COUNT(*)/NULLIF(COUNT(DISTINCT user_id),0),2) FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{start}' AND '{end}' """) value = parse_rows(raw, ["orders", "amount", "active_users", "active_days", "orders_per_user"])[0] order_periods.append({ "project_key": project_key, "project_name": project_name, "period": period, "start_date": start.isoformat(), "end_date": end.isoformat(), "config_source": source, **value, }) except (RuntimeError, subprocess.TimeoutExpired, IndexError): continue for field, labels, label_field, target in dimensions: if field not in columns: continue try: raw = safe_query(helper, binary, config, f""" SELECT {field},COUNT(*),COUNT(DISTINCT user_id),ROUND(COALESCE(SUM(total_price),0),2) FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{start}' AND '{end}' GROUP BY {field} ORDER BY COUNT(*) DESC """) except (RuntimeError, subprocess.TimeoutExpired): continue parsed = parse_rows(raw, ["dimension_value", "orders", "active_users", "amount"]) total = sum(int(row["orders"]) for row in parsed) for row in parsed: try: code = int(row["dimension_value"]) except (TypeError, ValueError): code = -1 target.append({ "project_key": project_key, "project_name": project_name, "period": period, label_field: labels.get(code, f"其他({row['dimension_value']})"), **row, "order_share": round(int(row["orders"]) / total, 6) if total else 0, }) for table, (feature, date_field, predicate, evidence_type) in FEATURE_TABLES.items(): if table not in tables: feature_activity.append({ "project_key": project_key, "project_name": project_name, "feature": feature, "evidence_type": evidence_type, "source_table": table, "table_state": "table_missing", "all_time_records": "", "current_30d_records": "", "previous_30d_records": "", }) continue try: raw = safe_query(helper, binary, config, f""" SELECT COUNT(*), COALESCE(SUM(({date_field}) BETWEEN '{CURRENT_START}' AND '{AS_OF} 23:59:59'),0), COALESCE(SUM(({date_field}) BETWEEN '{PREVIOUS_START}' AND '{PREVIOUS_END} 23:59:59'),0) FROM {table} WHERE {predicate} """) value = parse_rows(raw, ["all_time_records", "current_30d_records", "previous_30d_records"])[0] state = "queried" except (RuntimeError, subprocess.TimeoutExpired, IndexError): value = {"all_time_records": "", "current_30d_records": "", "previous_30d_records": ""} state = "query_failed" feature_activity.append({ "project_key": project_key, "project_name": project_name, "feature": feature, "evidence_type": evidence_type, "source_table": table, "table_state": state, **value, }) return order_periods, channel_mix, payment_mix, meal_mix, feature_activity def main() -> None: OUT.mkdir(parents=True, exist_ok=True) current = fetch_central(CURRENT_START, AS_OF) previous = fetch_central(PREVIOUS_START, PREVIOUS_END) (OUT / "central-current30.json").write_text(json.dumps(current, ensure_ascii=False, indent=2), encoding="utf-8") (OUT / "central-previous30.json").write_text(json.dumps(previous, ensure_ascii=False, indent=2), encoding="utf-8") registry = read_csv(REGISTRY_PATH) db_core = read_csv(DB_CORE_PATH) db_core_safe = sanitize_db_core(db_core) active = read_csv(ACTIVE_PATH) future = read_csv(FUTURE_PATH) master = build_master(current, db_core, registry, active, future) write_csv("project-master.csv", master) write_csv("db-project-status.csv", db_core_safe) central_current_rows = [] previous_by_code = {row["project_code"]: row for row in previous["data"]["project_analysis"]} for row in current["data"]["project_analysis"]: prev = previous_by_code.get(row["project_code"], {}) central_current_rows.append({ "project_code": row["project_code"], "project_name": row["project_name"], "operating_status": row["operating_status"], "sync_status": row["sync_status"], "user_count": row["user_count"], "current30_orders": row["order_count"], "current30_amount": row["order_amount"], "previous30_orders": prev.get("order_count", 0), "previous30_amount": prev.get("order_amount", 0), }) write_csv("central-project-summary.csv", central_current_rows) helper = load_helper() order_periods, channels, payments, meals, features = analyze_databases(helper, registry, db_core) live_keys = {row["project_key"] for row in db_core if row["data_status"] == "live"} for _ in range(2): period_counts: dict[str, int] = {} for row in order_periods: period_counts[row["project_key"]] = period_counts.get(row["project_key"], 0) + 1 missing_keys = {key for key in live_keys if period_counts.get(key, 0) < 2} if not missing_keys: break order_periods = [row for row in order_periods if row["project_key"] not in missing_keys] channels = [row for row in channels if row["project_key"] not in missing_keys] payments = [row for row in payments if row["project_key"] not in missing_keys] meals = [row for row in meals if row["project_key"] not in missing_keys] features = [row for row in features if row["project_key"] not in missing_keys] retry_registry = [row for row in registry if row["project_key"] in missing_keys] retried = analyze_databases(helper, retry_registry, db_core) order_periods.extend(retried[0]) channels.extend(retried[1]) payments.extend(retried[2]) meals.extend(retried[3]) features.extend(retried[4]) write_csv("db-order-periods.csv", order_periods) write_csv("db-channel-mix.csv", channels) write_csv("db-payment-mix.csv", payments) write_csv("db-meal-mix.csv", meals) write_csv("db-feature-activity.csv", features) outputs = { "project_master": master, "central_project_summary": central_current_rows, "db_core": db_core_safe, "db_order_periods": order_periods, "db_channel_mix": channels, "db_payment_mix": payments, "db_meal_mix": meals, "db_feature_activity": features, } snapshot = { "generated_at": dt.datetime.now().astimezone().isoformat(timespec="seconds"), "as_of": AS_OF.isoformat(), "periods": { "current30": [CURRENT_START.isoformat(), AS_OF.isoformat()], "previous30": [PREVIOUS_START.isoformat(), PREVIOUS_END.isoformat()], }, "source_counts": { "reconstructed_candidates": len(master), "central_registered": sum(row["central_registered"] == "yes" for row in master), "db_registry": sum(row["db_registry_registered"] == "yes" for row in master), "db_live": sum(row["database_state"] == "live" for row in master), }, "datasets": outputs, } (OUT / "analysis-snapshot.json").write_text(json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"row_counts": {key: len(value) for key, value in outputs.items()}}, ensure_ascii=False)) if __name__ == "__main__": main()