feat: Auto-Verify闭环 — verify API+定时验证+OKR联动+周报

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}）"
+        ),
+    }

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):

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())

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 = []

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()

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__":

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()

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:
+    if existing and existing.actual_value is not None and existing.data_status == "verified":
-        logger.info(f"  [{kpi_code}] Fallback: 沿用最近期 {existing.period}={existing.actual_value}")
+        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,