#!/usr/bin/env python3 # -*- coding: utf-8 -*- """G0 数据就绪核验（只读）— 经营预警能力落地 P0-A 任务键: fullstack-20261017-warning-capability-build-p0（上游 yanxue-20261017-warning-capability-build-p0） 用途: 对「资金 / 库存 / 销售 / 客户 / 员工」五类数据在 CMA 库内的就绪程度取证： 行数（分 entity）、期间上下界、枚举完整度、缺口清单。 纪律: · 只读：仅 SELECT / information_schema；凭据用 cma_reader（仅 GRANT SELECT ON cma.*） · 不改库、不注数、不建表；缺表/缺期间如实列为 gap，不用 0 或示例数填充 · 期间上下界取数据自身字段（product_sales.period_month / cash_plans.plan_date / *_inventory.source_batch），无期间字段的数据集明确标 period_field=null 用法: CMA_READER_ENV=/path/to/.secrets/cma-reader.env \ python3 backend/scripts/audit_warning_data_readiness.py [--entity 1] [--out FILE] 退出码: 0 = 核验已执行（无论是否存在数据缺口）；2 = 连接/权限失败 """ import argparse import json import os import sys from datetime import datetime import pymysql DEFAULT_READER_ENV = "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-reader.env" TARGET_ENTITY_DEFAULT = 1 # 酣客 # 需要取证的数据集（表名 → 期间字段/断言口径） TARGETS = [ # 资金（G1 依赖） {"table": "cash_plans", "period_field": "plan_date", "kind": "fact", "enum_fields": ["plan_type", "status", "source"], "amount_fields": ["amount"]}, {"table": "alert_rules", "period_field": None, "kind": "config", "expect_period": False, "enum_fields": ["rule_type", "trigger_on", "enabled"], "amount_fields": []}, {"table": "kpi_alerts", "period_field": None, "kind": "fact", "expect_period": False, "enum_fields": ["alert_level", "status", "alert_type"], "amount_fields": []}, # 库存（G3 依赖） {"table": "product_inventory", "period_field": "source_batch", "kind": "agg", "enum_fields": ["product_category", "source_batch"], "amount_fields": ["end_amount", "init_amount"]}, {"table": "product_inventory_detail", "period_field": "source_batch", "kind": "fact", "enum_fields": ["unit", "source_batch"], "amount_fields": ["end_amount", "init_amount"]}, # 销售（G3 依赖） {"table": "product_sales", "period_field": "period_month", "kind": "fact", "enum_fields": ["unit", "period_month"], "amount_fields": ["sales_amount", "cost_amount"]}, # 客户（G4 依赖） {"table": "customer_profitability_snapshot", "period_field": "period", "kind": "fact", "enum_fields": ["period", "calib_status"], "amount_fields": ["revenue", "contrib"]}, # 智享通维度/事实表（方案 §G0：应在接收侧路由，此处核验是否已入 CMA 库） {"table": "productinout_recent", "period_field": "None", "kind": "absent-probe", "enum_fields": [], "amount_fields": []}, {"table": "productinout_history", "period_field": "None", "kind": "absent-probe", "enum_fields": [], "amount_fields": []}, {"table": "units_dict", "period_field": None, "kind": "absent-probe", "enum_fields": [], "amount_fields": []}, {"table": "employee_dict", "period_field": None, "kind": "absent-probe", "enum_fields": [], "amount_fields": []}, {"table": "product_dict", "period_field": None, "kind": "absent-probe", "enum_fields": [], "amount_fields": []}, ] def load_env(path): env = {} with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if line and "=" in line and not line.startswith("#"): k, v = line.split("=", 1) env[k.strip()] = v.strip().strip('"').strip("'") missing = [k for k in ("CMA_READER_DB_HOST", "CMA_READER_DB_PORT", "CMA_READER_DB_USER", "CMA_READER_DB_PASS", "CMA_READER_DB_NAME") if k not in env] if missing: raise SystemExit(f"凭据文件缺字段: {missing}") return env def table_exists(cur, name): cur.execute("SELECT COUNT(*) FROM information_schema.tables " "WHERE table_schema=DATABASE() AND table_name=%s", (name,)) return cur.fetchone()[0] > 0 def columns(cur, name): cur.execute("SELECT column_name, column_type, is_nullable FROM information_schema.columns " "WHERE table_schema=DATABASE() AND table_name=%s ORDER BY ordinal_position", (name,)) return [{"name": c, "type": t, "nullable": (n == "YES")} for c, t, n in cur.fetchall()] def scalar(cur, sql, args=None): cur.execute(sql, args or ()) row = cur.fetchone() return None if row is None else row[0] def inspect_table(cur, spec, entity_id): name = spec["table"] out = {"table": name, "kind": spec["kind"], "exists": table_exists(cur, name), "expect_period": spec.get("expect_period", True)} if not out["exists"]: out["gap"] = "表不存在于 CMA 库（entity 无关）" return out cols = columns(cur, name) col_names = {c["name"] for c in cols} out["columns"] = cols out["row_count_total"] = int(scalar(cur, "SELECT COUNT(*) FROM `%s`" % name) or 0) if "entity_id" in col_names: cur.execute("SELECT entity_id, COUNT(*) FROM `%s` GROUP BY entity_id ORDER BY entity_id" % name) out["rows_by_entity"] = {str(e if e is not None else "NULL"): int(c) for e, c in cur.fetchall()} out["row_count_entity"] = out["rows_by_entity"].get(str(entity_id), 0) else: out["rows_by_entity"] = None out["row_count_entity"] = out["row_count_total"] out["gap_no_entity_column"] = True pf = spec["period_field"] if pf and pf in col_names and pf != "None": where = " WHERE entity_id=%s" % int(entity_id) if "entity_id" in col_names else "" cur.execute("SELECT MIN(`%s`), MAX(`%s`), COUNT(DISTINCT `%s`) FROM `%s`%s" % (pf, pf, pf, name, where)) lo, hi, distinct = cur.fetchone() out["period_field"] = pf out["period_bounds"] = {"min": _jsonable(lo), "max": _jsonable(hi), "distinct": int(distinct or 0)} cur.execute("SELECT `%s`, COUNT(*) FROM `%s`%s GROUP BY `%s` ORDER BY `%s`" % (pf, name, where, pf, pf)) out["period_rows"] = [[_jsonable(k), int(v)] for k, v in cur.fetchall()] else: out["period_field"] = None out["period_bounds"] = None out["gap_no_period_field"] = True # 枚举完整度：值分布 + 空值率 enum_detail = {} where = " WHERE entity_id=%s" % int(entity_id) if "entity_id" in col_names else "" denom = out["row_count_entity"] or 0 for ef in spec["enum_fields"]: if ef not in col_names: enum_detail[ef] = {"missing_column": True} continue cur.execute("SELECT `%s`, COUNT(*) FROM `%s`%s GROUP BY `%s` ORDER BY COUNT(*) DESC LIMIT 25" % (ef, name, where, ef)) dist = [{"value": _jsonable(v), "rows": int(c)} for v, c in cur.fetchall()] nulls = int(scalar(cur, "SELECT COUNT(*) FROM `%s`%s %s `%s` IS NULL" % (name, where, "AND" if where else "WHERE", ef)) or 0) blanks = 0 if any(c["type"].startswith(("varchar", "char", "text")) for c in cols if c["name"] == ef): blanks = int(scalar(cur, "SELECT COUNT(*) FROM `%s`%s %s `%s`=''" % (name, where, "AND" if where else "WHERE", ef)) or 0) enum_detail[ef] = {"distinct": len(dist), "distribution": dist, "null_rows": nulls, "blank_rows": blanks, "null_rate": round(nulls / denom, 4) if denom else None} out["enum_completeness"] = enum_detail # 金额字段空值与合计（仅 entity 范围） amt = {} for af in spec["amount_fields"]: if af in col_names: cur.execute("SELECT COUNT(*), COALESCE(SUM(`%s`),0) FROM `%s`%s %s `%s` IS NULL" % (af, name, where, "AND" if where else "WHERE", af)) nulls, _ = cur.fetchone() total = scalar(cur, "SELECT COALESCE(SUM(`%s`),0) FROM `%s`%s" % (af, name, where)) amt[af] = {"null_rows": int(nulls or 0), "sum": _jsonable(total)} out["amount_fields"] = amt return out def data_quality_checks(cur, entity_id: int) -> list: """数据质量取证（只读）：不只看「有没有表/几行」，还要看这批数据能不能支撑看板口径。 每条 finding 都带 SQL 结果，供回执逐条引用；发现问题不等于改库，只如实登记。 """ findings = [] has = table_exists(cur, "product_inventory") if has: cur.execute( "SELECT COUNT(*), COALESCE(SUM(init_qty),0), COALESCE(SUM(init_amount),0), " "SUM(CASE WHEN end_qty < 0 THEN 1 ELSE 0 END), COUNT(DISTINCT source_batch) " "FROM product_inventory WHERE entity_id=%s", (entity_id,)) rows, init_qty, init_amt, neg_rows, batches = cur.fetchone() if float(init_qty or 0) == 0 and float(init_amt or 0) == 0: findings.append({ "table": "product_inventory", "type": "init_all_zero", "detail": "entity_%s 全部行 init_qty/init_amount = 0 ⇒ 期初口径不可用，" "看板期初列必须显示『数据未就绪』" % entity_id, "evidence": {"rows": int(rows), "sum_init_qty": float(init_qty or 0), "sum_init_amount": float(init_amt or 0)}}) if int(neg_rows or 0) > 0: findings.append({ "table": "product_inventory", "type": "negative_end_qty", "detail": "entity_%s 有负期末数量行 ⇒ 该批为「当期进销」而非「期初+期末」，" "期末/周转需标注口径来源" % entity_id, "evidence": {"rows_with_negative_end_qty": int(neg_rows)}}) if int(batches or 0) == 1: findings.append({ "table": "product_inventory", "type": "single_batch_only", "detail": "仅 1 个 source_batch ⇒ 库存周转只能算单期快照，不能做多期环比/呆滞趋势", "evidence": {"distinct_source_batch": int(batches)}}) for kw in ("employ", "staff", "salesman"): cur.execute("SELECT table_name FROM information_schema.tables " "WHERE table_schema=DATABASE() AND table_name LIKE %s", ("%" + kw + "%",)) hits = [r[0] for r in cur.fetchall()] if hits: findings.append({"table": hits[0], "type": "employee_dim_suspected", "detail": "库内存在疑似员工维度表：%s（需逐表核口径）" % ",".join(hits)}) cur.execute("SELECT table_name FROM information_schema.tables " "WHERE table_schema=DATABASE() AND (table_name LIKE '%%employ%%' " "OR table_name LIKE '%%staff%%' OR table_name LIKE '%%salesman%%')") if not cur.fetchall(): findings.append({ "table": "(none)", "type": "no_employee_dimension_table", "detail": "CMA 库内无任何员工/业务员维度表 ⇒ 员工绩效明细（G4）当前无数据源，" "不只是口径未定（D2），需先解决 employee_dict 落库"}) return findings def _jsonable(v): if v is None: return None if isinstance(v, (int, float, str)): return v if hasattr(v, "isoformat"): return v.isoformat() from decimal import Decimal if isinstance(v, Decimal): return float(v) return str(v) def build_gaps(reports, entity_id): gaps, flags = [], [] for r in reports: if not r.get("exists"): gaps.append({"table": r["table"], "type": "missing_table", "detail": "CMA 库无此表（zxt 接收侧仅路由到通道，未落地 CMA）"}) continue if r.get("row_count_entity", 0) == 0: gaps.append({"table": r["table"], "type": "no_rows_for_entity", "detail": "entity_%s 无行" % entity_id}) for f, v in (r.get("enum_completeness") or {}).items(): if v.get("missing_column"): gaps.append({"table": r["table"], "type": "missing_column", "detail": f}) elif v.get("null_rate") == 1.0: flags.append({"table": r["table"], "field": f, "type": "enum_all_null", "detail": "该字段全为空 → 不能作为看板分组/口径依据"}) if r.get("gap_no_period_field") and r.get("expect_period", True): gaps.append({"table": r["table"], "type": "no_period_field", "detail": "无期间字段 → 无法声明覆盖区间（页面必须标『数据未就绪』）"}) return gaps, flags def main(): ap = argparse.ArgumentParser() ap.add_argument("--entity", type=int, default=TARGET_ENTITY_DEFAULT) ap.add_argument("--out", default="") args = ap.parse_args() reader_env = os.environ.get("CMA_READER_ENV", DEFAULT_READER_ENV) env = load_env(reader_env) try: conn = pymysql.connect(host=env["CMA_READER_DB_HOST"], port=int(env["CMA_READER_DB_PORT"]), user=env["CMA_READER_DB_USER"], password=env["CMA_READER_DB_PASS"], database=env["CMA_READER_DB_NAME"], charset="utf8mb4") except Exception as exc: # noqa: BLE001 print(json.dumps({"ok": False, "error": "connect_failed: %s" % str(exc)[:200]}, ensure_ascii=False), file=sys.stderr) return 2 reports, db_meta = [], {} try: cur = conn.cursor() cur.execute("SELECT CURRENT_USER(), DATABASE(), VERSION(), NOW()") u, db, ver, now = cur.fetchone() or (None, None, None, None) db_meta = {"current_user": u, "database": db, "version": ver, "server_now": _jsonable(now), "readonly_account": True, "reader_env_file": os.path.basename(reader_env)} for spec in TARGETS: reports.append(inspect_table(cur, spec, args.entity)) gaps, flags = build_gaps(reports, args.entity) data_quality = data_quality_checks(cur, args.entity) for f in data_quality: gaps.append({"table": f["table"], "type": f["type"], "detail": f["detail"], "evidence": f.get("evidence")}) cur.close() finally: conn.close() payload = { "task": "fullstack-20261017-warning-capability-build-p0", "stage": "G0 data readiness (read-only)", "generated_at": datetime.now().isoformat(timespec="seconds"), "entity_id": args.entity, "db": db_meta, "datasets": reports, "gaps": gaps, "gap_count": len(gaps), "data_quality": data_quality, "flags": flags, "flag_count": len(flags), } text = json.dumps(payload, ensure_ascii=False, indent=2, default=str) if args.out: os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) with open(args.out, "w", encoding="utf-8") as fh: fh.write(text + "\n") print("evidence written: %s (%d bytes)" % (args.out, len(text))) print("gap_count=%d" % len(gaps)) for g in gaps: print("GAP %-32s %-22s %s" % (g["table"], g["type"], g["detail"])) else: print(text) return 0 if __name__ == "__main__": sys.exit(main())