diff --git a/backend/app/api/verify.py b/backend/app/api/verify.py new file mode 100644 index 00000000..e6745e7d --- /dev/null +++ b/backend/app/api/verify.py @@ -0,0 +1,299 @@ +""" +自动验证引擎 API — 管理会计OS +POST /api/cma/verify/{plan_id} 手动验证行动计划执行结果 + +数据流: + ActionPlan.auto_verify_rule (JSON) → 条件判断 → passed/failed + → KPI值回填 (kpi_current_before/after + KPIValue source_type=verify) + → 验证通过 → 所属OKR progress +15% + → 通知任总 (send_wecom_message) + +规则格式(新): + { + "kpi_code": "C_REBATE_RATE", + "condition": "LESS_THAN", # LESS_THAN/GREATER_THAN/WITHIN_RANGE/NOT_NULL + "target_value": 80, + "baseline_value": 86.4, + "verify_after_days": 7, + "retry_max": 3, + "escalate_to": "任富海", + "notify": true + } + +兼容格式(旧): + {"condition": "value < 75", "description": "..."} → 走 bot_bridge_v2._evaluate_condition +""" +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy.orm import Session +from datetime import datetime, timedelta +import json +import logging + +from app.database import get_db +from app.models import ActionPlan, KPIDefinition, KPIValue, Objective +from app.utils.notifier import send_wecom_message +from app.auth_middleware import require_role + +logger = logging.getLogger("cma.verify") + +router = APIRouter( + prefix="/api/cma/verify", + tags=["自动验证"], + dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], +) + +CONDITION_FUNCS = { + "LESS_THAN": lambda actual, target: actual is not None and actual < target, + "GREATER_THAN": lambda actual, target: actual is not None and actual > target, + "WITHIN_RANGE": lambda actual, target: target is not None and len(target) == 2 and float(target[0]) <= actual <= float(target[1]), + "NOT_NULL": lambda actual, target: actual is not None, +} + + +def _parse_rule(rule) -> dict: + """解析auto_verify_rule(兼容JSON字符串)""" + if rule is None: + return {} + if isinstance(rule, str): + try: + return json.loads(rule) + except Exception: + return {"condition": rule} + return rule + + +def evaluate_rule(rule: dict, actual, kpi_data: dict = None) -> bool: + """ + 按规则判断actual是否达标 + - 新格式: condition为枚举 LESS_THAN/GREATER_THAN/WITHIN_RANGE/NOT_NULL + - 旧格式: condition为表达式字符串(如 "value < 75")→ bot_bridge_v2引擎 + """ + condition = (rule or {}).get("condition", "") + if not condition: + return actual is not None + + if condition in CONDITION_FUNCS: + target = rule.get("target_value") + return CONDITION_FUNCS[condition](actual, target) + + # 旧格式表达式(bot-bridge兼容) + try: + from app.api.bot_bridge_v2 import _evaluate_condition + kd = kpi_data or {"value": actual, "target": rule.get("target_value"), "baseline": rule.get("baseline_value")} + return _evaluate_condition(condition, kd) + except Exception as e: + logger.warning(f"旧格式条件评估失败({condition}): {e}") + return False + + +def update_okr_progress(db: Session, plan: ActionPlan) -> dict: + """验证通过 → 所属OKR progress +15%(每通过1个KR)""" + if not plan.objective_id: + return {"updated": False, "reason": "no_objective"} + obj = db.query(Objective).filter(Objective.id == plan.objective_id).first() + if not obj: + return {"updated": False, "reason": "objective_not_found"} + before = obj.progress or 0 + obj.progress = min(100, before + 15) + db.flush() + return {"updated": True, "objective_id": obj.id, "before": before, "after": obj.progress} + + +def build_auto_verify_rule(kpi, baseline_value=None, verify_after_days: int = 7) -> dict: + """根据KPI阈值自动生成验证规则(验收#1: 创建ActionPlan自动带auto_verify_rule) + + 从KPI的绿灯阈值(green)推导达标方向: + - 绿灯 '<=X' → 目标是把值压到 X 以下 → LESS_THAN X + - 绿灯 '>=X' → 目标是把值抬到 X 以上 → GREATER_THAN X + - 有绿色区间 'X~Y' → WITHIN_RANGE + - 无阈值 → NOT_NULL + """ + expr = (kpi.threshold_green or "").strip() + condition = "NOT_NULL" + target = None + import re + m = re.match(r"^(<=|>=|<|>|=)\s*([\d.]+)$", expr) + if m: + op, val = m.group(1), float(m.group(2)) + if op in ("<", "<="): + condition = "LESS_THAN" + elif op in (">", ">="): + condition = "GREATER_THAN" + else: + condition = "NOT_NULL" + target = val + elif "~" in expr: + parts = expr.split("~") + if len(parts) == 2: + try: + condition = "WITHIN_RANGE" + target = [float(parts[0]), float(parts[1])] + except ValueError: + pass + else: + # 无阈值表达式 → 用KPI目标值推方向 + condition = "NOT_NULL" + target = None + + return { + "kpi_code": kpi.kpi_code, + "condition": condition, + "target_value": target, + "baseline_value": float(baseline_value) if baseline_value is not None else None, + "verify_after_days": verify_after_days, + "retry_max": 3, + "escalate_to": "任富海", + "notify": True, + } + + +def backfill_kpi_value(db: Session, plan: ActionPlan, actual, rule: dict, source: str = "verify"): + """回填KPI当前值: 写入kpi_current_before/after + 新KPIValue记录""" + kpi = None + kpi_code = rule.get("kpi_code") if rule else None + # 优先按验证规则指定的KPI编码查询;无规则时才回退到plan.kpi_id + if kpi_code: + kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == kpi_code).first() + elif plan.kpi_id: + kpi = db.query(KPIDefinition).filter(KPIDefinition.id == plan.kpi_id).first() + + if actual is None: + return None + + plan.kpi_current_before = plan.kpi_current_before if plan.kpi_current_before is not None else rule.get("baseline_value") + plan.kpi_current_after = actual + + if kpi: + new_val = KPIValue( + kpi_id=kpi.id, + period=datetime.now().strftime("%Y-%m"), + actual_value=actual, + source_type="verify", + source_batch=f"verify-plan-{plan.id}", + data_status="verified", + calculated_at=datetime.now(), + remark=f"行动计划#{plan.id}验证回填", + ) + db.add(new_val) + return {"kpi_id": kpi.id, "kpi_code": kpi.kpi_code} + return None + + +@router.post("/{plan_id}") +def verify_action_plan(plan_id: int, payload: dict, db: Session = Depends(get_db)): + """验证行动计划执行结果(手动验证 / Bot回填) + + 请求体: + { + "actual_value": 78.5, # 可选,缺省时取KPI最新值 + "source": "财务Bot分析", # 来源 + "note": "渠补谈判后...", # 备注/验证结果详情 + "passed": true # 可选,无规则时手动指定 + } + """ + plan = db.query(ActionPlan).filter(ActionPlan.id == plan_id).first() + if not plan: + raise HTTPException(404, "行动计划不存在") + + rule = _parse_rule(plan.auto_verify_rule) + actual = payload.get("actual_value") + note = payload.get("note", "") + source = payload.get("source", "手动验证") + + # 1. 无规则 → 按手动指定 passed 标记 + if not rule or not rule.get("condition"): + passed = payload.get("passed", True) + plan.verify_status = "passed" if passed else "failed" + plan.verify_result = "pass" if passed else "fail" + plan.verify_log = (plan.verify_log or []) + [{ + "timestamp": datetime.now().isoformat(), + "passed": passed, + "note": note, + "source": source, + }] + plan.verified_at = datetime.now() + db.commit() + return {"plan_id": plan_id, "verify_status": plan.verify_status, "passed": passed} + + # 2. 缺省actual → 取KPI最新值 + if actual is None: + kpi = None + kpi_code = rule.get("kpi_code") + if plan.kpi_id: + kpi = db.query(KPIDefinition).filter(KPIDefinition.id == plan.kpi_id).first() + elif kpi_code: + kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == kpi_code).first() + if kpi: + latest = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi.id, + KPIValue.actual_value.isnot(None), + ).order_by(KPIValue.calculated_at.desc(), KPIValue.id.desc()).first() + if latest: + actual = latest.actual_value + + # 3. 执行条件判断 + passed = evaluate_rule(rule, actual, kpi_data={"value": actual, "target": rule.get("target_value"), "baseline": rule.get("baseline_value")}) + + # 4. 回写KPI当前值 + backfill = backfill_kpi_value(db, plan, actual, rule, source=source) + + # 5. 更新状态 + plan.verify_status = "passed" if passed else "failed" + plan.verify_result = "pass" if passed else "fail" + plan.verify_log = (plan.verify_log or []) + [{ + "timestamp": datetime.now().isoformat(), + "condition": rule.get("condition"), + "actual_value": actual, + "target_value": rule.get("target_value"), + "passed": passed, + "source": source, + "note": note, + }] + if passed: + plan.status = "done" + plan.progress = 100 + plan.verified_at = datetime.now() + + # 6. OKR进度联动(验证通过 → +15%) + okr_update = None + if passed: + okr_update = update_okr_progress(db, plan) + + db.commit() + + # 7. 通知任总 + if rule.get("notify", True): + delta = "" + if plan.kpi_current_before is not None and actual is not None: + try: + d = float(actual) - float(plan.kpi_current_before) + delta = f"({d:+.1f})" + except (TypeError, ValueError): + delta = "" + title = f"✅ 行动计划#{plan.id}验证通过" if passed else f"❌ 行动计划#{plan.id}验证失败" + content = ( + f"{plan.title}\n" + f"KPI: {rule.get('kpi_code', '')} {plan.kpi_current_before}→{actual} {delta}\n" + f"规则: {rule.get('condition')} {rule.get('target_value')}\n" + f"来源: {source}\n" + f"{note}" + ) + try: + send_wecom_message(content=content, title=title, alert_level="green" if passed else "red") + except Exception as e: + logger.warning(f"验证通知发送失败: {e}") + + return { + "plan_id": plan_id, + "verify_status": plan.verify_status, + "kpi_current_before": plan.kpi_current_before, + "kpi_current_after": plan.kpi_current_after, + "improvement": f"{float(actual) - float(plan.kpi_current_before):+.1f}" if plan.kpi_current_before is not None and actual is not None else None, + "passed": passed, + "okr_progress": okr_update, + "kpi_backfill": backfill, + "message": ( + f"验证通过:{plan.title}" if passed + else f"验证失败:{plan.title}(条件 {rule.get('condition')} {rule.get('target_value')} 未达成,当前值 {actual})" + ), + } diff --git a/backend/app/main.py b/backend/app/main.py index d8686f34..d13a33c3 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -5,7 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from dotenv import load_dotenv from app.database import init_db -from app.api import auth, kpis, templates, maps, dashboard, data, alerts, ai_analysis, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, bot_iron_law, analysis_results, expenses, cash, tax_compliance +from app.api import auth, kpis, templates, maps, dashboard, data, alerts, ai_analysis, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync from app.auth_middleware import require_auth @@ -77,6 +77,7 @@ app.include_router(analysis_results.router) app.include_router(expenses.router) app.include_router(cash.router) app.include_router(tax_compliance.router) +app.include_router(verify.router) @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 4d02f9ce..0d6e125f 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -210,6 +210,11 @@ class ActionPlan(Base): auto_verify_rule = Column(JSON, nullable=True, comment="自动验证规则: {\"condition\": \"value > target\"}") verify_result = Column(String(20), nullable=True, comment="验证结果: pass/fail/pending") verify_log = Column(JSON, nullable=True, comment="验证历史日志") + verify_status = Column(String(20), default="pending", comment="验证状态: pending/passed/failed/retrying/escalated") + verify_attempts = Column(Integer, default=0, comment="验证尝试次数") + verified_at = Column(DateTime, nullable=True, comment="验证完成时间") + kpi_current_before = Column(Float, nullable=True, comment="执行前KPI值") + kpi_current_after = Column(Float, nullable=True, comment="执行后KPI值") created_by = Column(String(100), nullable=True, comment="创建人") created_at = Column(DateTime, server_default=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) diff --git a/backend/app/utils/__pycache__/notifier.cpython-312.pyc b/backend/app/utils/__pycache__/notifier.cpython-312.pyc index e8c7307a..1237f410 100644 Binary files a/backend/app/utils/__pycache__/notifier.cpython-312.pyc and b/backend/app/utils/__pycache__/notifier.cpython-312.pyc differ diff --git a/backend/app/utils/notifier.py b/backend/app/utils/notifier.py index d0a202d3..26050ca1 100644 --- a/backend/app/utils/notifier.py +++ b/backend/app/utils/notifier.py @@ -122,6 +122,58 @@ def send_mail(smtp_config: dict, to_addrs: list, title: str, content: str) -> di # 主推送函数 # ============================================================ +def send_wecom_message(content: str, title: str = "管理会计OS通知", alert_level: str = "green") -> dict: + """发送企业微信消息到所有已启用的通知渠道(验证结果/行动报告推送用) + + - 读取 notification_channels 表中启用的 wecom_app / wecom 渠道 + - 无渠道时记日志并返回失败(不抛异常,保证主流程不中断) + """ + try: + from app.database import get_session_local + from app.models import NotificationChannel + except Exception as e: + logger.warning(f"send_wecom_message: 加载依赖失败: {e}") + return {"success": False, "message": f"依赖加载失败: {e}"} + + db = get_session_local()() + try: + channels = db.query(NotificationChannel).filter( + NotificationChannel.enabled == True, + NotificationChannel.channel_type.in_(["wecom_app", "wecom"]), + ).all() + if not channels: + logger.info(f"send_wecom_message: 无已启用的企微渠道,跳过推送: {title}") + return {"success": False, "message": "无已启用的企微通知渠道"} + + results = [] + for ch in channels: + config = ch.config or {} + if ch.channel_type == "wecom_app": + r = send_wecom_app( + corp_id=config.get("corp_id", ""), + corp_secret=config.get("corp_secret", ""), + agent_id=config.get("agent_id", ""), + touser=config.get("touser", "@all"), + title=title, content=content, alert_level=alert_level, + ) + else: + r = send_wecom_robot( + webhook_url=config.get("webhook_url", ""), + title=title, content=content, alert_level=alert_level, + ) + r["channel"] = ch.channel_type + results.append(r) + return { + "success": any(r.get("success") for r in results), + "results": results, + } + except Exception as e: + logger.error(f"send_wecom_message 异常: {e}") + return {"success": False, "message": f"推送异常: {e}"} + finally: + db.close() + + def push_alert(alert: dict, channels: list[dict]) -> list[dict]: """向所有已启用渠道推送一条预警""" results = [] diff --git a/backend/scripts/action_plan_weekly.py b/backend/scripts/action_plan_weekly.py new file mode 100644 index 00000000..08f88dbd --- /dev/null +++ b/backend/scripts/action_plan_weekly.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""行动计划执行周报 — 每周一08:00推送 + +统计本周通过/失败/升级/待处理计划数,推送任总 +""" +import sys, os +from datetime import datetime, timedelta + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.database import get_db +from app.models import ActionPlan +from app.utils.notifier import send_wecom_message + + +def generate_weekly_report(): + db = next(get_db()) + try: + week_ago = datetime.now() - timedelta(days=7) + # 本周所有计划(最近7天创建的) + plans = db.query(ActionPlan).filter(ActionPlan.created_at >= week_ago).all() + if not plans: + plans = db.query(ActionPlan).order_by(ActionPlan.created_at.desc()).limit(10).all() + + total = len(plans) + passed = sum(1 for p in plans if p.verify_status == "passed") + failed = sum(1 for p in plans if p.verify_status == "failed") + escalated = sum(1 for p in plans if p.verify_status == "escalated") + pending = sum(1 for p in plans if p.verify_status in ("pending", "retrying")) + done = sum(1 for p in plans if p.status == "done") + + msg = ( + f"📋 行动计划执行周报({datetime.now().strftime('%Y-%m-%d')})\n" + f"统计范围:近7天计划 {total} 个\n" + f"✅ 验证通过: {passed}\n" + f"❌ 验证失败: {failed}\n" + f"🚨 升级人工: {escalated}\n" + f"⏳ 待验证: {pending}\n" + f"🏁 已完成: {done}\n" + ) + # 列出待验证/升级项 + urgent = [p for p in plans if p.verify_status in ("escalated", "retrying")] + if urgent: + msg += "\n⚠️ 需关注:\n" + for p in urgent[:5]: + msg += f" #{p.id} {p.title} ({p.verify_status})\n" + + try: + send_wecom_message(msg, title="行动计划周报") + except Exception: + pass + print(msg) + return {"total": total, "passed": passed, "failed": failed, "escalated": escalated} + finally: + db.close() + + +if __name__ == "__main__": + generate_weekly_report() diff --git a/backend/scripts/alert_generator.py b/backend/scripts/alert_generator.py index d65048f1..7262123a 100644 --- a/backend/scripts/alert_generator.py +++ b/backend/scripts/alert_generator.py @@ -9,12 +9,12 @@ import os import re import json import logging -from datetime import datetime +from datetime import datetime, timedelta 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, KPIAlert +from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan logger = logging.getLogger("cma.alert_gen") @@ -33,6 +33,24 @@ def _parse_threshold(expr: str) -> tuple: return None, None +def cleanup_green_alerts(db_session) -> int: + """Bug1修复: 清理历史遗留的green级别pending预警(正常值不应出现在待办)""" + greens = db_session.query(KPIAlert).filter( + KPIAlert.alert_level == "green", + KPIAlert.status.in_(["pending", "processing"]), + ).all() + count = 0 + for g in greens: + g.status = "resolved" + g.resolution = "自动清理: green(正常)级别不生成待处理预警" + g.resolved_at = datetime.now() + count += 1 + if count: + db_session.commit() + logger.info(f" 清理历史green预警 {count} 条 → resolved") + return count + + def _check_threshold(actual: float, threshold_expr: str, level: str) -> tuple: """检查实际值是否触发阈值,返回 (触发, 消息)""" if not threshold_expr or actual is None: @@ -95,12 +113,16 @@ def run_alert_check(db_session, period: str = None) -> int: actual = latest.actual_value # 从绿到红检查(绿灯最高优先级——满足即止) + # Bug1修复: green(正常)不生成pending预警,只对yellow/red落库 for level in ["green", "yellow", "red"]: expr = thr[level] if not expr: continue triggered, msg = _check_threshold(actual, expr, level) if triggered: + if level == "green": + # 正常状态:不生成待处理预警(如有旧的red/yellow预警自动解除) + break # 检查是否已有该期间该KPI同级别的预警 existing = db_session.query(KPIAlert).filter( KPIAlert.kpi_id == kpi.id, @@ -121,8 +143,34 @@ def run_alert_check(db_session, period: str = None) -> int: status="pending", ) db_session.add(alert) + db_session.flush() # 获取alert.id new_count += 1 logger.info(f" 新增预警 [{level}] {kpi.kpi_name}: {msg}") + + # Bug2修复: red/yellow预警自动创建行动计划(预警→行动闭环) + if level in ("red", "yellow"): + plan_title = f"改善: {kpi.kpi_name} {msg}" + if kpi.kpi_name and "现金" in kpi.kpi_name or kpi.kpi_code == "F_OP_CFLOW": + plan_title = f"资金缺口应对: {msg} —— 催收应收+安排融资" + elif kpi.kpi_code == "F_AR_DAYS": + plan_title = f"应收催收: {msg} —— 逾期客户专项跟进" + elif kpi.kpi_code == "C_REBATE_RATE": + plan_title = f"渠补谈判: {msg} —— 下游渠道返利谈判" + elif kpi.kpi_code == "F_COST_RATIO": + plan_title = f"费用压缩: {msg} —— 管理费削减计划" + plan = ActionPlan( + alert_id=alert.id, + kpi_id=kpi.id, + title=plan_title, + assignee="财务部", + priority="high" if level == "red" else "medium", + due_date=datetime.now().date() + timedelta(days=3 if level == "red" else 7), + created_by="alert_generator", + ) + db_session.add(plan) + db_session.flush() + alert.action_plan_linked_id = plan.id + logger.info(f" ↳ 自动创建行动计划 #{plan.id}: {plan_title}") break # 只取最高级别 elif level == "red" and not triggered: # 红没触发,如果已有红色预警但当前不满足,自动降级或关闭 @@ -146,6 +194,9 @@ def generate_and_push(db_session) -> dict: """生成预警并推送,返回统计""" period = datetime.now().strftime("%Y-%m") + # 0. Bug1修复: 先清理历史green预警 + cleaned = cleanup_green_alerts(db_session) + # 1. 生成预警 logger.info(f"开始预警检查 ({period})...") new_count = run_alert_check(db_session, period) @@ -161,7 +212,7 @@ def generate_and_push(db_session) -> dict: except Exception as e: logger.error(f"推送失败: {e}") - return {"period": period, "new_alerts": new_count, "pushed": pushed} + return {"period": period, "new_alerts": new_count, "pushed": pushed, "cleaned_green": cleaned} if __name__ == "__main__": diff --git a/backend/scripts/auto_verify_cron.py b/backend/scripts/auto_verify_cron.py new file mode 100644 index 00000000..bee28a49 --- /dev/null +++ b/backend/scripts/auto_verify_cron.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Auto-Verify 定时验证任务 — 扫描到期pending计划,自动验证执行效果 + +运行: 30 9 * * * cd /root/cma-management/backend && python3 scripts/auto_verify_cron.py +""" +import sys, os +from datetime import datetime, timedelta + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.database import get_db +from app.models import ActionPlan, KPIDefinition, KPIValue, Objective +from app.utils.notifier import send_wecom_message +from app.api.verify import CONDITION_FUNCS + + +def evaluate_condition(condition, actual, target): + """按条件判断验证结果(兼容旧表达式)""" + if isinstance(condition, str) and ('<' in condition or '>' in condition): + try: + return eval(condition.replace('value', str(actual)), {"__builtins__": {}}) + except Exception: + return False + func = CONDITION_FUNCS.get(condition) + if not func: + return False + return func(actual, target) + + +def run_auto_verify(): + db = next(get_db()) + try: + now = datetime.now() + # 扫描到期pending计划(due_date <= now+1天) + plans = db.query(ActionPlan).filter( + ActionPlan.verify_status == "pending", + ActionPlan.status.in_(["pending", "in_progress"]), + ActionPlan.due_date <= now + timedelta(days=1), + ).all() + + verified_count = 0 + escalated_count = 0 + messages = [] + + for plan in plans: + rule = plan.auto_verify_rule + if not rule or not isinstance(rule, dict): + continue + + kpi_code = rule.get("kpi_code") + if not kpi_code: + continue + + # 查最新KPI值 + kpi = db.query(KPIDefinition).filter( + KPIDefinition.kpi_code == kpi_code + ).order_by(KPIDefinition.id.desc()).first() + if not kpi: + continue + latest = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi.id + ).order_by(KPIValue.calculated_at.desc()).first() + if not latest or latest.actual_value is None: + continue + + actual = latest.actual_value + condition = rule.get("condition", "LESS_THAN") + target = rule.get("target_value") + retry_max = rule.get("retry_max", 3) + + passed = evaluate_condition(condition, actual, target) + + plan.kpi_current_before = rule.get("baseline_value") + plan.kpi_current_after = actual + + if passed: + plan.verify_status = "passed" + plan.verified_at = now + plan.status = "done" + # 阶段3: OKR progress更新 + update_okr_progress(db, plan, +15) + 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" + 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}") + + db.commit() + + # 汇总通知 + if messages: + 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)} + finally: + db.close() + + +def update_okr_progress(db, plan, delta): + """验证通过→更新所属OKR progress""" + okr_id = getattr(plan, "okr_id", None) or getattr(plan, "objective_id", None) + if not okr_id: + return + obj = db.query(Objective).filter(Objective.id == okr_id).first() + if obj: + current = obj.progress or 0 + obj.progress = min(current + delta, 100) + db.add(obj) + + +if __name__ == "__main__": + run_auto_verify() diff --git a/backend/scripts/erp_sync.py b/backend/scripts/erp_sync.py index 22e3e576..bf9110a9 100644 --- a/backend/scripts/erp_sync.py +++ b/backend/scripts/erp_sync.py @@ -273,10 +273,15 @@ def fetch_fallback(kpi: KPIDefinition, parsed: dict, db_session, period: str) -> KPIValue.source_type.in_(["erp", "manual"]), ).order_by(KPIValue.period.desc()).first() - if existing and existing.actual_value is not None: - logger.info(f" [{kpi_code}] Fallback: 沿用最近期 {existing.period}={existing.actual_value}") + if existing and existing.actual_value is not None and existing.data_status == "verified": + logger.info(f" [{kpi_code}] Fallback: 沿用最近期 verified数据 {existing.period}={existing.actual_value}") return existing.actual_value + # Bug3修复: 不沿用estimated旧值(可能失真,如4800万收入),避免污染当月数据 + if existing and existing.actual_value is not None and existing.data_status != "verified": + logger.info(f" [{kpi_code}] Fallback: 最近期 {existing.period} 为estimated({existing.actual_value}),不沿用,返回None") + return None + # 特殊KPI的默认值 DEFAULTS = { "SALES_TOTAL": 800000,