group_xinghuo_jinrong/app/api/analyst.py

"""数据分析 Agent 路由（D-01~D-12 / N-03/07/08）。"""
from __future__ import annotations

from fastapi import APIRouter, Depends, HTTPException

from app.api.analyst_auth_adapter import (
    AnalystAuthContext,
    AnalystAuthError,
    analyst_auth_from_deps,
    assert_analyst_query_access,
)
from app.api.deps import AuthContext, get_platform_auth_context
from app.model.analyst_schemas import AnalystResponse, AssetCreateRequest, ChatRequest
from app.service.analyst_agent import AnalystAgent
from app.utils.trace import current_trace

router = APIRouter(prefix="/api/analyst", tags=["analyst"])

_agent: AnalystAgent | None = None


def get_agent() -> AnalystAgent:
    global _agent
    if _agent is None:
        _agent = AnalystAgent()
    return _agent


def _analyst_ctx(auth: AuthContext = Depends(get_platform_auth_context)) -> AnalystAuthContext:
    return analyst_auth_from_deps(auth, trace_id=current_trace() or "")


@router.post("/chat", response_model=AnalystResponse)
def chat(
    req: ChatRequest,
    auth: AnalystAuthContext = Depends(_analyst_ctx),
    agent: AnalystAgent = Depends(get_agent),
) -> AnalystResponse:
    return agent.run(req.question, auth, req.session_id, req.trace_id)


@router.get("/dashboard")
def dashboard(
    auth: AnalystAuthContext = Depends(_analyst_ctx),
    agent: AnalystAgent = Depends(get_agent),
):
    """智能看数板（D-12）后端：按角色返回卡片。"""
    try:
        domain = assert_analyst_query_access(auth)
    except AnalystAuthError as exc:
        raise HTTPException(status_code=403, detail=exc.message) from exc
    role = next(
        (r for r in auth.roles if r in ("analyst", "advisor", "risk_officer", "ops", "customer")),
        "analyst",
    )
    cards = {
        "analyst": ["客户总数", "总持仓规模", "今日交易笔数与金额", "待处理预警数", "口径字典资产数"],
        "advisor": ["名下客户数", "名下资产规模", "盈亏分布", "风险等级分布"],
        "risk_officer": ["待处理预警数", "预警按类型分布", "近7天新增趋势"],
        "customer": ["我的持仓规模", "近30日交易笔数", "风险等级", "盈亏概览"],
        "ops": ["近30天申购金额", "近30天赎回金额", "各产品类型规模TOP"],
    }.get(role, [])
    metrics: dict = {}
    try:
        if role == "analyst":
            res = agent.repo.execute_readonly(
                "SELECT (SELECT COUNT(*) FROM core_customer) AS customers, "
                "(SELECT COALESCE(SUM(market_value),0) FROM core_holding) AS holdings, "
                "(SELECT COUNT(*) FROM jinrong_agent.risk_alert WHERE status='pending_review') AS pending, "
                "(SELECT COUNT(*) FROM jinrong_agent.analytics_metric_dict WHERE status='published') AS dict_count"
            )
            if res["rows"]:
                metrics = dict(zip(res["columns"], res["rows"][0]))
        elif role == "customer" and auth.customer_id:
            cid = auth.customer_id
            res = agent.repo.execute_readonly(
                f"SELECT (SELECT COALESCE(SUM(market_value),0) FROM core_holding WHERE customer_id='{cid}') AS holdings, "
                f"(SELECT COUNT(*) FROM core_trade WHERE customer_id='{cid}' AND trade_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)) AS trades_30d"
            )
            if res["rows"]:
                metrics = dict(zip(res["columns"], res["rows"][0]))
        elif role == "advisor" and auth.subject_id:
            scope = agent.repo.resolve_advisor_scope(auth.subject_id)
            if scope:
                ids = ",".join(f"'{x}'" for x in scope)
                res = agent.repo.execute_readonly(
                    f"SELECT COUNT(DISTINCT customer_id) AS clients, "
                    f"COALESCE(SUM(market_value),0) AS aum FROM core_holding WHERE customer_id IN ({ids})"
                )
                if res["rows"]:
                    metrics = dict(zip(res["columns"], res["rows"][0]))
            else:
                metrics = {"clients": 0, "aum": 0}
        elif role == "risk_officer":
            res = agent.repo.execute_readonly(
                "SELECT COUNT(*) AS pending FROM jinrong_agent.risk_alert WHERE status='pending_review'"
            )
            if res["rows"]:
                metrics = dict(zip(res["columns"], res["rows"][0]))
    except Exception:  # noqa: BLE001
        pass
    return {"role": role, "domain": domain, "cards": cards, "metrics": metrics}


@router.post("/assets")
def create_asset(
    req: AssetCreateRequest,
    auth: AnalystAuthContext = Depends(_analyst_ctx),
    agent: AnalystAgent = Depends(get_agent),
):
    """沉淀资产（D-11）：仅分析师可写。"""
    if "analyst" not in auth.roles:
        raise HTTPException(status_code=403, detail="仅分析师可沉淀资产")
    if req.kind not in ("dict", "few_shot", "template"):
        raise HTTPException(status_code=400, detail="kind 必须是 dict/few_shot/template")
    asset_id = agent.repo.insert_asset(req.kind, req.payload, auth.subject_id)
    return {"ok": True, "id": asset_id, "kind": req.kind}


@router.get("/ops/metrics")
def ops_metrics(
    auth: AnalystAuthContext = Depends(_analyst_ctx),
    agent: AnalystAgent = Depends(get_agent),
):
    """运营指标（N-08，P1 最小版）。"""
    if "analyst" not in auth.roles:
        raise HTTPException(status_code=403, detail="仅分析师可查看运营指标")
    res = agent.repo.execute_readonly(
        "SELECT COUNT(*) AS total, "
        "SUM(exec_status='blocked') AS blocked "
        "FROM jinrong_agent.analytics_query_log"
    )
    return res["rows"][0] if res["rows"] else {"total": 0, "blocked": 0}