"""预算↔现金流↔行动 三闭环异常自检 — 路线图R5 (2026-08-30) 预算闭环加固：预算执行率异常（<70% 或 >110%）的KPI， 检查是否同步了 现金流计划(CashPlan) 和 行动方案(ActionPlan)， 缺失则输出提示（防止"预算改了，现金流/行动没跟上"）。 输出：控制台 + reports/closed_loop_check_YYYYMMDD.md 用法: /root/cma-management/backend/venv/bin/python3 scripts/closed_loop_check.py [--period 2026-08] [--push] """ import sys import os import json import argparse from datetime import datetime 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 KPIDefinition, KPIValue, BudgetPlan, CashPlan, ActionPlan LOW_RATIO = 0.7 HIGH_RATIO = 1.1 REPORTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "reports") def check_entity(db, entity_id: int, period: str) -> dict: """检测一个账套的闭环状态""" issues = [] rows = db.query(BudgetPlan).filter( BudgetPlan.entity_id == entity_id, BudgetPlan.status == "active", BudgetPlan.period == period, BudgetPlan.budget_value > 0, ).all() seen = set() for b in rows: key = (b.kpi_id, b.period) if key in seen: continue seen.add(key) k = db.query(KPIDefinition).filter(KPIDefinition.id == b.kpi_id).first() kpi_name = k.kpi_name if k else f"KPI#{b.kpi_id}" actual = db.query(KPIValue).filter( KPIValue.kpi_id == b.kpi_id, KPIValue.period == b.period, KPIValue.actual_value.isnot(None), ).order_by(KPIValue.calculated_at.desc()).first() actual_val = actual.actual_value if actual else None if actual_val is None: continue ratio = actual_val / b.budget_value abnormal = ratio < LOW_RATIO or ratio > HIGH_RATIO if not abnormal: continue # 现金流检查：该KPI该期间是否有收付款计划（related_kpi_id 或 budget_plan_id 关联） period_start = datetime.strptime(period + "-01", "%Y-%m-%d") if period.endswith("-12"): period_end = datetime(period_start.year + 1, 1, 1) else: period_end = datetime(period_start.year, period_start.month + 1, 1) cash_plans = db.query(CashPlan).filter( CashPlan.entity_id == entity_id, CashPlan.status.in_(["pending", "completed"]), CashPlan.plan_date >= period_start, CashPlan.plan_date < period_end, ).filter( (CashPlan.related_kpi_id == b.kpi_id) | (CashPlan.budget_plan_id == b.id) ).count() # 兜底：无关联但期间内有任意现金流计划也算基本闭环 any_cash = db.query(CashPlan).filter( CashPlan.entity_id == entity_id, CashPlan.status.in_(["pending", "completed"]), CashPlan.plan_date >= period_start, CashPlan.plan_date < period_end, ).count() # 行动检查：该KPI是否有非完成的行动方案 actions = db.query(ActionPlan).filter( ActionPlan.kpi_id == b.kpi_id, ActionPlan.status.in_(["pending", "in_progress"]), ).count() missing = [] if cash_plans == 0: if any_cash > 0: missing.append("现金流(本期间有其他计划但未关联本KPI)") else: missing.append("现金流") if actions == 0: missing.append("行动方案") level = "critical" if ratio > HIGH_RATIO else "warning" issues.append({ "kpi_id": b.kpi_id, "kpi_name": kpi_name, "period": period, "budget_value": b.budget_value, "actual_value": actual_val, "exec_ratio": round(ratio * 100, 1), "abnormal_type": "超预算" if ratio > HIGH_RATIO else "低执行", "level": level, "cash_plan_count": cash_plans, "action_plan_count": actions, "missing": missing, "suggestion": ( f"预算执行率{ratio*100:.0f}%异常，请同步" + ("现金流计划" if "现金流" in missing else "现金流情况核对") + ("、行动方案" if "行动方案" in missing else "") + f"（{kpi_name} {period}）" ), }) return {"entity_id": entity_id, "period": period, "issues": issues} def build_report(results: list, checked_at: str) -> str: lines = [f"# 预算↔现金流↔行动 闭环自检报告", f"**检查时间**: {checked_at}", ""] total_issues = 0 for r in results: lines.append(f"## 账套 #{r['entity_id']} · 期间 {r['period']}") if not r["issues"]: lines.append("- ✅ 无预算执行率异常") for it in r["issues"]: total_issues += 1 icon = "🔴" if it["level"] == "critical" else "🟡" lines.append(f"- {icon} {it['kpi_name']}（{it['period']}）") lines.append(f" 预算 {it['budget_value']:g} / 实际 {it['actual_value']:g} = 执行率 {it['exec_ratio']}%（{it['abnormal_type']}）") lines.append(f" 现金流计划: {it['cash_plan_count']} 条 | 行动方案: {it['action_plan_count']} 条") if it["missing"]: lines.append(f" ⚠️ 缺失: {'、'.join(it['missing'])}") lines.append(f" 💡 {it['suggestion']}") else: lines.append(f" ✅ 三闭环已同步") lines.append("") lines.append(f"---") lines.append(f"共发现异常 {total_issues} 项") return "\n".join(lines) def main(): parser = argparse.ArgumentParser() parser.add_argument("--period", default=datetime.now().strftime("%Y-%m")) parser.add_argument("--entity-id", type=int, default=1) parser.add_argument("--push", action="store_true", help="异常时推送企微(8800/send)") args = parser.parse_args() os.makedirs(REPORTS_DIR, exist_ok=True) checked_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") db = get_session_local()() try: result = check_entity(db, args.entity_id, args.period) report = build_report([result], checked_at) print(report) # 写报告文件 fname = f"closed_loop_check_{datetime.now().strftime('%Y%m%d')}.md" fpath = os.path.join(REPORTS_DIR, fname) with open(fpath, "w", encoding="utf-8") as f: f.write(report) print(f"\n📄 报告已写入: {fpath}") # 异常推送 if args.push and result["issues"]: try: import urllib.request import urllib.parse content = f"## 🔄 预算闭环自检({args.period})\n" for it in result["issues"][:10]: content += f"- {it['kpi_name']} 执行率{it['exec_ratio']}% 缺{'/'.join(it['missing']) or '无'}\n" content += f"\n共{len(result['issues'])}项异常，详见系统报告" data = urllib.parse.urlencode({"msg": content, "source": "管理会计OS"}).encode("utf-8") req = urllib.request.Request("http://127.0.0.1:8800/send", data=data) with urllib.request.urlopen(req, timeout=15) as resp: print("推送:", resp.read().decode()[:200]) except Exception as e: print(f"推送失败: {e}") finally: db.close() if __name__ == "__main__": main()