feat(r1-touch): 建议分级+决策类推送+应用前预览 (alert不推送/同title防轰炸/preview对比)

backend/app/api/ai_analysis.py

@@ -8,7 +8,7 @@ from app.deps import get_entity_id
 from app.auth_middleware import require_auth, require_role
 from app.models import KPIDefinition, KPIValue, KPIAlert, StrategicMap, User, ActionPlan, BudgetPlan, AISuggestion
 from app.utils.cache import get as cache_get, set as cache_set
-import json, hashlib, httpx, os
+import json, hashlib, httpx, os, urllib.request
 from datetime import datetime, date
 router = APIRouter(prefix="/api/cma/ai", tags=["AI分析"],
     dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
@@ -25,6 +25,8 @@ def _sug_dict(s: AISuggestion) -> dict:
         "source": s.source,
         "suggestion_type": s.suggestion_type,
         "target_type": s.target_type,
+        "category": s.category or "decision",
+        "pushed": s.pushed or 0,
         "target_id": s.target_id,
         "title": s.title,
         "content": s.content,
@@ -76,6 +78,7 @@ def generate_rule_suggestions(db: Session, entity_id: int,
             source=source,
             suggestion_type=suggestion_type,
             target_type=target_type,
+            category="alert" if target_type == "alert" else "decision",
             target_id=tid,
             title=title,
             content=content,
@@ -170,9 +173,51 @@ def generate_rule_suggestions(db: Session, entity_id: int,
         db.commit()
         for s in created:
             db.refresh(s)
+        # R1触达修复(2026-08-31): 只对新建的决策类建议推送企微（预警类不推防噪音）
+        # 防轰炸: 同 title 建议幂等不重建 + pushed 标记只推一次；存量不推（只推新建）
+        for s in created:
+            if s.category == "decision" and not s.pushed:
+                ok = _push_decision_suggestion(s)
+                if ok:
+                    s.pushed = 1
+                    db.commit()
     return created
 
 
+_TYPE_LABELS = {"kpi_target": "KPI目标", "budget_adjust": "预算调整", "action_plan": "行动方案"}
+
+
+def _push_decision_suggestion(s: AISuggestion) -> bool:
+    """决策类建议推送到企微（8800 relay，与 lead.py 同款已验证）
+
+    仅 decision 类；预警类不进推送流。失败不影响主流程（try/except）。
+    """
+    if getattr(s, "category", "decision") != "decision":
+        return False
+    type_label = _TYPE_LABELS.get(s.suggestion_type, s.suggestion_type)
+    content = (
+        f"## 📌 AI决策建议\n"
+        f"**{s.title}**\n"
+        f"{str(s.content or '')[:120]}\n"
+        f"类型标签: {type_label}\n"
+        f"---\n"
+        f"⏰ {datetime.now().strftime('%Y-%m-%d %H:%M')}"
+    )
+    msg = {"msgtype": "markdown", "markdown": {"content": content}}
+    try:
+        data = json.dumps(msg, ensure_ascii=False).encode("utf-8")
+        req = urllib.request.Request(
+            "http://127.0.0.1:8800/send",
+            data=data,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        urllib.request.urlopen(req, timeout=5)
+        return True
+    except Exception:
+        return False
+
+
 def _unapplied_suggestions(db: Session, entity_id: int, limit: int = 20) -> list:
     items = db.query(AISuggestion).filter(
         AISuggestion.entity_id == entity_id,

backend/app/api/ai_suggestions.py

@@ -27,6 +27,8 @@ def _sug_dict(s: AISuggestion) -> dict:
         "source": s.source,
         "suggestion_type": s.suggestion_type,
         "target_type": s.target_type,
+        "category": s.category or "decision",
+        "pushed": s.pushed or 0,
         "target_id": s.target_id,
         "title": s.title,
         "content": s.content,
@@ -62,6 +64,7 @@ def create_suggestion(
         source=data.get("source", "manual"),
         suggestion_type=suggestion_type,
         target_type=data.get("target_type", "kpi"),
+        category="alert" if data.get("target_type") == "alert" else data.get("category", "decision"),
         target_id=data.get("target_id"),
         title=title,
         content=data.get("content"),
@@ -78,6 +81,7 @@ def create_suggestion(
 def list_suggestions(
     status: Optional[str] = Query(None, description="unapplied/applied/dismissed"),
     suggestion_type: Optional[str] = Query(None),
+    category: Optional[str] = Query(None, description="decision/alert 建议分类过滤"),
     entity_id: int = Depends(get_entity_id),
     db: Session = Depends(get_db),
 ):
@@ -87,6 +91,8 @@ def list_suggestions(
         query = query.filter(AISuggestion.status == status)
     if suggestion_type:
         query = query.filter(AISuggestion.suggestion_type == suggestion_type)
+    if category:
+        query = query.filter(AISuggestion.category == category)
     items = query.order_by(AISuggestion.created_at.desc()).limit(200).all()
     return {"data": [_sug_dict(s) for s in items], "total": len(items)}
 
@@ -275,6 +281,60 @@ _APPLYERS = {
 }
 
 
+@router.get("/{suggestion_id}/preview")
+def preview_suggestion(suggestion_id: int, db: Session = Depends(get_db),
+                       entity_id: int = Depends(get_entity_id)):
+    """应用前预览：将变更什么（当前值 → 新值），建立信任 (R1触达修复 2026-08-31)
+
+    - kpi_target:   {kpi_name, current_target, new_target}
+    - budget_adjust:{kpi_name, period, current_budget, new_budget}
+    - action_plan:  {kpi_name, plan_title, assignee, priority, due_date}
+    """
+    sug = db.query(AISuggestion).filter(AISuggestion.id == suggestion_id).first()
+    if not sug:
+        raise HTTPException(404, "建议不存在")
+    sd = sug.suggestion_data or {}
+    kpi = None
+    kpi_id = sd.get("kpi_id") or sug.target_id
+    if kpi_id:
+        kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
+
+    if sug.suggestion_type == "kpi_target":
+        return {"data": {
+            "type": "kpi_target",
+            "kpi_name": kpi.kpi_name if kpi else "KPI#" + str(kpi_id),
+            "current_target": kpi.target_value if kpi else None,
+            "new_target": sd.get("target_value"),
+        }}
+    if sug.suggestion_type == "budget_adjust":
+        period = sd.get("period") or sug.target_type
+        current_budget = None
+        if kpi and period:
+            bp = db.query(BudgetPlan).filter(
+                BudgetPlan.entity_id == sug.entity_id,
+                BudgetPlan.kpi_id == kpi.id,
+                BudgetPlan.period == period,
+                BudgetPlan.status == "active",
+            ).order_by(BudgetPlan.id.desc()).first()
+            current_budget = bp.budget_value if bp else None
+        return {"data": {
+            "type": "budget_adjust",
+            "kpi_name": kpi.kpi_name if kpi else "KPI#" + str(kpi_id),
+            "period": period,
+            "current_budget": current_budget,
+            "new_budget": sd.get("budget_value"),
+        }}
+    # action_plan
+    return {"data": {
+        "type": "action_plan",
+        "kpi_name": kpi.kpi_name if kpi else "KPI#" + str(kpi_id),
+        "plan_title": sd.get("title") or sug.title,
+        "assignee": sd.get("assignee") or "",
+        "priority": sd.get("priority") or "medium",
+        "due_date": sd.get("due_date") or "",
+    }}
+
+
 @router.post("/{suggestion_id}/apply")
 def apply_suggestion(
     suggestion_id: int,

backend/app/models/__init__.py

@@ -922,6 +922,8 @@ class AISuggestion(Base):
     source = Column(String(30), default="dashboard", comment="来源: dashboard/kpi/budget/manual/rule")
     suggestion_type = Column(String(30), nullable=False, comment="kpi_target/budget_adjust/action_plan")
     target_type = Column(String(30), nullable=False, comment="kpi/budget/action_plan")
+    category = Column(String(20), default="decision", comment="分类: decision决策类 / alert预警类（预警类不推送）")
+    pushed = Column(Integer, default=0, comment="决策类建议是否已推送企微 0/1（防轰炸）")
     target_id = Column(Integer, nullable=True, comment="目标ID (KPI ID/预算KPI ID等)")
     title = Column(String(300), nullable=False, comment="建议标题")
     content = Column(Text, nullable=True, comment="建议内容/理由")

backend/tests/conftest.py

@@ -90,6 +90,17 @@ def setup_db():
     cache_util.delete("ai")
 
 
+@pytest.fixture(autouse=True)
+def _disable_ai_suggestion_push(monkeypatch):
+    """R1触达修复(2026-08-31): 测试库把企微推送替换为 no-op，防测试建议推真实企微群
+
+    生产环境真实推送（8800 relay）；测试只验证推送逻辑（决策类推/预警不推/幂等）不打真实企微。
+    测试类如需断言推送内容，可自行 monkeypatch.setattr 覆盖本 no-op。
+    """
+    from app.api import ai_analysis
+    monkeypatch.setattr(ai_analysis, "_push_decision_suggestion", lambda s: True)
+
+
 @pytest.fixture
 def db() -> Generator[Session, None, None]:
     """提供数据库 session"""

backend/tests/test_ai_suggestions.py

@@ -240,3 +240,269 @@ class TestRuleSuggestions:
         assert resp.status_code == 200
         s = db.query(AISuggestion).filter(AISuggestion.suggestion_type == "budget_adjust").all()
         assert len(s) >= 1
+
+
+class TestSuggestionCategoryPreview:
+    """R1触达修复(2026-08-31)：建议分级(alert/decision) + 列表过滤 + 应用前预览 + 推送开关"""
+
+    def test_create_marks_category(self, client, db):
+        """手动创建：target_type=alert → category=alert；其余 → decision"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+
+        r_alert = _create_suggestion(client, token, kpi.id, target_type="alert",
+                                     suggestion_type="action_plan", title="预警类建议")
+        assert r_alert.json()["data"]["category"] == "alert"
+
+        r_decision = _create_suggestion(client, token, kpi.id, title="决策类建议")
+        assert r_decision.json()["data"]["category"] == "decision"
+
+    def test_category_filter(self, client, db):
+        """列表接口 category 过滤"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+        _create_suggestion(client, token, kpi.id, target_type="alert",
+                           suggestion_type="action_plan", title="预警A")
+        _create_suggestion(client, token, kpi.id, title="决策B")
+
+        lst_alert = client.get("/api/cma/ai/suggestions", params={"category": "alert"},
+                               headers=auth_header(token)).json()
+        assert lst_alert["total"] == 1
+        assert all(x["category"] == "alert" for x in lst_alert["data"])
+
+        lst_decision = client.get("/api/cma/ai/suggestions", params={"category": "decision"},
+                                  headers=auth_header(token)).json()
+        assert lst_decision["total"] == 1
+        assert all(x["category"] == "decision" for x in lst_decision["data"])
+
+    def test_generate_marks_decision(self, client, db):
+        """规则生成：执行率<70%建议（target_type=kpi）→ category=decision"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, target_value=100.0)
+        db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
+        db.commit()
+
+        client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
+        sug = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).first()
+        assert sug is not None
+        assert sug.category == "decision"
+
+    def test_push_disabled_in_test_env(self, client, db):
+        """conftest no-op 推送（monkeypatch）→ 生成决策建议不真推企微，pushed 标记置 1"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, target_value=100.0)
+        db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
+        db.commit()
+
+        client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
+        sug = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).first()
+        assert sug is not None
+        assert sug.pushed == 1
+
+    def test_preview_kpi_target(self, client, db):
+        """preview：kpi_target 返回 当前目标 → 新目标"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, target_value=100.0)
+        r = _create_suggestion(client, token, kpi.id,
+                               suggestion_data={"kpi_id": kpi.id, "target_value": 150.0})
+        sug_id = r.json()["data"]["id"]
+
+        pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token))
+        assert pv.status_code == 200, pv.text
+        data = pv.json()["data"]
+        assert data["type"] == "kpi_target"
+        assert data["kpi_name"] == "测试KPI"
+        assert data["current_target"] == 100.0
+        assert data["new_target"] == 150.0
+
+    def test_preview_budget_adjust(self, client, db):
+        """preview：budget_adjust 返回 当前预算 → 新预算"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+        db.add(BudgetPlan(entity_id=1, kpi_id=kpi.id, period="2026-09", budget_value=8000.0,
+                          budget_year=2026, budget_month=9, status="active"))
+        db.commit()
+        r = _create_suggestion(client, token, kpi.id, suggestion_type="budget_adjust",
+                               title="调预算预览", suggestion_data={"kpi_id": kpi.id, "period": "2026-09",
+                                                                   "budget_value": 9999.0})
+        sug_id = r.json()["data"]["id"]
+
+        pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token))
+        assert pv.status_code == 200, pv.text
+        data = pv.json()["data"]
+        assert data["type"] == "budget_adjust"
+        assert data["period"] == "2026-09"
+        assert data["current_budget"] == 8000.0
+        assert data["new_budget"] == 9999.0
+
+    def test_preview_action_plan(self, client, db):
+        """preview：action_plan 返回计划信息"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+        r = _create_suggestion(client, token, kpi.id, suggestion_type="action_plan",
+                               title="建行动方案预览", suggestion_data={"kpi_id": kpi.id,
+                                                                       "title": "专项改善", "priority": "high",
+                                                                       "due_date": "2026-09-30"})
+        sug_id = r.json()["data"]["id"]
+
+        pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token))
+        assert pv.status_code == 200, pv.text
+        data = pv.json()["data"]
+        assert data["type"] == "action_plan"
+        assert data["plan_title"] == "专项改善"
+        assert data["priority"] == "high"
+        assert data["due_date"] == "2026-09-30"
+
+
+class TestCategoryAndPreview:
+    """R1触达修复(2026-08-31)：建议分级 + 应用前预览"""
+
+    def _generate(self, client, db, kpi_id, actual, target=100.0):
+        """造一条KPI数据并触发 dashboard-analysis 规则生成（避开缓存）"""
+        db.add(KPIValue(kpi_id=kpi_id, period="2026-07", actual_value=actual))
+        db.commit()
+        from app.utils.cache import delete as cache_delete
+        cache_delete("ai", f"dashboard_analysis:ceo:{kpi_id}")
+        resp = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(get_token_for_user(client)))
+        assert resp.status_code == 200
+        return resp.json()
+
+    def test_generate_marks_category(self, client, db, monkeypatch):
+        """生成建议时: target_type=alert → category=alert；其余 → decision"""
+        from app.api import ai_analysis
+        pushed = []
+        ai_analysis._push_decision_suggestion = lambda s: pushed.append(s) or True
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, target_value=100.0)
+        db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
+        db.commit()
+
+        client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
+
+        kpi_sugs = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).all()
+        assert len(kpi_sugs) >= 1
+        for s in kpi_sugs:
+            assert s.category == "decision", f"KPI建议应决策类: {s.title}"
+
+        # 建一条预警 → 规则4生成 alert 类建议
+        from app.models import KPIAlert
+        db.add(KPIAlert(kpi_id=kpi.id, alert_level="yellow", alert_message="测试预警",
+                        alert_type="threshold", status="pending"))
+        db.commit()
+        client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
+        alert_sugs = db.query(AISuggestion).filter(AISuggestion.target_type == "alert").all()
+        assert len(alert_sugs) >= 1
+        for s in alert_sugs:
+            assert s.category == "alert", f"预警建议应alert类: {s.title}"
+
+    def test_alert_not_pushed_decision_pushed(self, client, db, monkeypatch):
+        """推送只发决策类：预警类不推，决策类推且只推一次(pushed=1)"""
+        from app.api import ai_analysis
+        pushed = []
+        ai_analysis._push_decision_suggestion = lambda s: pushed.append(s) or True
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, target_value=100.0)
+        db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
+        db.commit()
+        client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
+
+        kpi_sugs = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).all()
+        assert len(pushed) >= 1
+        assert all(s.category == "decision" for s in pushed)
+        for s in pushed:
+            assert s.pushed == 1
+
+        # 预警类建议不在推送流
+        from app.models import KPIAlert
+        db.add(KPIAlert(kpi_id=kpi.id, alert_level="red", alert_message="测试预警2",
+                        alert_type="threshold", status="pending"))
+        db.commit()
+        before = len(pushed)
+        client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
+        alert_sugs = db.query(AISuggestion).filter(AISuggestion.target_type == "alert").all()
+        assert len(alert_sugs) >= 1
+        assert len(pushed) == before, "预警类不应触发推送"
+
+        # 幂等：重复生成不重推（同title建议不重建）
+        client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
+        assert len(pushed) == before
+
+    def test_list_category_filter(self, client, db):
+        """列表接口 category 过滤"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+        _create_suggestion(client, token, kpi.id, title="决策类A")
+        _create_suggestion(client, token, kpi.id, title="决策类B")
+        _create_suggestion(client, token, kpi.id, title="预警类C", target_type="alert")
+
+        lst = client.get("/api/cma/ai/suggestions?category=decision", headers=auth_header(token)).json()
+        assert lst["total"] == 2
+        assert all(x["category"] == "decision" for x in lst["data"])
+        lst2 = client.get("/api/cma/ai/suggestions?category=alert", headers=auth_header(token)).json()
+        assert lst2["total"] == 1
+        assert lst2["data"][0]["category"] == "alert"
+
+    def test_preview_kpi_target(self, client, db):
+        """preview: kpi_target 返回 current_target → new_target"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, target_value=100.0)
+        r = _create_suggestion(client, token, kpi.id, suggestion_data={"kpi_id": kpi.id, "target_value": 150.0})
+        sug_id = r.json()["data"]["id"]
+
+        pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token)).json()["data"]
+        assert pv["type"] == "kpi_target"
+        assert pv["kpi_name"] == kpi.kpi_name
+        assert pv["current_target"] == 100.0
+        assert pv["new_target"] == 150.0
+
+    def test_preview_budget_adjust(self, client, db):
+        """preview: budget_adjust 返回 current_budget → new_budget"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+        db.add(BudgetPlan(entity_id=1, kpi_id=kpi.id, period="2026-09", budget_value=5000.0,
+                          budget_year=2026, budget_month=9, status="active"))
+        db.commit()
+        r = _create_suggestion(client, token, kpi.id, suggestion_type="budget_adjust",
+                               title="调预算", suggestion_data={"kpi_id": kpi.id, "period": "2026-09", "budget_value": 8888.0})
+        sug_id = r.json()["data"]["id"]
+
+        pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token)).json()["data"]
+        assert pv["type"] == "budget_adjust"
+        assert pv["current_budget"] == 5000.0
+        assert pv["new_budget"] == 8888.0
+        assert pv["period"] == "2026-09"
+
+    def test_preview_action_plan(self, client, db):
+        """preview: action_plan 返回计划参数"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+        r = _create_suggestion(client, token, kpi.id, suggestion_type="action_plan",
+                               title="建行动方案", suggestion_data={"kpi_id": kpi.id, "title": "改善专项",
+                                                                   "assignee": "李四", "priority": "high", "due_date": "2026-10-01"})
+        sug_id = r.json()["data"]["id"]
+
+        pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token)).json()["data"]
+        assert pv["type"] == "action_plan"
+        assert pv["plan_title"] == "改善专项"
+        assert pv["assignee"] == "李四"
+        assert pv["priority"] == "high"
+        assert pv["due_date"] == "2026-10-01"
+
+    def test_preview_not_found(self, client, db):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        r = client.get("/api/cma/ai/suggestions/99999/preview", headers=auth_header(token))
+        assert r.status_code == 404