From 9b3a131070311f0ab5018e16c3cabb15141969d4 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Wed, 16 Sep 2026 23:46:48 +0800 Subject: [PATCH] fix(cma): cashflow statements importer + auto_verify/alert_generator observability - add backend/scripts/import_cashflow_statements.py: load zxt cashflow stat (CNY) into cashflow_statements in 10k CNY, dry-run default, --apply gated - auto_verify_cron: per-reason skip counters + legacy-condition rule now resolves kpi_code via plan.kpi_id (was unreachable) + --dry-run - alert_generator: coverage self-check warns when current-month KPI values are absent (0 alerts != no trigger) task: finance-20260916-cma-health-gaps-3items (items 1/2/3) --- backend/scripts/alert_generator.py | 49 +++++- backend/scripts/auto_verify_cron.py | 84 +++++++-- backend/scripts/import_cashflow_statements.py | 163 ++++++++++++++++++ 3 files changed, 275 insertions(+), 21 deletions(-) create mode 100644 backend/scripts/import_cashflow_statements.py diff --git a/backend/scripts/alert_generator.py b/backend/scripts/alert_generator.py index 6203e0a8..0f0850f5 100644 --- a/backend/scripts/alert_generator.py +++ b/backend/scripts/alert_generator.py @@ -217,6 +217,38 @@ def generate_and_push(db_session) -> dict: new_count = run_alert_check(db_session, period) logger.info(f"预警检查完成: 新增 {new_count} 条") + # 1a. 覆盖度自检（2026-09-16 新增，任务 finance-20260916-cma-health-gaps-3items · 项3）： + # run_alert_check 的期间口径 = 当月（period=current month），若当月尚无任何 KPIValue， + # 则「0 条新预警」并不代表"确无阈值触发"，而是"本次检查未覆盖任何数据"—— + # 该区别原实现不可见（日志与汇总行都只说 0 条），导致连续多日 0 新增无法归因。 + try: + kpis_with_values = 0 + active_kpis = db_session.query(KPIDefinition).filter(KPIDefinition.status == "active").all() + for k in active_kpis: + if not any((k.threshold_red, k.threshold_yellow, k.threshold_green)): + continue + hit = db_session.query(KPIValue).filter( + KPIValue.kpi_id == k.id, KPIValue.period == period, + KPIValue.actual_value.isnot(None), + ).first() + if hit: + kpis_with_values += 1 + coverage = {"period": period, "kpis_with_threshold_and_value": kpis_with_values} + if kpis_with_values == 0: + latest_period = db_session.query(KPIValue.period).filter( + KPIValue.actual_value.isnot(None), + KPIValue.period.op("REGEXP")("^[0-9]{4}-[0-9]{2}$"), # 只取 YYYY-MM 月期间（排除 2026-H1/H2 伪期间） + ).order_by(KPIValue.period.desc()).first() + logger.warning( + f"覆盖度告警: 当月({period})无任何 KPI 实际值 → 本次预警检查未覆盖任何 KPI，" + f"「0 条新预警」= 无数据可判，非「确无阈值触发」。库内最新期间=" + f"{latest_period[0] if latest_period else 'N/A'}" + ) + coverage["latest_period_in_db"] = latest_period[0] if latest_period else None + except Exception as e: + logger.error(f"覆盖度自检失败: {e}") + coverage = {"error": str(e)} + # 1b. 规则引擎检查（预测偏差 / 客户集中度，P2 2026-09-10） rule_stats = run_rule_engine_checks(db_session) @@ -231,7 +263,20 @@ def generate_and_push(db_session) -> dict: logger.error(f"推送失败: {e}") return {"period": period, "new_alerts": new_count, "pushed": pushed, - "cleaned_green": cleaned, "rule_engine": rule_stats} + "cleaned_green": cleaned, "rule_engine": rule_stats, + "coverage": coverage} + + +def coverage_note(result: dict) -> str: + """汇总行追加提示：当月无 KPI 值时显式说明「0 条 = 无数据可判」而非「确无触发」 + + 2026-09-16 新增（项3）。独立成函数便于回查（无需触发真实写路径即可断言文案）。 + """ + cov = result.get("coverage") or {} + if cov.get("kpis_with_threshold_and_value") == 0: + return (f" | ⚠️ 当月({result['period']})无KPI值: 本次检查未覆盖任何KPI" + f"（库内最新月期间 {cov.get('latest_period_in_db')}）→ 0 条不等于确无触发") + return "" if __name__ == "__main__": @@ -244,6 +289,6 @@ if __name__ == "__main__": try: result = generate_and_push(db) print(f"预警检查: {result['new_alerts']}条新预警 / {result['pushed']}条已推送" - f" / 规则引擎{result['rule_engine']}") + f" / 规则引擎{result['rule_engine']}{coverage_note(result)}") finally: db.close() diff --git a/backend/scripts/auto_verify_cron.py b/backend/scripts/auto_verify_cron.py index 90eefe40..fdc75c09 100644 --- a/backend/scripts/auto_verify_cron.py +++ b/backend/scripts/auto_verify_cron.py @@ -27,7 +27,19 @@ def evaluate_condition(condition, actual, target): return func(actual, target) -def run_auto_verify(): +def run_auto_verify(dry_run=False): + """扫描到期pending计划，按 auto_verify_rule 自动验证执行效果。 + + 2026-09-16 修复（任务 finance-20260916-cma-health-gaps-3items · 项2）： + ① 统计口径透明化——原汇总行「共处理N个到期计划」把「扫描到但全部 continue 跳过」的 + 计划也算作"处理"，导致连续 36 天「验证通过0个」无法区分空转与失效。现输出 + 逐原因计数（无规则 / 无KPI代码 / KPI定义未匹配 / 无KPI值 / 实际评估）。 + ② 旧表达式规则可达——原 `if not kpi_code: continue` 位于 evaluate_condition 的 + 旧表达式兼容分支之前，使 rule={"condition":"value < 75"}（无 kpi_code，仅挂 + plan.kpi_id）永远无法验证，该兼容分支成死代码。现先用 plan.kpi_id 反查 + kpi_code 再判定（只用于本轮取数，不回写 DB）。 + ③ 新增 --dry-run：只读复算、不写库（供巡检/复核留痕用）。 + """ db = next(get_db()) try: now = datetime.now() @@ -41,14 +53,26 @@ def run_auto_verify(): verified_count = 0 escalated_count = 0 messages = [] + # 跳过原因计数（2026-09-16 新增，用于区分「确无对象」与「判据失效」） + skip = {"no_rule": 0, "no_kpi_code": 0, "kpi_def_missing": 0, "no_kpi_value": 0} + evaluated = 0 for plan in plans: rule = plan.auto_verify_rule if not rule or not isinstance(rule, dict): + skip["no_rule"] += 1 continue kpi_code = rule.get("kpi_code") + if not kpi_code and plan.kpi_id: + # 旧表达式规则（只有 condition，无 kpi_code）→ 用计划关联 KPI 反查 + link_kpi = db.query(KPIDefinition).filter(KPIDefinition.id == plan.kpi_id).first() + if link_kpi and link_kpi.kpi_code: + kpi_code = link_kpi.kpi_code + rule = dict(rule) + rule["kpi_code"] = kpi_code if not kpi_code: + skip["no_kpi_code"] += 1 continue # 查最新KPI值 @@ -63,6 +87,7 @@ def run_auto_verify(): q = q.filter(KPIDefinition.entity_id == entity_id) kpi = q.order_by(KPIDefinition.id.desc()).first() if not kpi: + skip["kpi_def_missing"] += 1 continue # 缺陷3残留修复（对齐 verify.py）：KPIValue 按 period <= 当前月过滤，跨月验证不取未来期间 period_limit = now.strftime("%Y-%m") @@ -72,6 +97,7 @@ def run_auto_verify(): KPIValue.period <= period_limit, ).order_by(KPIValue.period.desc(), KPIValue.calculated_at.desc(), KPIValue.id.desc()).first() if not latest or latest.actual_value is None: + skip["no_kpi_value"] += 1 continue actual = latest.actual_value @@ -80,43 +106,58 @@ def run_auto_verify(): retry_max = rule.get("retry_max", 3) passed = evaluate_condition(condition, actual, target) + evaluated += 1 - plan.kpi_current_before = rule.get("baseline_value") - plan.kpi_current_after = actual + if not dry_run: + plan.kpi_current_before = rule.get("baseline_value") + plan.kpi_current_after = actual if passed: # 缺陷1残留修复（对齐 verify.py）：plan 已处于"验证通过"状态（passed + verified_at 非空）则跳过 OKR 累加，防重复累加 already_verified = bool(plan.verify_status == "passed" and plan.verified_at is not None) - plan.verify_status = "passed" - plan.verified_at = now - plan.status = "completed" # 缺陷4残留修复："done" 不在枚举(pending/in_progress/completed/cancelled)，改 completed - plan.progress = 100 - # 阶段3: OKR progress更新 - update_okr_progress(db, plan, +15, already_verified=already_verified) + if not dry_run: + plan.verify_status = "passed" + plan.verified_at = now + plan.status = "completed" # 缺陷4残留修复："done" 不在枚举(pending/in_progress/completed/cancelled)，改 completed + plan.progress = 100 + # 阶段3: OKR progress更新 + update_okr_progress(db, plan, +15, already_verified=already_verified) messages.append(f"✅ 行动计划#{plan.id}验证通过: {plan.title} ({kpi_code}: {rule.get('baseline_value')}→{actual})") verified_count += 1 else: - plan.verify_attempts = (plan.verify_attempts or 0) + 1 - if plan.verify_attempts >= retry_max: - plan.verify_status = "escalated" + attempts = (plan.verify_attempts or 0) + (0 if dry_run else 1) + if not dry_run: + plan.verify_attempts = attempts + if attempts >= retry_max: + if not dry_run: + plan.verify_status = "escalated" messages.append(f"🚨 行动计划#{plan.id}连续{retry_max}次验证失败，升级人工: {plan.title}") escalated_count += 1 else: - plan.verify_status = "retrying" - messages.append(f"⚠️ 行动计划#{plan.id}验证失败(第{plan.verify_attempts}次)，待重试: {plan.title}") + if not dry_run: + plan.verify_status = "retrying" + messages.append(f"⚠️ 行动计划#{plan.id}验证失败(第{attempts}次)，待重试: {plan.title}") - db.commit() + if not dry_run: + db.commit() # 汇总通知 - if messages: + if messages and not dry_run: summary = "\n".join(messages) try: send_wecom_message(f"📋 行动计划自动验证日报\n{summary}", title="Auto-Verify") except Exception: pass - print(f"[{now.strftime('%Y-%m-%d %H:%M')}] 验证通过{verified_count}个，升级{escalated_count}个，共处理{len(plans)}个到期计划") - return {"verified": verified_count, "escalated": escalated_count, "total": len(plans)} + # 汇总行：保留「验证通过N个」前缀（巡检脚本 cma-daily-healthcheck.py 按该字面量取数）， + # 追加逐原因明细 —— 使「0 通过」可区分为「无验证对象」还是「判据失效」 + print(f"[{now.strftime('%Y-%m-%d %H:%M')}] 验证通过{verified_count}个，升级{escalated_count}个，" + f"共处理{len(plans)}个到期计划 | 实际评估{evaluated}个；跳过明细: " + f"无规则{skip['no_rule']}个/无KPI代码{skip['no_kpi_code']}个/" + f"KPI定义未匹配{skip['kpi_def_missing']}个/无KPI值{skip['no_kpi_value']}个" + + ("（dry-run 未写库）" if dry_run else "")) + return {"verified": verified_count, "escalated": escalated_count, "total": len(plans), + "evaluated": evaluated, "skip": skip, "dry_run": dry_run} finally: db.close() @@ -142,4 +183,9 @@ def update_okr_progress(db, plan, delta, already_verified=False): if __name__ == "__main__": - run_auto_verify() + import sys as _sys + _dry = "--dry-run" in _sys.argv + res = run_auto_verify(dry_run=_dry) + if _dry: + import json as _json + print(_json.dumps(res, ensure_ascii=False)) diff --git a/backend/scripts/import_cashflow_statements.py b/backend/scripts/import_cashflow_statements.py new file mode 100644 index 00000000..7a4eaf08 --- /dev/null +++ b/backend/scripts/import_cashflow_statements.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""现金流量表（三大活动净额）入库器 —— 酣客(智享通) 现金流量统计表 → cashflow_statements + +背景（任务 finance-20260916-cma-health-gaps-3items · 项1）： + cashflow_statements 表 0 行 = **无导入链路**（此前只有 API 读/写端点，无任何 loader； + 上游 智享通 数据虽已落盘 /root/WB_WS/_data/zxt-hanke/cashflow//，但从未入库）。 + +数据源（权威口径 = 智享通「现金流量统计表」项目级汇总，单位：元）： + /root/WB_WS/_data/zxt-hanke/cashflow//cashflow_stat.csv + 列：分类,编码,项目名称,方向,金额 分类∈{经营活动,投资活动,筹资活动} + 兼容 2026.08 的另存名 cashflow_stat_.csv + +口径： + - 经营/投资/筹资 净额 = Σ(流入) − Σ(流出)，单位 元 → 万元（÷10000，保留 2 位） + - 某活动无项目行 → 记 0.00（表列为 NOT NULL；同时写 remark 说明"无该项目=0"） + - source='official'（官方报表口径）；期初/期末现金 留空（主表采集缺陷，见报告 §项1-②） + - 同 (entity_id, period) 重复执行 = 覆盖更新（幂等） + +用法： + python3 scripts/import_cashflow_statements.py # dry-run（默认，只算不写） + python3 scripts/import_cashflow_statements.py --period 2026-07 # 单期 dry-run + python3 scripts/import_cashflow_statements.py --apply # 落库（需放行） +""" +import argparse +import csv +import glob +import os +import sys + +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 CashFlowStructureValue +from app.api.cashflow_structure import classify_enterprise + +ZKT_BASE = "/root/WB_WS/_data/zxt-hanke/cashflow" +CLASSES = ("经营活动", "投资活动", "筹资活动") + + +def _read_stat_file(path): + """读项目级汇总 CSV → [(分类, 编码, 项目名称, 方向, 金额元)]""" + rows = [] + with open(path, encoding="utf-8-sig") as fh: + for r in csv.reader(fh): + if len(r) < 5 or not r[0].strip() or r[1].strip() == "编码": + continue + try: + amt = float(r[4].strip().replace(",", "")) + except ValueError: + continue + rows.append((r[0].strip(), r[1].strip(), r[2].strip(), r[3].strip(), amt)) + return rows + + +def _nets(rows): + agg = {c: {"in": 0.0, "out": 0.0} for c in CLASSES} + for cls, _code, _name, direction, amt in rows: + if cls not in agg: + continue + agg[cls]["in" if "流入" in direction else "out"] += amt + return {c: round(v["in"] - v["out"], 2) for c, v in agg.items()}, agg + + +def collect(period_filter=None): + """扫描盘上各期 → [payload dict]（万元，未写库）""" + payloads = [] + for d in sorted(glob.glob(os.path.join(ZKT_BASE, "2026.*"))): + period = os.path.basename(d).replace(".", "-") + if period_filter and period != period_filter: + continue + cands = sorted(glob.glob(os.path.join(d, "cashflow_stat*.csv"))) + # 优先取当期明细名（cashflow_stat_.csv），其次通用名 + src = None + for c in cands: + if os.path.basename(c) != "cashflow_stat.csv": + src = c + src = src or (cands[0] if cands else None) + if not src: + payloads.append({"period": period, "status": "SKIP_no_source_file", "dir": d}) + continue + rows = _read_stat_file(src) + if not rows: + payloads.append({"period": period, "status": "SKIP_empty", + "src": src, "rows": 0}) + continue + nets_yuan, agg = _nets(rows) + missing = [c for c in CLASSES if not agg[c]["in"] and not agg[c]["out"]] + remark = ("源: 智享通现金流量统计表 %s（单位元→万元 /10000）；项目行 %d 条；缺项目活动: %s" + % (os.path.basename(src), len(rows), "、".join(missing) or "无")) + payloads.append({ + "period": period, + "status": "READY", + "src": src, + "rows": len(rows), + "nets_yuan": nets_yuan, + "nets_wan": {k: round(v / 10000.0, 2) for k, v in nets_yuan.items()}, + "missing_classes": missing, + "remark": remark, + }) + return payloads + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--entity-id", type=int, default=1, + help="账套：1=酣客(白酒, 智享通通道) 2=博海(IT, 无凭证明细)") + ap.add_argument("--period", default=None, help="仅处理某期 YYYY-MM") + ap.add_argument("--apply", action="store_true", help="真正写库（缺省=dry-run 只算不写）") + args = ap.parse_args() + + payloads = collect(args.period) + if not payloads: + print("未发现任何期数据源（%s/2026.*）" % ZKT_BASE) + return 2 + + mode = "APPLY" if args.apply else "DRY-RUN" + print("=== 现金流量表入库器 %s | entity_id=%d ===" % (mode, args.entity_id)) + db = get_session_local()() if args.apply else None + written = 0 + try: + for p in payloads: + if p["status"] != "READY": + print(" %-8s %s (%s)" % (p["period"], p["status"], p.get("src") or p.get("dir"))) + continue + n = p["nets_wan"] + cls = classify_enterprise(n["经营活动"], n["投资活动"], n["筹资活动"]) + print(" %-8s O=%8.2f I=%8.2f F=%8.2f 万 | 判定=%s | 源=%s" + % (p["period"], n["经营活动"], n["投资活动"], n["筹资活动"], + ("类型%d %s" % (cls["type_no"], cls["name"])) if cls.get("classified") + else ("未判定: " + cls.get("reason", "")[:28]), + os.path.basename(p["src"]))) + if not args.apply: + continue + rec = db.query(CashFlowStructureValue).filter( + CashFlowStructureValue.entity_id == args.entity_id, + CashFlowStructureValue.period == p["period"], + ).first() + created = rec is None + if rec is None: + rec = CashFlowStructureValue(entity_id=args.entity_id, period=p["period"]) + db.add(rec) + rec.operating_net = n["经营活动"] + rec.investing_net = n["投资活动"] + rec.financing_net = n["筹资活动"] + rec.source = "official" + rec.remark = p["remark"] + db.flush() + written += 1 + print(" → %s (id=%s)" % ("入库" if created else "更新", rec.id)) + if args.apply: + db.commit() + print("=== 已提交 %d 期 ===" % written) + else: + print("=== dry-run 结束：未写库（%d 期可入库，加 --apply 落库）===" % len( + [p for p in payloads if p["status"] == "READY"])) + finally: + if db is not None: + db.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main())