From 8adc783501ec622ec314767628650cd5ace19af3 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Mon, 14 Sep 2026 17:25:49 +0800 Subject: [PATCH] fix(budget): attribution backfill computes in-process and previews on dry-run (yanxue-20260914-budget-p13p14-activation) --- .../scripts/kpi_alert_attribution_backfill.py | 189 ++++++++++-------- 1 file changed, 105 insertions(+), 84 deletions(-) diff --git a/backend/scripts/kpi_alert_attribution_backfill.py b/backend/scripts/kpi_alert_attribution_backfill.py index f81e42db..86d6cba4 100644 --- a/backend/scripts/kpi_alert_attribution_backfill.py +++ b/backend/scripts/kpi_alert_attribution_backfill.py @@ -1,156 +1,177 @@ #!/usr/bin/env python3 """预算偏差告警归因回填 —— yanxue-20260914-budget-p13p14-activation（P1-③ 激活） -两类动作: - 1) attribution 为空的历史 open 告警 → 走现成接口 GET /budget/deviation-alerts/{id}/attribution - （端点在响应前现场组装归因并落库，见 app/api/budget.py:1004-1012） - 2) 已有旧口径归因的告警 → 用修正后的科目口径重算覆盖（旧值见 /root/WB_WS/_backups/p13p14-20260914/） -另: scenario_id 为空且 alert_type 命中 scenario_suggestions → 补 scenario_id（让告警"能解释、能关"） +两类动作（都在本进程内组装，不依赖线上服务是否已加载新代码）: + 1) attribution 为空的历史告警 → 现算归因并落库（等价于现成接口 + GET /budget/deviation-alerts/{id}/attribution 的组装逻辑，见 app/api/budget.py） + 2) 已有旧口径归因的告警 → 用修正后的科目口径重算覆盖（旧值备份见 + /root/WB_WS/_backups/p13p14-20260914/） +另: alert_type 为空 → 按 KPI 特征推断补上；scenario_id 为空且命中 + scenario_suggestions → 补 scenario_id（让告警"能解释、能关"） 用法: - /usr/bin/python3 scripts/kpi_alert_attribution_backfill.py # dry-run - /usr/bin/python3 scripts/kpi_alert_attribution_backfill.py --apply + /usr/bin/python3 scripts/kpi_alert_attribution_backfill.py # dry-run（只算不写） + /usr/bin/python3 scripts/kpi_alert_attribution_backfill.py --apply # 落库 + +落库前置: cma_api 需对 budget_deviation_alerts 有 UPDATE 权限 +（SHOW GRANTS 现状: 该表未在授权清单内 → 1142 UPDATE command denied）。 """ import json import os import sys -import urllib.error -import urllib.request +from types import SimpleNamespace if __name__ == "__main__": sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.database import get_session_local -from app.models import BudgetDeviationAlert, KPIDefinition, ScenarioSuggestion +from app.models import BudgetDeviationAlert, KPIDefinition, KPIValueSource, ScenarioSuggestion from app.utils.alert_attribution import build_attribution, match_scenario -API = "http://127.0.0.1:8010" -SECRET = "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-admin.env" OUT = "/tmp/budget-p13p14/attribution-conclusions.json" -def load_env(path): - vals = {} - with open(path) as fh: - for line in fh: - line = line.strip() - if line and not line.startswith("#") and "=" in line: - k, v = line.split("=", 1) - vals[k.strip()] = v.strip() - return vals +def mapping_status(db, kpi_id): + src = db.query(KPIValueSource).filter( + KPIValueSource.kpi_id == kpi_id, + KPIValueSource.source_table == "voucher_details", + ).order_by(KPIValueSource.status.asc()).first() + return src.status if src else None -def api(method, path, body=None, token=None): - data = json.dumps(body).encode() if body is not None else None - req = urllib.request.Request(API + path, data=data, method=method) - req.add_header("Content-Type", "application/json") - if token: - req.add_header("Authorization", f"Bearer {token}") - try: - with urllib.request.urlopen(req, timeout=90) as r: - return r.status, json.loads(r.read().decode()) - except urllib.error.HTTPError as e: - return e.code, {"_error": e.read().decode()[:200]} - - -def conclusion(alert, kpi, attribution, scenario): +def conclusion(alert, kpi, attribution, scenario, map_status): dims = attribution.get("dimensions") or [] subs = attribution.get("subjects") or [] trend = attribution.get("trend") or {} vtype = attribution.get("variance_type") hot = [s for s in subs if abs(s.get("amount_diff") or 0) > 0.01][:3] - sub_txt = "; ".join( - f"{s['subject_code']} {s.get('subject_name','')} {s['amount_diff']:+.2f}元(权重{s.get('weight')})" - for s in hot) or "（科目明细为空：该KPI无可用取数映射，见报告§未接通清单）" + if hot: + sub_txt = "; ".join( + f"{s['subject_code']} {s.get('subject_name', '')} " + f"{s['amount_diff']:+.2f}元(x{s.get('weight')})" for s in hot) + depth = f"科目级({map_status or 'kpi_subject_map原口径'})" + elif map_status: + sub_txt = (f"科目明细为空/金额为0（映射={map_status}，当期凭证明细无该科目发生额）") + depth = f"KPI级(映射={map_status})" + else: + sub_txt = "无取数映射（kpi_subject_map 科目未机械桥接到 voucher_details 科目）→ 仅 KPI 级结论" + depth = "KPI级(无映射)" dim_txt = "; ".join( f"{d.get('kpi_name')} 偏差{d.get('deviation_value')}" for d in dims if d.get("deviation_value") is not None and abs(d["deviation_value"]) > 0.01) or "无下级KPI层级" trend_txt = (f"{trend.get('type')}({trend.get('message')})" if trend.get("anomaly") else "无连续3期异常") - return ( - f"[{alert.id}] {kpi.kpi_code if kpi else alert.kpi_id} {alert.period}: " + view = ( + f"[{alert.id}] {kpi.kpi_code if kpi else alert.kpi_id} {kpi.kpi_name if kpi else ''} {alert.period}: " f"实际{alert.actual_value} vs 预算{alert.budget_value}（偏差{alert.deviation_rate}%, {alert.alert_level}）| " f"场景={alert.alert_type or '未分类'}{'(建议:' + scenario['title'] + ')' if scenario else ''} | " - f"量价={vtype} | 主科目={sub_txt} | 子KPI={dim_txt} | 趋势={trend_txt}" + f"深度={depth} | 量价={vtype} | 主科目={sub_txt} | 子KPI={dim_txt} | 趋势={trend_txt}" ) + return view, depth def main(): apply = "--apply" in sys.argv - sec = load_env(SECRET) - st, login = api("POST", "/api/cma/auth/login", - {"username": sec["CMA_ADMIN_USER"], "password": sec["CMA_ADMIN_PASS"], "entity_id": 1}) - token = login.get("token") - print(f"login: HTTP {st} entity={login.get('entity_short_name')}") - db = get_session_local()() try: alerts = db.query(BudgetDeviationAlert).filter( BudgetDeviationAlert.entity_id == 1).order_by(BudgetDeviationAlert.id).all() - print(f"== 告警: {len(alerts)} 条（open {sum(1 for a in alerts if a.status == 'open')}）") + print(f"== 告警: {len(alerts)} 条（open {sum(1 for a in alerts if a.status == 'open')}）" + f" | mode={'APPLY' if apply else 'DRY-RUN'}") - api_filled, recomputed, scenario_filled = 0, 0, 0 + filled, recomputed, type_filled, scenario_filled = 0, 0, 0, 0 lines = [] for a in alerts: kpi = db.query(KPIDefinition).filter(KPIDefinition.id == a.kpi_id).first() + mstat = mapping_status(db, a.kpi_id) + att, inferred = build_attribution(db, a.kpi_id, a.period, a.alert_type) had = a.attribution is not None - if not had and apply: - # 1) 现成接口路径：现场组装并落库 - st_a, resp = api("GET", f"/api/cma/budget/deviation-alerts/{a.id}/attribution", token=token) - if st_a == 200: - api_filled += 1 - db.expire(a) # 端点已 commit，重读 - a = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.id == a.id).first() - elif had and apply: - # 2) 旧口径 → 按修正科目口径重算 - att, inferred = build_attribution(db, a.kpi_id, a.period, a.alert_type) - a.attribution = att - if a.alert_type is None: + if apply: + if not had: + a.attribution = att + filled += 1 + else: + if (a.attribution or {}) != att: + a.attribution = att + recomputed += 1 + if a.alert_type is None and inferred: a.alert_type = inferred - recomputed += 1 + type_filled += 1 - scenario = match_scenario(db, a.alert_type) - if scenario and a.scenario_id is None and apply: + eff_type = a.alert_type or inferred + scenario = match_scenario(db, eff_type) + if apply and scenario and a.scenario_id is None: a.scenario_id = scenario["scenario_id"] scenario_filled += 1 - att_now = a.attribution or {} + alert_copy = a + if not apply: + # 只读预览：不改 ORM 对象（避免 autoflush 落库） + alert_copy = SimpleNamespace( + id=a.id, kpi_id=a.kpi_id, period=a.period, status=a.status, + alert_level=a.alert_level, budget_value=a.budget_value, + actual_value=a.actual_value, deviation_rate=a.deviation_rate, + deviation_value=a.deviation_value, alert_type=eff_type, + scenario_id=a.scenario_id) + + view, depth = conclusion(alert_copy, kpi, att, scenario, mstat) lines.append({ "id": a.id, "kpi_id": a.kpi_id, "kpi_code": kpi.kpi_code if kpi else None, + "kpi_name": kpi.kpi_name if kpi else None, "period": a.period, "status": a.status, "alert_level": a.alert_level, "budget_value": a.budget_value, "actual_value": a.actual_value, - "deviation_rate": a.deviation_rate, "alert_type": a.alert_type, - "scenario_id": a.scenario_id, "has_attribution": bool(a.attribution), - "attribution_keys": sorted(att_now.keys()), - "view": conclusion(a, kpi, att_now, scenario), + "deviation_rate": a.deviation_rate, "alert_type": eff_type, + "scenario_id": a.scenario_id, + "mapping_status": mstat, "depth": depth, + "n_subjects": len(att.get("subjects") or []), + "n_subject_moves": sum(1 for s in (att.get("subjects") or []) + if abs(s.get("amount_diff") or 0) > 0.01), + "variance_type": att.get("variance_type"), + "had_attribution": had, + "view": view, }) if apply: - db.commit() + try: + db.commit() + except Exception as e: + db.rollback() + print(f"!! 落库失败（未写任何行）: {type(e).__name__}: {str(e)[:200]}") + return 2 for row in lines: print(row["view"]) + os.makedirs(os.path.dirname(OUT), exist_ok=True) with open(OUT, "w") as fh: json.dump(lines, fh, ensure_ascii=False, indent=1) - print(f"\n== 输出结论: {OUT} ({len(lines)} 条)") + print(f"\n== 结论落盘: {OUT} ({len(lines)} 条)") - n_attr = db.query(BudgetDeviationAlert).filter( - BudgetDeviationAlert.entity_id == 1, - BudgetDeviationAlert.attribution.isnot(None)).count() - total = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.entity_id == 1).count() - n_open = db.query(BudgetDeviationAlert).filter( - BudgetDeviationAlert.entity_id == 1, BudgetDeviationAlert.status == "open").count() - n_scen = db.query(BudgetDeviationAlert).filter( - BudgetDeviationAlert.entity_id == 1, - BudgetDeviationAlert.scenario_id.isnot(None)).count() - print(f"== 归因覆盖: {n_attr}/{total}（open {n_open}）| scenario_id 已补: {n_scen} | " - f"接口落库 {api_filled} / 重算 {recomputed} / 场景补全 {scenario_filled}" - f"{'' if apply else ' [dry-run 未写库]'}") + depths = {} + for row in lines: + depths[row["depth"]] = depths.get(row["depth"], 0) + 1 + print(f"== 归因深度分布: {json.dumps(depths, ensure_ascii=False)}") + opens = [r for r in lines if r["status"] == "open"] + print(f"== open 告警有结论: {len(opens)}/{len(opens)}" + f" | 其中科目级可解释: {sum(1 for r in opens if r['depth'].startswith('科目级'))}") + if apply: + n_attr = db.query(BudgetDeviationAlert).filter( + BudgetDeviationAlert.entity_id == 1, + BudgetDeviationAlert.attribution.isnot(None)).count() + total = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.entity_id == 1).count() + n_scen = db.query(BudgetDeviationAlert).filter( + BudgetDeviationAlert.entity_id == 1, + BudgetDeviationAlert.scenario_id.isnot(None)).count() + print(f"== 落库结果: attribution {n_attr}/{total} | scenario_id {n_scen} | " + f"新填 {filled} / 重算 {recomputed} / 场景类型补全 {type_filled} / " + f"场景建议补全 {scenario_filled}") + else: + print("== DRY-RUN: 未写库（落库需 --apply，且 cma_api 需 UPDATE 授权）") + return 0 finally: db.close() if __name__ == "__main__": - main() + sys.exit(main())