#!/usr/bin/env python3 from __future__ import annotations import csv import html import os import shutil import subprocess from datetime import datetime from pathlib import Path from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill, Border, Side from openpyxl.utils import get_column_letter ROOT = Path(__file__).resolve().parents[3] WORK_DIR = ROOT / "work/2026-06-30-online-project-usage-consumption-report" OUT_DIR = WORK_DIR / "outputs" DESKTOP_DIR = Path.home() / "Desktop/上线项目使用人数消费额度统计_20260630" ENV_FILE = ROOT / "config/local/jsr-db.env" MYSQL = "/usr/local/mysql/bin/mysql" PROJECTS = [ { "project_name": "江苏国信智慧营养健康餐厅", "project_key": "guoxin", "db_scope": "当前库", "store_db": "zhct_guoxin", "ai_db": "aizhct_guoxin", "domain": "https://jsgx.yyangpt.cn/", "deployment_evidence": "work_win_deploy_guoxin/worklog/2026-06-19-guoxin-linux-cloud-deployment.md", }, { "project_name": "机场智慧营养健康餐厅", "project_key": "airport", "db_scope": "旧库", "store_db": "zhct_jc", "ai_db": "aizhct_jc", "domain": "https://jc.yyangpt.cn/", "deployment_evidence": "application/config/prod/zhct_jc/config.php", }, { "project_name": "机场智慧营养健康餐厅", "project_key": "airport", "db_scope": "新库", "store_db": "zhct_jc_new", "ai_db": "aizhct_jc_new", "domain": "未在当前 prod config 中单独登记", "deployment_evidence": "只读数据库 SHOW DATABASES 可见", }, { "project_name": "金斯瑞智慧营养健康餐厅", "project_key": "jsr", "db_scope": "当前库", "store_db": "zhct_jsr", "ai_db": "aizhct_jsr", "domain": "https://jsr.yyangpt.cn/", "deployment_evidence": "control/jsr-deployment-versions.md", }, { "project_name": "赛迪物业智慧营养健康餐厅", "project_key": "saidi", "db_scope": "当前库", "store_db": "zhct_saidi", "ai_db": "aizhct_saidi", "domain": "https://saidi.yyangpt.cn/", "deployment_evidence": "control/saidi-deployment-versions.md", }, { "project_name": "山西焦煤智慧营养健康餐厅", "project_key": "sxjm", "db_scope": "当前库", "store_db": "zhct_sxjm", "ai_db": "aizhct_sxjm", "domain": "https://sxjm.yyangpt.cn/", "deployment_evidence": "application/config/prod/zhct_sxjm/config.php", }, ] KNOWN_DEPLOYED_BUT_NOT_VISIBLE = [ ("产业园智慧营养健康餐厅", "zhct", "aizhct", "应用配置存在；本只读账号未显示该库，待补只读数据源"), ("首通智城智慧营养健康餐厅", "zhct_stzc", "aizhct_stzc", "control/stzc-deployment-versions.md 有部署记录；本只读账号未显示该库"), ("湖南体职院", "zhct_hntzy", "未确认", "应用配置存在；本只读账号未显示该库"), ("四川射洪智慧营养健康餐厅", "zhct_sh", "未确认", "应用配置存在；本只读账号未显示该库"), ("西康宾馆智慧营养健康餐厅", "zhct_xkbg", "未确认", "应用配置存在；本只读账号未显示该库"), ("无锡新吴区区政府智慧营养健康餐厅", "zhct_xwq", "未确认", "应用配置存在；本只读账号未显示该库"), ("海开集团智慧营养健康餐厅", "zhct_haikai", "未确认", "应用配置存在；本只读账号未显示该库"), ] def load_env(path: Path) -> dict[str, str]: env: dict[str, str] = {} for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) env[key.strip()] = value.strip().strip('"').strip("'") return env def run_mysql(env: dict[str, str], database: str | None, sql: str) -> list[dict[str, str]]: cmd = [ MYSQL, "--connect-timeout=8", "--batch", "--raw", "-h", env["JSR_DB_HOST"], "-u", env["JSR_DB_USER"], f"-p{env['JSR_DB_PASSWORD']}", ] if database: cmd.append(database) cmd.extend(["-e", sql]) proc = subprocess.run(cmd, text=True, capture_output=True, check=False) if proc.returncode != 0: raise RuntimeError(proc.stderr.strip()) lines = [line for line in proc.stdout.splitlines() if line.strip()] if not lines: return [] header = lines[0].split("\t") rows: list[dict[str, str]] = [] for line in lines[1:]: values = line.split("\t") rows.append(dict(zip(header, values))) return rows def decimal_text(value: str | None) -> str: if value in (None, "", "NULL"): return "0.00" return f"{float(value):.2f}" def int_text(value: str | None) -> str: if value in (None, "", "NULL"): return "0" return str(int(float(value))) def main() -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) DESKTOP_DIR.mkdir(parents=True, exist_ok=True) env = load_env(ENV_FILE) generated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") login_rows = run_mysql( env, None, "SELECT CURRENT_USER() AS login_user", ) login_user = login_rows[0]["login_user"] if login_rows else "unknown" login_user_label = "已验证只读账号" db_rows: list[dict[str, str]] = [] for project in PROJECTS: store_db = project["store_db"] ai_db = project["ai_db"] staff = run_mysql( env, store_db, f""" SELECT '{store_db}' AS store_db, COUNT(*) AS staff_total, SUM(CASE WHEN is_show = 1 AND status = 1 THEN 1 ELSE 0 END) AS staff_visible_active FROM ydy_staff """, )[0] order = run_mysql( env, store_db, f""" SELECT '{store_db}' AS store_db, COUNT(*) AS paid_orders, COUNT(DISTINCT user_id) AS paid_users, COALESCE(SUM(pay_price), 0) AS paid_pay_price, COALESCE(SUM(cash), 0) AS paid_cash, COALESCE(SUM(subsidy), 0) AS paid_subsidy, COALESCE(SUM(refund_amount), 0) AS refund_amount, COALESCE(SUM(pay_price - refund_amount), 0) AS paid_pay_price_net, MIN(pay_time) AS first_pay_time, MAX(pay_time) AS last_pay_time FROM ydy_meal_order WHERE pay_status = 20 AND is_delete = 0 """, )[0] ai_user = run_mysql( env, ai_db, f""" SELECT '{ai_db}' AS ai_db, COUNT(*) AS ai_users, COALESCE(SUM(balance), 0) AS cash_balance, COALESCE(SUM(subsidy_balance), 0) AS user_subsidy_balance, COALESCE(SUM(pay_money), 0) AS user_pay_money, COALESCE(SUM(expend_money), 0) AS user_expend_money FROM yoshop_user WHERE is_delete = 0 """, )[0] subsidy = run_mysql( env, ai_db, f""" SELECT '{ai_db}' AS ai_db, COUNT(*) AS subsidy_rows, COUNT(DISTINCT user_id) AS subsidy_users, COALESCE(SUM(money), 0) AS subsidy_remaining, COALESCE(SUM(used_money), 0) AS subsidy_used, COALESCE(SUM(CASE WHEN expiration_end_time >= NOW() THEN money ELSE 0 END), 0) AS unexpired_subsidy_remaining FROM yoshop_user_subsidy WHERE is_delete = 0 """, )[0] current_available = float(ai_user["cash_balance"]) + float(subsidy["unexpired_subsidy_remaining"]) db_rows.append( { "项目Key": project["project_key"], "项目": project["project_name"], "数据库范围": project["db_scope"], "store库": store_db, "AI库": ai_db, "访问域名": project["domain"], "支撑人员数_人员表": int_text(staff["staff_total"]), "当前有效人员数": int_text(staff["staff_visible_active"]), "已消费人数": int_text(order["paid_users"]), "已支付订单数": int_text(order["paid_orders"]), "累计实付消费额": decimal_text(order["paid_pay_price"]), "累计现金消费额": decimal_text(order["paid_cash"]), "累计补贴消费额": decimal_text(order["paid_subsidy"]), "累计退款额": decimal_text(order["refund_amount"]), "累计净消费额": decimal_text(order["paid_pay_price_net"]), "AI账户数": int_text(ai_user["ai_users"]), "当前现金余额": decimal_text(ai_user["cash_balance"]), "当前补贴余额": decimal_text(subsidy["subsidy_remaining"]), "当前未过期补贴余额": decimal_text(subsidy["unexpired_subsidy_remaining"]), "当前可用额度_现金加未过期补贴": f"{current_available:.2f}", "首笔支付时间": order["first_pay_time"] if order["first_pay_time"] != "NULL" else "", "最新支付时间": order["last_pay_time"] if order["last_pay_time"] != "NULL" else "", "证据等级": "A-只读生产库", "部署/配置证据": project["deployment_evidence"], } ) report_rows = aggregate_project_rows(db_rows) gap_rows = [ { "项目": name, "store库": store_db, "AI库": ai_db, "当前状态": note, "本次处理": "未纳入金额汇总；需补同等级只读账号或客户侧导出表后再统计", } for name, store_db, ai_db, note in KNOWN_DEPLOYED_BUT_NOT_VISIBLE ] report_csv = OUT_DIR / "online-project-usage-consumption-summary.csv" with report_csv.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(report_rows[0].keys())) writer.writeheader() writer.writerows(report_rows) detail_csv = OUT_DIR / "online-project-usage-consumption-db-detail.csv" with detail_csv.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(db_rows[0].keys())) writer.writeheader() writer.writerows(db_rows) gap_csv = OUT_DIR / "online-project-data-gaps.csv" with gap_csv.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(gap_rows[0].keys())) writer.writeheader() writer.writerows(gap_rows) workbook_path = OUT_DIR / "online-project-usage-consumption-report-20260630.xlsx" wb = Workbook() ws = wb.active ws.title = "上线项目汇总" write_sheet(ws, report_rows) ws_detail = wb.create_sheet("数据库拆分明细") write_sheet(ws_detail, db_rows) ws_gap = wb.create_sheet("待补数据源") write_sheet(ws_gap, gap_rows) ws_notes = wb.create_sheet("口径说明") notes = [ ["生成时间", generated_at], ["数据库登录用户", login_user_label], ["实时统计范围", "当前只读账号可见的 zhct_guoxin、zhct_jc、zhct_jc_new、zhct_jsr、zhct_saidi、zhct_sxjm 及对应 aizhct 库"], ["支撑人员数", "ydy_staff 全量人员数；当前有效人员数为 is_show=1 且 status=1；同一项目新旧库并存时，项目汇总页取当前最大人员规模，数据库拆分页保留各库明细"], ["已消费人数", "ydy_meal_order 中 pay_status=20 且 is_delete=0 的 distinct user_id"], ["累计实付消费额", "ydy_meal_order.pay_price 汇总，条件为 pay_status=20 且 is_delete=0"], ["累计净消费额", "ydy_meal_order.pay_price - refund_amount 汇总"], ["当前可用额度", "aizhct 库 yoshop_user.balance + yoshop_user_subsidy 未过期 money"], ["未纳入范围", "只在应用配置或部署文档中存在、但当前只读账号不可见的项目不估算金额"], ] for row in notes: ws_notes.append(row) style_sheet(ws_notes) wb.save(workbook_path) md_path = OUT_DIR / "online-project-usage-consumption-report.md" md_path.write_text(build_markdown(report_rows, gap_rows, generated_at, login_user_label), encoding="utf-8") html_path = OUT_DIR / "online-project-usage-consumption-report.html" html_path.write_text(build_html(report_rows, gap_rows, generated_at, login_user_label), encoding="utf-8") sql_path = OUT_DIR / "query-sql.md" sql_path.write_text(build_sql_note(generated_at, login_user_label), encoding="utf-8") for path in [report_csv, detail_csv, gap_csv, workbook_path, md_path, html_path, sql_path]: shutil.copy2(path, DESKTOP_DIR / path.name) print(workbook_path) print(html_path) print(DESKTOP_DIR) def aggregate_project_rows(db_rows: list[dict[str, str]]) -> list[dict[str, str]]: grouped: dict[str, list[dict[str, str]]] = {} for row in db_rows: grouped.setdefault(row["项目Key"], []).append(row) summary_rows: list[dict[str, str]] = [] for rows in grouped.values(): first = rows[0] latest_times = [row["最新支付时间"] for row in rows if row["最新支付时间"]] first_times = [row["首笔支付时间"] for row in rows if row["首笔支付时间"]] current_staff = max(int(row["当前有效人员数"]) for row in rows) total_staff = max(int(row["支撑人员数_人员表"]) for row in rows) db_scope = " + ".join(row["数据库范围"] for row in rows) summary_rows.append( { "项目": first["项目"], "数据库范围": db_scope, "store库": " + ".join(row["store库"] for row in rows), "AI库": " + ".join(row["AI库"] for row in rows), "访问域名": first["访问域名"], "支撑人员数_人员表": str(total_staff), "当前有效人员数": str(current_staff), "已消费人数": str(sum(int(row["已消费人数"]) for row in rows)), "已消费人数口径": "多库相加，未跨库去重" if len(rows) > 1 else "单库 distinct user_id", "已支付订单数": str(sum(int(row["已支付订单数"]) for row in rows)), "累计实付消费额": f"{sum(float(row['累计实付消费额']) for row in rows):.2f}", "累计现金消费额": f"{sum(float(row['累计现金消费额']) for row in rows):.2f}", "累计补贴消费额": f"{sum(float(row['累计补贴消费额']) for row in rows):.2f}", "累计退款额": f"{sum(float(row['累计退款额']) for row in rows):.2f}", "累计净消费额": f"{sum(float(row['累计净消费额']) for row in rows):.2f}", "AI账户数": str(max(int(row["AI账户数"]) for row in rows)), "当前现金余额": f"{sum(float(row['当前现金余额']) for row in rows):.2f}", "当前补贴余额": f"{sum(float(row['当前补贴余额']) for row in rows):.2f}", "当前未过期补贴余额": f"{sum(float(row['当前未过期补贴余额']) for row in rows):.2f}", "当前可用额度_现金加未过期补贴": f"{sum(float(row['当前可用额度_现金加未过期补贴']) for row in rows):.2f}", "首笔支付时间": min(first_times) if first_times else "", "最新支付时间": max(latest_times) if latest_times else "", "证据等级": "A-只读生产库", "部署/配置证据": "；".join(row["部署/配置证据"] for row in rows), } ) return summary_rows def write_sheet(ws, rows: list[dict[str, str]]) -> None: headers = list(rows[0].keys()) ws.append(headers) for row in rows: ws.append([row[h] for h in headers]) style_sheet(ws) def style_sheet(ws) -> None: header_fill = PatternFill("solid", fgColor="1F4E79") header_font = Font(name="微软雅黑", size=12, bold=True, color="FFFFFF") body_font = Font(name="微软雅黑", size=12) thin = Side(style="thin", color="D9E2F3") border = Border(left=thin, right=thin, top=thin, bottom=thin) for row in ws.iter_rows(): for cell in row: cell.font = body_font cell.alignment = Alignment(vertical="top", wrap_text=True) cell.border = border for cell in ws[1]: cell.fill = header_fill cell.font = header_font ws.freeze_panes = "A2" for col in ws.columns: max_len = 0 letter = get_column_letter(col[0].column) for cell in col: max_len = max(max_len, len(str(cell.value or ""))) ws.column_dimensions[letter].width = min(max(max_len + 2, 12), 32) def money(value: str) -> str: return f"{float(value):,.2f}" def build_markdown(rows, gaps, generated_at, login_user) -> str: total_staff = sum(int(row["支撑人员数_人员表"]) for row in rows) total_paid_users = sum(int(row["已消费人数"]) for row in rows) total_paid = sum(float(row["累计实付消费额"]) for row in rows) total_available = sum(float(row["当前可用额度_现金加未过期补贴"]) for row in rows) lines = [ "# 上线项目使用人数与消费额度统计", "", f"- 生成时间：{generated_at}", f"- 数据库登录用户：`{login_user}`", "- 统计范围：当前只读账号可见的 6 个线上 store 库及对应 AI 库。", "- 证据等级：可见库实时统计为 A；只在部署/配置中存在但本账号不可见的项目列为待补数据源。", "", "## 汇总结论", "", f"- 可见上线库合计支撑人员数：{total_staff:,} 人。", f"- 已产生消费的人数：{total_paid_users:,} 人。", f"- 截至生成时间累计实付消费额：¥{money(str(total_paid))}。", f"- 当前可用额度（现金余额 + 未过期补贴余额）：¥{money(str(total_available))}。", "", "## 项目明细", "", "| 项目 | 支撑人员数 | 已消费人数 | 累计实付消费额 | 累计净消费额 | 当前可用额度 | 最新支付时间 |", "| --- | ---: | ---: | ---: | ---: | ---: | --- |", ] for row in rows: lines.append( f"| {row['项目']} | {int(row['支撑人员数_人员表']):,} | {int(row['已消费人数']):,} | " f"¥{money(row['累计实付消费额'])} | ¥{money(row['累计净消费额'])} | " f"¥{money(row['当前可用额度_现金加未过期补贴'])} | {row['最新支付时间']} |" ) lines.extend( [ "", "## 待补数据源", "", "| 项目 | store库 | AI库 | 状态 |", "| --- | --- | --- | --- |", ] ) for row in gaps: lines.append(f"| {row['项目']} | `{row['store库']}` | `{row['AI库']}` | {row['当前状态']} |") lines.extend( [ "", "## 口径", "", "- 支撑人员数：`ydy_staff` 全量人员数；当前有效人员数为 `is_show=1 AND status=1`。", "- 累计实付消费额：`ydy_meal_order.pay_price`，筛选 `pay_status=20 AND is_delete=0`。", "- 累计净消费额：`pay_price - refund_amount`。", "- 当前可用额度：`aizhct_*.yoshop_user.balance + yoshop_user_subsidy` 未过期 `money`。", "- 未估算不可见库，避免把部署配置当成生产经营数据。", ] ) return "\n".join(lines) + "\n" def build_html(rows, gaps, generated_at, login_user) -> str: total_staff = sum(int(row["支撑人员数_人员表"]) for row in rows) total_paid_users = sum(int(row["已消费人数"]) for row in rows) total_paid = sum(float(row["累计实付消费额"]) for row in rows) total_available = sum(float(row["当前可用额度_现金加未过期补贴"]) for row in rows) def td(value, cls=""): return f"{html.escape(str(value))}" row_html = "\n".join( "" + td(row["项目"]) + td(f"{int(row['支撑人员数_人员表']):,}", "num") + td(f"{int(row['已消费人数']):,}", "num") + td(f"¥{money(row['累计实付消费额'])}", "num") + td(f"¥{money(row['累计净消费额'])}", "num") + td(f"¥{money(row['当前现金余额'])}", "num") + td(f"¥{money(row['当前未过期补贴余额'])}", "num") + td(f"¥{money(row['当前可用额度_现金加未过期补贴'])}", "num") + td(row["最新支付时间"]) + "" for row in rows ) gap_html = "\n".join( "" + td(row["项目"]) + td(row["store库"]) + td(row["AI库"]) + td(row["当前状态"]) + "" for row in gaps ) return f""" 上线项目使用人数与消费额度统计

