#!/usr/bin/env python3 """Read-only aggregate feature-usage analysis for Saidi and Capital Airport. The script intentionally emits only aggregate data. Database credentials are loaded through the existing ignored production configuration and are never written to the output directory. """ from __future__ import annotations import csv import datetime as dt import importlib.util import json from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[3] OUTPUT_DIR = Path(__file__).resolve().parent DAU_SCRIPT = ROOT / "work_store/data-queries/all-project-dau-dashboard/refresh_all_project_dau_dashboard.py" STORE_REPO = Path("/Users/jack/code/010-cpt/008-zhct/zhct/zhctproject/store") 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) TREND_START = AS_OF - dt.timedelta(days=89) PROJECTS = { "airport": { "project_name": "首都机场智慧营养健康餐厅", "config_dir": "zhct_jc", }, "saidi": { "project_name": "北京赛迪物业智慧营养健康餐厅", "config_dir": "zhct_saidi", }, } ORDER_SOURCE_LABELS = { 1: "绑盘机", 2: "消费机", 3: "虚拟订单", 4: "线上订餐", 5: "外部订单接入", 6: "闸机", 10: "普通订单", } PAY_TYPE_LABELS = {10: "线上支付", 20: "刷卡", 30: "消费码", 40: "刷脸"} MEAL_LABELS = {1: "早餐", 2: "午餐", 3: "晚餐", 4: "夜宵"} WEEKDAY_LABELS = {1: "周一", 2: "周二", 3: "周三", 4: "周四", 5: "周五", 6: "周六", 7: "周日"} EXPORT_LABELS = { 1: "消费订单导出", 2: "人员消费报表", 3: "部门消费报表", 4: "设备消费报表", 5: "充值统计", 6: "补贴统计", 7: "菜品销量统计", 8: "餐厅经营汇总", 9: "档口营收结算", 11: "现金退款", } FEATURE_TABLES = { "ydy_meal_order_refund": ("退款申请", "apply_date", "is_delete=0", "user_id", "主动业务动作"), "ydy_order_evaluate": ("就餐评价", "create_time", "is_delete=0", "user_id", "主动业务动作"), "ydy_meal_order_pickup": ("订餐取餐流程", "meal_date", "1=1", "", "业务流程记录"), "ydy_menus_subscribe": ("餐单预约", "menu_date", "1=1", "staff_uuid", "主动业务动作"), "ydy_menus_attendance": ("就餐考勤", "menu_date", "1=1", "staff_uuid", "业务流程记录"), "ydy_recharge_order": ("线上充值", "recharge_date", "is_delete=0", "user_id", "主动业务动作"), "ydy_ai_conversation": ("AI 营养对话", "create_time", "1=1", "user_id", "主动业务动作"), "ydy_measurement_session": ("健康测量会话", "start_time", "1=1", "", "主动业务动作"), "ydy_health_record": ("健康数据记录", "start_time", "1=1", "", "业务流程记录"), "ydy_sign_daka": ("签到打卡", "daka_date", "status=1", "staff_uuid", "主动业务动作"), "ydy_export_log": ("后台报表导出", "create_time", "status=1", "create_by_uuid", "主动业务动作"), "ydy_consume_limit_block_log": ("消费限额拦截", "consume_time", "1=1", "user_id", "规则触发记录"), "ydy_security_audit_log": ("后台安全审计操作", "create_time", "is_delete=0", "operator", "业务流程记录"), "ydy_staff_card": ("实体卡建档", "create_time", "is_delete=0", "staff_uuid", "配置/建档"), "ydy_staff_face": ("人脸建档", "create_time", "1=1", "staff_uuid", "配置/建档"), "ydy_device_weight_event": ("称重设备事件", "event_time", "is_delete=0", "", "业务流程记录"), "ydy_purchase": ("采购单", "create_date", "is_delete=0", "", "主动业务动作"), "ydy_billing_record": ("账单缴费", "issue_time", "is_delete=0", "user_id", "主动业务动作"), "ydy_locker_pickup_api_log": ("取餐柜接口", "request_time", "1=1", "", "业务流程记录"), "ydy_vending_shipment_result": ("售货机出货", "create_time", "1=1", "", "业务流程记录"), } def load_dau_module(): spec = importlib.util.spec_from_file_location("dau_refresh", DAU_SCRIPT) if not spec or not spec.loader: raise RuntimeError("unable_to_load_existing_db_helper") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def as_number(value: str) -> int | float | str: if value == "NULL": return "" try: number = float(value) except (TypeError, ValueError): return value return int(number) if number.is_integer() else number def parse_rows(raw: str, fields: list[str]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for line in raw.splitlines(): if not line: continue values = line.split("\t") rows.append({field: as_number(values[index]) if index < len(values) else "" for index, field in enumerate(fields)}) return rows def write_csv(name: str, rows: list[dict[str, Any]]) -> None: path = OUTPUT_DIR / name if not rows: path.write_text("", encoding="utf-8") return 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 main() -> None: OUTPUT_DIR.mkdir(parents=True, exist_ok=True) helper = load_dau_module() mysql_binary = helper.mysql_binary() def query(config: dict[str, str], sql: str, timeout: int = 60) -> str: statement = ( "SET SESSION TRANSACTION READ ONLY; " "START TRANSACTION WITH CONSISTENT SNAPSHOT; " f"{sql.rstrip().rstrip(';')}; ROLLBACK;" ) return helper.run_mysql(mysql_binary, config, statement, timeout) order_summary: list[dict[str, Any]] = [] channel_mix: list[dict[str, Any]] = [] payment_mix: list[dict[str, Any]] = [] meal_mix: list[dict[str, Any]] = [] restaurant_rank: list[dict[str, Any]] = [] daily_usage: list[dict[str, Any]] = [] monthly_usage: list[dict[str, Any]] = [] frequency: list[dict[str, Any]] = [] weekday_mix: list[dict[str, Any]] = [] feature_activity: list[dict[str, Any]] = [] management_activity: list[dict[str, Any]] = [] data_quality: list[dict[str, Any]] = [] project_meta: list[dict[str, Any]] = [] for project_key, project in PROJECTS.items(): config, config_source = helper.load_config(STORE_REPO, project["config_dir"]) if not config: raise RuntimeError(f"config_missing:{project_key}:{config_source}") identity = parse_rows( query(config, "SELECT DATABASE(), CURRENT_DATE(), NOW()"), ["database_name", "database_date", "queried_at"], )[0] existing_tables = set(helper.run_mysql(mysql_binary, config, "SHOW TABLES;", 30).splitlines()) project_name = project["project_name"] valid = "pay_status=20 AND order_status=30 AND is_delete=0 AND user_id>0" launch_start = dt.date.fromisoformat( str( parse_rows( query( config, f"SELECT MIN(meal_date) FROM ydy_meal_order WHERE {valid} AND meal_date<='{AS_OF}'", ), ["launch_start"], )[0]["launch_start"] ) ) project_meta.append( { "project_key": project_key, "project_name": project_name, "database_name": identity["database_name"], "database_date": identity["database_date"], "queried_at": identity["queried_at"], "config_source": config_source, "observable_launch_date": launch_start.isoformat(), "observable_launch_definition": "首个有效订单日", } ) periods = [ ("上线以来", launch_start, AS_OF), ("近30天", CURRENT_START, AS_OF), ("前30天", PREVIOUS_START, PREVIOUS_END), ("近90天", TREND_START, AS_OF), ] for period_name, start, end in periods: raw = query( 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), ROUND(COALESCE(SUM(total_price),0)/NULLIF(COUNT(*),0),2), COALESCE(SUM(total_price=0),0), COALESCE(SUM(refund_status>0),0) FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{start}' AND '{end}' """, ) values = parse_rows( raw, ["orders", "amount", "active_users", "active_days", "orders_per_user", "avg_order_value", "zero_amount_orders", "refunded_orders"], )[0] order_summary.append( { "project_key": project_key, "project_name": project_name, "period": period_name, "start_date": start.isoformat(), "end_date": end.isoformat(), **values, } ) for dimension, labels, target in ( ("source", ORDER_SOURCE_LABELS, channel_mix), ("pay_type", PAY_TYPE_LABELS, payment_mix), ("meal_times", MEAL_LABELS, meal_mix), ): for period_name, start, end in ( ("上线以来", launch_start, AS_OF), ("近30天", CURRENT_START, AS_OF), ("前30天", PREVIOUS_START, PREVIOUS_END), ): raw = query( config, f""" SELECT {dimension}, COUNT(*), COUNT(DISTINCT user_id), ROUND(SUM(total_price),2) FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{start}' AND '{end}' GROUP BY {dimension} ORDER BY COUNT(*) DESC """, ) rows = parse_rows(raw, ["dimension_value", "orders", "active_users", "amount"]) total = sum(int(row["orders"]) for row in rows) for row in rows: numeric_value = int(row["dimension_value"]) target.append( { "project_key": project_key, "project_name": project_name, "period": period_name, "start_date": start.isoformat(), "end_date": end.isoformat(), "label": labels.get(numeric_value, f"其他({numeric_value})"), "dimension_value": numeric_value, "orders": row["orders"], "active_users": row["active_users"], "amount": row["amount"], "order_share": round(int(row["orders"]) / total, 6) if total else 0, } ) raw = query( config, f""" SELECT o.restaurant_id, COALESCE(NULLIF(r.name,''),CONCAT('餐厅#',o.restaurant_id)), COUNT(*), COUNT(DISTINCT o.user_id), ROUND(SUM(o.total_price),2) FROM ydy_meal_order o LEFT JOIN ydy_restaurants r ON r.id=o.restaurant_id WHERE o.pay_status=20 AND o.order_status=30 AND o.is_delete=0 AND o.user_id>0 AND o.meal_date BETWEEN '{launch_start}' AND '{AS_OF}' GROUP BY o.restaurant_id,r.name ORDER BY COUNT(*) DESC """, ) restaurant_rows = parse_rows(raw, ["restaurant_id", "restaurant_name", "orders", "active_users", "amount"]) restaurant_total = sum(int(row["orders"]) for row in restaurant_rows) for rank, row in enumerate(restaurant_rows, start=1): restaurant_rank.append( { "project_key": project_key, "project_name": project_name, "rank": rank, **row, "order_share": round(int(row["orders"]) / restaurant_total, 6) if restaurant_total else 0, } ) raw = query( config, f""" SELECT meal_date, COUNT(*), COUNT(DISTINCT user_id), ROUND(SUM(total_price),2) FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{TREND_START}' AND '{AS_OF}' GROUP BY meal_date ORDER BY meal_date """, ) observed = {str(row["date"]): row for row in parse_rows(raw, ["date", "orders", "active_users", "amount"])} current_date = TREND_START while current_date <= AS_OF: key = current_date.isoformat() row = observed.get(key, {"date": key, "orders": 0, "active_users": 0, "amount": 0}) daily_usage.append({"project_key": project_key, "project_name": project_name, **row}) current_date += dt.timedelta(days=1) monthly_rows = parse_rows( query( config, f""" SELECT DATE_FORMAT(meal_date,'%Y-%m'),COUNT(*),COUNT(DISTINCT user_id),ROUND(SUM(total_price),2) FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{launch_start}' AND '{AS_OF}' GROUP BY DATE_FORMAT(meal_date,'%Y-%m') ORDER BY DATE_FORMAT(meal_date,'%Y-%m') """, ), ["month", "orders", "active_users", "amount"], ) for row in monthly_rows: monthly_usage.append({"project_key": project_key, "project_name": project_name, **row}) for frequency_period, frequency_start, frequency_end in ( ("上线以来", launch_start, AS_OF), ("近30天", CURRENT_START, AS_OF), ): raw = query( config, f""" SELECT bucket, COUNT(*), SUM(orders) FROM ( SELECT user_id, COUNT(*) orders, CASE WHEN COUNT(*)=1 THEN '1次' WHEN COUNT(*)<=5 THEN '2-5次' WHEN COUNT(*)<=10 THEN '6-10次' WHEN COUNT(*)<=20 THEN '11-20次' ELSE '21次以上' END bucket FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{frequency_start}' AND '{frequency_end}' GROUP BY user_id ) u GROUP BY bucket ORDER BY FIELD(bucket,'1次','2-5次','6-10次','11-20次','21次以上') """, ) for row in parse_rows(raw, ["frequency_bucket", "users", "orders"]): frequency.append( { "project_key": project_key, "project_name": project_name, "period": frequency_period, "start_date": frequency_start.isoformat(), "end_date": frequency_end.isoformat(), **row, } ) for weekday_period, weekday_start, weekday_end in ( ("上线以来", launch_start, AS_OF), ("近30天", CURRENT_START, AS_OF), ): raw = query( config, f""" SELECT WEEKDAY(meal_date)+1, COUNT(*), COUNT(DISTINCT user_id), ROUND(SUM(total_price),2) FROM ydy_meal_order WHERE {valid} AND meal_date BETWEEN '{weekday_start}' AND '{weekday_end}' GROUP BY WEEKDAY(meal_date) ORDER BY WEEKDAY(meal_date) """, ) for row in parse_rows(raw, ["weekday_number", "orders", "active_users", "amount"]): weekday_number = int(row["weekday_number"]) weekday_mix.append( { "project_key": project_key, "project_name": project_name, "period": weekday_period, "weekday": WEEKDAY_LABELS[weekday_number], **row, } ) for table_name, (feature_name, date_expression, extra_filter, actor_field, evidence_type) in FEATURE_TABLES.items(): if table_name not in existing_tables: continue actor_sql = ( f", COUNT(DISTINCT CASE WHEN ({date_expression}) BETWEEN '{launch_start}' " f"AND '{AS_OF} 23:59:59' THEN {actor_field} END)" if actor_field else ", 0" ) raw = query( config, f""" SELECT COUNT(*), COALESCE(SUM(({date_expression}) BETWEEN '{launch_start}' AND '{AS_OF} 23:59:59'),0), COALESCE(SUM(({date_expression}) BETWEEN '{CURRENT_START}' AND '{AS_OF} 23:59:59'),0), COALESCE(SUM(({date_expression}) BETWEEN '{PREVIOUS_START}' AND '{PREVIOUS_END} 23:59:59'),0) {actor_sql} FROM {table_name} WHERE {extra_filter} """, ) values = parse_rows( raw, [ "table_total_records", "lifetime_records", "current_30d_records", "previous_30d_records", "lifetime_distinct_actors", ], )[0] feature_activity.append( { "project_key": project_key, "project_name": project_name, "feature": feature_name, "evidence_type": evidence_type, "source_table": table_name, **values, } ) for management_period, management_start, management_end in ( ("上线以来", launch_start, AS_OF), ("近30天", CURRENT_START, AS_OF), ): export_rows = parse_rows( query( config, f""" SELECT type,status,COUNT(*),COUNT(DISTINCT create_by_uuid) FROM ydy_export_log WHERE create_time BETWEEN '{management_start}' AND '{management_end} 23:59:59' GROUP BY type,status ORDER BY COUNT(*) DESC """, ), ["type", "status", "events", "distinct_operators"], ) for row in export_rows: export_type = int(row["type"]) management_activity.append( { "project_key": project_key, "project_name": project_name, "period": management_period, "category": "报表导出", "activity": EXPORT_LABELS.get(export_type, f"其他导出({export_type})"), "result": "成功" if int(row["status"]) == 1 else f"状态{row['status']}", "events": row["events"], "distinct_operators": row["distinct_operators"], } ) if "ydy_security_audit_log" in existing_tables: for management_period, management_start, management_end in ( ("上线以来", launch_start, AS_OF), ("近30天", CURRENT_START, AS_OF), ): audit_rows = parse_rows( query( config, f""" SELECT module,action,result,COUNT(*),COUNT(DISTINCT operator) FROM ydy_security_audit_log WHERE is_delete=0 AND create_time BETWEEN '{management_start}' AND '{management_end} 23:59:59' GROUP BY module,action,result ORDER BY COUNT(*) DESC LIMIT 30 """, ), ["module", "action", "result", "events", "distinct_operators"], ) for row in audit_rows: management_activity.append( { "project_key": project_key, "project_name": project_name, "period": management_period, "category": f"审计:{row['module']}", "activity": row["action"], "result": row["result"], "events": row["events"], "distinct_operators": row["distinct_operators"], } ) quality = parse_rows( query( config, f""" SELECT COUNT(*),COUNT(DISTINCT id),COUNT(DISTINCT order_no), COALESCE(SUM(user_id=0),0),COALESCE(SUM(meal_date>'{AS_OF}'),0), COALESCE(SUM(pay_status<>20),0),COALESCE(SUM(order_status<>30),0), COALESCE(SUM(is_delete<>0),0),MIN(meal_date),MAX(meal_date) FROM ydy_meal_order """, ), ["rows", "distinct_ids", "distinct_order_numbers", "missing_user_orders", "future_meal_orders", "not_paid", "not_completed", "deleted", "first_meal_date", "last_meal_date"], )[0] data_quality.append({"project_key": project_key, "project_name": project_name, **quality}) if "ydy_meal_order_pickup" in existing_tables: for pickup_period, pickup_start, pickup_end in ( ("上线以来", launch_start, AS_OF), ("近30天", CURRENT_START, AS_OF), ): pickup_rows = parse_rows( query( config, f""" SELECT prepare_status,COUNT(*) FROM ydy_meal_order_pickup WHERE meal_date BETWEEN '{pickup_start}' AND '{pickup_end}' GROUP BY prepare_status ORDER BY COUNT(*) DESC """, ), ["prepare_status", "records"], ) for row in pickup_rows: data_quality.append( { "project_key": project_key, "project_name": project_name, "period": pickup_period, "quality_check": "取餐流程状态", "prepare_status": row["prepare_status"], "records": row["records"], } ) outputs = { "project_meta.csv": project_meta, "order_summary.csv": order_summary, "channel_mix.csv": channel_mix, "payment_mix.csv": payment_mix, "meal_mix.csv": meal_mix, "restaurant_rank.csv": restaurant_rank, "daily_usage.csv": daily_usage, "monthly_usage.csv": monthly_usage, "user_frequency.csv": frequency, "weekday_mix.csv": weekday_mix, "feature_activity.csv": feature_activity, "management_activity.csv": management_activity, "data_quality.csv": data_quality, } for filename, rows in outputs.items(): write_csv(filename, rows) snapshot = { "generated_at": dt.datetime.now().astimezone().isoformat(timespec="seconds"), "as_of": AS_OF.isoformat(), "timezone": "Asia/Shanghai", "periods": { "current_30d": [CURRENT_START.isoformat(), AS_OF.isoformat()], "previous_30d": [PREVIOUS_START.isoformat(), PREVIOUS_END.isoformat()], "trend_90d": [TREND_START.isoformat(), AS_OF.isoformat()], }, "datasets": {filename.removesuffix(".csv"): rows for filename, rows in outputs.items()}, } (OUTPUT_DIR / "analysis-snapshot.json").write_text( json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8" ) print(json.dumps({"output_dir": str(OUTPUT_DIR), "row_counts": {name: len(rows) for name, rows in outputs.items()}}, ensure_ascii=False)) if __name__ == "__main__": main()