wecom-fullstack
467472311d
task dama-entity-isolation-sweep-20260923 (sections 2-2 4/5, merged by projectBot 2026-09-23 18:2x)
4) operation_logs write side: every audit write carries the account (no DDL, no column added)
- new app/utils/oplog.py::log_op() single wrapper: sets detail.entity_id; scope=global for
by-design non-entity writes (platform dict / role-level); scope=batch + OPLOG_BATCH_NO_ENTITY
sentinel for cross-account batches; no context -> OPLOG_NO_ENTITY sentinel (never guess)
- 16 remaining construction points converted (data/alert_rules/bi_reports/kpi_causality/
bot_bridge x2/budget_generate/driver_budget/kpis _log x3/templates _log x2/erp_sync);
pre-4 state had 20 sites without entity_id, now 0 outside the wrapper
- read side untouched (knowledge_service still user_id fallback); history not backfilled
5) docs/entity-dimension-data-dictionary.md: 8-table three-state registry
(kpi_templates/driver_factor_templates/scenario_suggestions/bot_bridge_config = non-entity;
kpi_subject_map = inherited via kpi_id; analysis_results/subjects = account dimension,
legacy rows still NULL -> other task; operation_logs = non-entity column + detail.entity_id)
Evidence: RED 2 failed/7 passed (reverted call sites) -> GREEN full 1428 passed; 9 new tests
including two read-back rows; app md5 restore verified identical; no cross-account writes
LOCAL commit only - push is gated (ahead contains the P1 entity-dim batch commit fb4609b)
429 lines
16 KiB
Python
429 lines
16 KiB
Python
"""AI决策建议 — 一键应用到KPI/预算/行动方案 (路线图R1 2026-08-30)
|
||
|
||
北极星④决策闭环：AI建议 → 点击应用 → 写库变更 → OperationLog留痕 → 前端可查已应用/未应用。
|
||
应用动作复用现有 kpis/budget/action_plans 数据模型，不新建业务接口。
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, Request
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import func
|
||
from typing import Optional
|
||
from datetime import datetime
|
||
|
||
from app.database import get_db
|
||
from app.deps import get_entity_id, resolve_entity_for_request
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.models import AISuggestion, KPIDefinition, KPIValue, OperationLog, BudgetPlan, ActionPlan
|
||
|
||
router = APIRouter(prefix="/api/cma/ai/suggestions", tags=["AI建议"],
|
||
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
||
)
|
||
|
||
|
||
def _sug_dict(s: AISuggestion) -> dict:
|
||
return {
|
||
"id": s.id,
|
||
"entity_id": s.entity_id,
|
||
"user_id": s.user_id,
|
||
"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,
|
||
"suggestion_data": s.suggestion_data or {},
|
||
"status": s.status,
|
||
"applied_by": s.applied_by,
|
||
"applied_at": s.applied_at.isoformat() if s.applied_at else None,
|
||
"apply_detail": s.apply_detail or [],
|
||
"created_at": s.created_at.isoformat() if s.created_at else None,
|
||
}
|
||
|
||
|
||
@router.post("")
|
||
def create_suggestion(
|
||
request: Request,
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(require_auth),
|
||
):
|
||
"""创建AI建议（前端AI分析/手动保存建议）"""
|
||
suggestion_type = data.get("suggestion_type") or data.get("type")
|
||
title = (data.get("title") or "").strip()
|
||
if not suggestion_type:
|
||
raise HTTPException(400, "缺少 suggestion_type (kpi_target/budget_adjust/action_plan)")
|
||
if not title:
|
||
raise HTTPException(400, "缺少 title")
|
||
|
||
entity_id = resolve_entity_for_request(request, data.get("entity_id") or 1)
|
||
|
||
sug = AISuggestion(
|
||
entity_id=entity_id,
|
||
user_id=getattr(current_user, "id", None),
|
||
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"),
|
||
suggestion_data=data.get("suggestion_data") or {},
|
||
status="unapplied",
|
||
)
|
||
db.add(sug)
|
||
db.commit()
|
||
db.refresh(sug)
|
||
return {"success": True, "message": "建议已保存", "data": _sug_dict(sug)}
|
||
|
||
|
||
@router.get("")
|
||
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),
|
||
):
|
||
"""建议列表（前端建议中心/详情页查看已应用/未应用状态）"""
|
||
query = db.query(AISuggestion).filter(AISuggestion.entity_id == entity_id)
|
||
if status:
|
||
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)}
|
||
|
||
|
||
@router.get("/{suggestion_id}")
|
||
def get_suggestion(suggestion_id: int, db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id)):
|
||
"""建议详情"""
|
||
s = db.query(AISuggestion).filter(
|
||
AISuggestion.id == suggestion_id,
|
||
AISuggestion.entity_id == entity_id, # 2026-09-18 派生单: 建议须归属会话账套
|
||
).first()
|
||
if not s:
|
||
raise HTTPException(404, "建议不存在")
|
||
return {"data": _sug_dict(s)}
|
||
|
||
|
||
def _apply_kpi_target(db: Session, sug: AISuggestion, params: dict, current_user) -> dict:
|
||
"""改KPI目标"""
|
||
target_value = params.get("target_value")
|
||
if target_value is None:
|
||
raise HTTPException(400, "应用kpi_target需要 target_value")
|
||
kpi_id = params.get("kpi_id") or sug.target_id
|
||
if not kpi_id:
|
||
raise HTTPException(400, "缺少 kpi_id")
|
||
kpi = db.query(KPIDefinition).filter(
|
||
KPIDefinition.id == kpi_id,
|
||
KPIDefinition.entity_id == sug.entity_id, # 2026-09-18 派生单: 目标KPI须归属建议账套
|
||
).first()
|
||
if not kpi:
|
||
raise HTTPException(404, f"KPI {kpi_id} 不存在")
|
||
before = kpi.target_value
|
||
kpi.target_value = float(target_value)
|
||
db.flush()
|
||
detail_item = {
|
||
"target_type": "kpi",
|
||
"target_id": kpi.id,
|
||
"target_name": kpi.kpi_name,
|
||
"action": "update_target_value",
|
||
"before": before,
|
||
"after": float(target_value),
|
||
}
|
||
db.add(OperationLog(
|
||
user_id=getattr(current_user, "id", None),
|
||
action="ai_suggestion_apply",
|
||
target_type="kpi",
|
||
target_id=kpi.id,
|
||
detail={
|
||
"entity_id": sug.entity_id, # 2026-09-23 ④：审计带账套（=建议账套，端点已校验=会话账套）
|
||
"suggestion_id": sug.id,
|
||
"suggestion_title": sug.title,
|
||
"apply_action": "kpi_target",
|
||
"before": before,
|
||
"after": float(target_value),
|
||
},
|
||
))
|
||
return detail_item
|
||
|
||
|
||
def _apply_budget_adjust(db: Session, sug: AISuggestion, params: dict, current_user) -> dict:
|
||
"""调预算（BudgetPlan upsert，按 kpi_id+period）"""
|
||
period = params.get("period")
|
||
budget_value = params.get("budget_value")
|
||
if not period or budget_value is None:
|
||
raise HTTPException(400, "应用budget_adjust需要 period + budget_value")
|
||
kpi_id = params.get("kpi_id") or sug.target_id
|
||
if not kpi_id:
|
||
raise HTTPException(400, "缺少 kpi_id")
|
||
kpi = db.query(KPIDefinition).filter(
|
||
KPIDefinition.id == kpi_id,
|
||
KPIDefinition.entity_id == sug.entity_id, # 2026-09-18 派生单: 目标KPI须归属建议账套
|
||
).first()
|
||
if not kpi:
|
||
raise HTTPException(404, f"KPI {kpi_id} 不存在")
|
||
|
||
# 解析期间 2026-09 → year=2026 month=9
|
||
try:
|
||
parts = period.split("-")
|
||
year = int(parts[0])
|
||
month = int(parts[1])
|
||
except Exception:
|
||
raise HTTPException(400, f"period格式错误: {period} (需要 YYYY-MM)")
|
||
|
||
plan = db.query(BudgetPlan).filter(
|
||
BudgetPlan.entity_id == sug.entity_id,
|
||
BudgetPlan.kpi_id == kpi_id,
|
||
BudgetPlan.period == period,
|
||
BudgetPlan.budget_year == year,
|
||
BudgetPlan.budget_month == month,
|
||
BudgetPlan.status == "active",
|
||
).first()
|
||
|
||
before = None
|
||
if plan:
|
||
before = plan.budget_value
|
||
plan.budget_value = float(budget_value)
|
||
else:
|
||
plan = BudgetPlan(
|
||
entity_id=sug.entity_id,
|
||
kpi_id=kpi_id,
|
||
period=period,
|
||
budget_value=float(budget_value),
|
||
budget_year=year,
|
||
budget_month=month,
|
||
version="v1.0",
|
||
status="active",
|
||
source_type="ai_suggestion",
|
||
calc_logic=f"AI建议应用 #{sug.id}: {sug.title}",
|
||
created_by=getattr(current_user, "name", "") or "",
|
||
)
|
||
db.add(plan)
|
||
db.flush()
|
||
detail_item = {
|
||
"target_type": "budget",
|
||
"target_id": plan.id,
|
||
"target_name": f"{kpi.kpi_name}[{period}]",
|
||
"action": "update_budget" if before is not None else "create_budget",
|
||
"before": before,
|
||
"after": float(budget_value),
|
||
}
|
||
db.add(OperationLog(
|
||
user_id=getattr(current_user, "id", None),
|
||
action="ai_suggestion_apply",
|
||
target_type="budget",
|
||
target_id=plan.id,
|
||
detail={
|
||
"entity_id": sug.entity_id, # 2026-09-23 ④：审计带账套（=建议账套）
|
||
"suggestion_id": sug.id,
|
||
"suggestion_title": sug.title,
|
||
"apply_action": "budget_adjust",
|
||
"kpi_id": kpi_id,
|
||
"period": period,
|
||
"before": before,
|
||
"after": float(budget_value),
|
||
},
|
||
))
|
||
return detail_item
|
||
|
||
|
||
def _apply_action_plan(db: Session, sug: AISuggestion, params: dict, current_user) -> dict:
|
||
"""建行动方案"""
|
||
title = (params.get("title") or "").strip() or sug.title
|
||
kpi_id = params.get("kpi_id") or sug.target_id
|
||
if not kpi_id:
|
||
raise HTTPException(400, "缺少 kpi_id")
|
||
kpi = db.query(KPIDefinition).filter(
|
||
KPIDefinition.id == kpi_id,
|
||
KPIDefinition.entity_id == sug.entity_id, # 2026-09-18 派生单 2030b: 目标KPI须归属建议账套
|
||
).first()
|
||
if not kpi:
|
||
raise HTTPException(404, f"KPI {kpi_id} 不存在")
|
||
|
||
due_date = None
|
||
if params.get("due_date"):
|
||
try:
|
||
due_date = datetime.strptime(str(params["due_date"])[:10], "%Y-%m-%d")
|
||
except Exception:
|
||
due_date = None
|
||
|
||
plan = ActionPlan(
|
||
entity_id=sug.entity_id, # 2026-09-18 派生单: 写建议归属账套（=会话账套，端点已校验）
|
||
kpi_id=kpi_id,
|
||
title=title,
|
||
description=params.get("description") or sug.content or f"由AI建议 #{sug.id} 生成: {sug.title}",
|
||
assignee=params.get("assignee") or "",
|
||
priority=params.get("priority") or "medium",
|
||
due_date=due_date,
|
||
status="pending",
|
||
progress=0,
|
||
created_by=getattr(current_user, "name", "") or "ai_suggestion",
|
||
)
|
||
db.add(plan)
|
||
db.flush()
|
||
detail_item = {
|
||
"target_type": "action_plan",
|
||
"target_id": plan.id,
|
||
"target_name": title,
|
||
"action": "create_action_plan",
|
||
"before": None,
|
||
"after": plan.id,
|
||
}
|
||
db.add(OperationLog(
|
||
user_id=getattr(current_user, "id", None),
|
||
action="ai_suggestion_apply",
|
||
target_type="action_plan",
|
||
target_id=plan.id,
|
||
detail={
|
||
"entity_id": sug.entity_id, # 2026-09-23 ④：审计带账套（=建议账套）
|
||
"suggestion_id": sug.id,
|
||
"suggestion_title": sug.title,
|
||
"apply_action": "action_plan",
|
||
"kpi_id": kpi_id,
|
||
"plan_title": title,
|
||
},
|
||
))
|
||
return detail_item
|
||
|
||
|
||
_APPLYERS = {
|
||
"kpi_target": _apply_kpi_target,
|
||
"budget_adjust": _apply_budget_adjust,
|
||
"action_plan": _apply_action_plan,
|
||
}
|
||
|
||
|
||
@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,
|
||
AISuggestion.entity_id == entity_id, # 2026-09-18 派生单: 预览须归属会话账套（原依赖未用于过滤）
|
||
).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:
|
||
# 2026-09-18 派生单 2030b: 目标KPI须归属建议账套；对端KPI降级为 KPI#<id>（不回显名/目标值）
|
||
kpi = db.query(KPIDefinition).filter(
|
||
KPIDefinition.id == kpi_id,
|
||
KPIDefinition.entity_id == sug.entity_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,
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
current_user=Depends(require_auth),
|
||
):
|
||
"""应用建议：改KPI目标 / 调预算 / 建行动方案（写库+操作日志留痕）
|
||
|
||
Body 示例:
|
||
{"action": "kpi_target", "target_value": 2000000}
|
||
{"action": "budget_adjust", "period": "2026-09", "budget_value": 100000}
|
||
{"action": "action_plan", "title": "...", "assignee": "...", "priority": "high", "due_date": "2026-09-30"}
|
||
"""
|
||
sug = db.query(AISuggestion).filter(
|
||
AISuggestion.id == suggestion_id,
|
||
AISuggestion.entity_id == entity_id, # 2026-09-18 派生单: 建议须归属会话账套
|
||
).first()
|
||
if not sug:
|
||
raise HTTPException(404, "建议不存在")
|
||
if sug.status == "applied":
|
||
raise HTTPException(400, "该建议已应用，不能重复应用")
|
||
if sug.status == "dismissed":
|
||
raise HTTPException(400, "该建议已忽略，如需应用请重新创建")
|
||
|
||
action = data.get("action") or sug.suggestion_type
|
||
applier = _APPLYERS.get(action)
|
||
if not applier:
|
||
raise HTTPException(400, f"不支持的应用动作: {action} (支持 kpi_target/budget_adjust/action_plan)")
|
||
|
||
# 应用参数 = 请求体参数 覆盖 建议默认参数
|
||
params = dict(sug.suggestion_data or {})
|
||
params.update({k: v for k, v in data.items() if k != "action" and v is not None})
|
||
|
||
detail_item = applier(db, sug, params, current_user)
|
||
sug.status = "applied"
|
||
sug.applied_by = getattr(current_user, "name", "") or ""
|
||
sug.applied_user_id = getattr(current_user, "id", None)
|
||
sug.applied_at = datetime.now()
|
||
sug.apply_detail = [detail_item]
|
||
db.commit()
|
||
db.refresh(sug)
|
||
return {
|
||
"success": True,
|
||
"message": "建议已应用并留痕",
|
||
"data": _sug_dict(sug),
|
||
}
|
||
|
||
|
||
@router.post("/{suggestion_id}/dismiss")
|
||
def dismiss_suggestion(
|
||
suggestion_id: int,
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
current_user=Depends(require_auth),
|
||
):
|
||
"""忽略建议"""
|
||
sug = db.query(AISuggestion).filter(
|
||
AISuggestion.id == suggestion_id,
|
||
AISuggestion.entity_id == entity_id, # 2026-09-18 派生单: 建议须归属会话账套
|
||
).first()
|
||
if not sug:
|
||
raise HTTPException(404, "建议不存在")
|
||
sug.status = "dismissed"
|
||
db.commit()
|
||
return {"success": True, "message": "建议已忽略"}
|