上线项目使用人数与消费额度统计

生成时间：{html.escape(generated_at)} ｜ 只读账号：{html.escape(login_user)} ｜ 统计范围：当前只读账号可见的 6 个线上库
支撑人员数{total_staff:,}
已消费人数{total_paid_users:,}
累计实付消费额¥{money(str(total_paid))}
当前可用额度¥{money(str(total_available))}

项目明细

{row_html}
项目支撑人员数已消费人数累计实付消费额累计净消费额现金余额未过期补贴余额当前可用额度最新支付时间

待补数据源

{gap_html}
项目store库AI库状态

口径说明

累计实付消费额取 ydy_meal_order.pay_price，筛选 pay_status=20 AND is_delete=0；当前可用额度取 AI 库现金余额加未过期补贴余额。不可见库不做估算。
""" def build_sql_note(generated_at: str, login_user: str) -> str: return f"""# 查询 SQL 口径 - 生成时间：{generated_at} - 数据库登录用户：`{login_user}` - 说明：本文件只保存 SQL 口径，不保存 host、账号、密码。 ## 人员数 ```sql SELECT COUNT(*) AS staff_total, SUM(CASE WHEN is_show = 1 AND status = 1 THEN 1 ELSE 0 END) AS staff_visible_active FROM ydy_staff; ``` ## 消费额 ```sql SELECT COUNT(*) AS paid_orders, COUNT(DISTINCT user_id) AS paid_users, COALESCE(SUM(pay_price), 0) AS paid_pay_price, COALESCE(SUM(cash), 0) AS paid_cash, COALESCE(SUM(subsidy), 0) AS paid_subsidy, COALESCE(SUM(refund_amount), 0) AS refund_amount, COALESCE(SUM(pay_price - refund_amount), 0) AS paid_pay_price_net, MIN(pay_time) AS first_pay_time, MAX(pay_time) AS last_pay_time FROM ydy_meal_order WHERE pay_status = 20 AND is_delete = 0; ``` ## 当前可用额度 ```sql SELECT COALESCE(SUM(balance), 0) AS cash_balance FROM yoshop_user WHERE is_delete = 0; SELECT COALESCE(SUM(money), 0) AS subsidy_remaining, COALESCE(SUM(CASE WHEN expiration_end_time >= NOW() THEN money ELSE 0 END), 0) AS unexpired_subsidy_remaining FROM yoshop_user_subsidy WHERE is_delete = 0; ``` """ if __name__ == "__main__": main()