wecom-fullstack
885cfddbff
ci/woodpecker/push/main Pipeline failed
P0-A（任务 fullstack-20261017-warning-capability-build-p0）： - cash_thresholds.py：阈值迁 per-entity（复用 alert_rules.rule_type=cash_gap_threshold， 默认 20/10 万不动；无配置/非法/异常一律回落默认，fail-safe 不改预警行为） - cash_forecast_engine.py：阈值改从配置读取，响应新增 threshold_source 明写来源 - verify_cash_gap_replay.py：同输入双跑逐字段 diff=0 + 手算 3/3 + 端点逐字段对拍(--live) - cash_gap_narrative.py：复用 ai_brief.call_deepseek，只写原因/建议； 断言生成文本数字集合 ⊆ 规则层数字集合，违例即不带文案上线 - audit_warning_data_readiness.py：G0 只读核验（cma_reader SELECT only）， 行数/期间/枚举 + 数据质量缺口（期初全 0、负期末、单批次、无员工维度表） 未 push（push 即部署，须项目侧书面放行）；未改生产库/未动 cron。
320 lines
15 KiB
Python
320 lines
15 KiB
Python
#!/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())
|