group_fqcd_jr/app/service/risk_analysis_service.py

"""预警研判、回访话术和工单摘要服务。"""

from __future__ import annotations

import json
import logging
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.contracts import RequestContext
from app.core.errors import ForbiddenAgentError, GenericResourceNotFoundError
from app.infrastructure.db import SessionFactory
from app.model.audit import InteractionAudit
from app.model.fund import FundRiskAlert
from app.repository.risk_repository import RiskRepository
from app.service.authorization_service import AuthorizationService
from app.service.risk_query_service import scope_from_context

OUTPUT_TYPES = ("预警研判", "回访话术", "工单摘要")
FORBIDDEN_ACTION_CLAIMS = (
    "我已确认接收", "我已关闭", "我已升级", "我已冻结", "我已放行",
    "已为您确认接收", "已为您关闭", "已为您升级",
)
SYSTEM_PROMPT = "你是南方基金风控助手，只能基于已给证据做研判辅助，不得自动处置交易或预警。"
logger = logging.getLogger(__name__)


class RiskAnalysisService:
    def __init__(
        self,
        session: AsyncSession,
        *,
        repository: RiskRepository | None = None,
        model_service: Any | None = None,
        endpoint_resolver: Any | None = None,
    ) -> None:
        self.session = session
        self.repository = repository
        self.model_service = model_service
        self.endpoint_resolver = endpoint_resolver

    @classmethod
    async def generate_for_context(
        cls,
        context: RequestContext,
        alert_no: str,
        output_type: str,
    ) -> dict[str, Any]:
        async with SessionFactory() as session:
            return await cls(session).generate(context, alert_no, output_type)

    async def generate(
        self,
        context: RequestContext,
        alert_no: str,
        output_type: str,
    ) -> dict[str, Any]:
        if output_type not in OUTPUT_TYPES:
            raise ValueError("不支持的预警分析类型")
        await AuthorizationService.require(context, "risk:alert:read")
        repository = self.repository or RiskRepository(
            self.session,
            scope=scope_from_context(context),
        )
        detail = await repository.get_alert_detail(alert_no)
        if detail is None:
            raise GenericResourceNotFoundError("预警不存在")
        content, source = await self._generate_content(output_type, detail.to_dict())
        await self._save_result(context, alert_no, output_type, content, source)
        return {"type": output_type, "content": content, "source": source}

    async def _generate_content(
        self,
        output_type: str,
        detail: dict[str, Any],
    ) -> tuple[str, str]:
        model_service = self.model_service
        resolver = self.endpoint_resolver
        if model_service is None or resolver is None:
            from app.service.agent.bootstrap import get_model_service
            from app.service.model_gateway import DatabaseModelEndpointResolver

            model_service = model_service or get_model_service()
            resolver = resolver or DatabaseModelEndpointResolver()
        try:
            endpoints = await resolver.resolve(
                agent_type="risk",
                task_type={
                    "预警研判": "risk_analysis",
                    "回访话术": "risk_script",
                    "工单摘要": "risk_summary",
                }[output_type],
            )
            prompt = (
                f"{SYSTEM_PROMPT}\n{_task_instruction(output_type)}\n"
                f"证据如下：{json.dumps(detail, ensure_ascii=False, default=str)}"
            )
            execution = await model_service.generate(endpoints, prompt, max_attempts=2)
            content = execution.text.strip()
            if (
                content
                and len(content) <= 6000
                and not any(claim in content for claim in FORBIDDEN_ACTION_CLAIMS)
            ):
                return content, "模型"
        except Exception:
            logger.exception("%s模型生成失败，已使用模板降级输出", output_type)
        return self._fallback(output_type, detail), "模板降级输出"

    async def _save_result(
        self,
        context: RequestContext,
        alert_no: str,
        output_type: str,
        content: str,
        source: str,
    ) -> None:
        statement = (
            select(FundRiskAlert)
            .where(FundRiskAlert.alert_no == alert_no)
            .with_for_update()
        )
        if context.data_scope != "all":
            if not context.customer_ids:
                raise ForbiddenAgentError("无权访问当前预警")
            statement = statement.where(
                FundRiskAlert.customer_id.in_(
                    tuple(int(customer_id) for customer_id in context.customer_ids)
                )
            )
        alert = await self.session.scalar(statement)
        if alert is None:
            raise GenericResourceNotFoundError("预警不存在")
        now = datetime.now(UTC).replace(tzinfo=None)
        analysis = dict(alert.ai_analysis or {})
        analysis[output_type] = {
            "type": output_type,
            "content": content,
            "source": source,
            "generated_at": now.isoformat(),
        }
        alert.ai_analysis = analysis
        alert.updated_at = now
        self.session.add(InteractionAudit(
            actor_type="agent",
            actor_id=int(context.user_id),
            target_customer_id=alert.customer_id,
            portal="api",
            action_type="risk_ai_analysis_generated",
            detail={"alert_no": alert_no, "output_type": output_type, "source": source},
            created_at=now,
        ))
        await self.session.commit()

    @staticmethod
    def _fallback(output_type: str, detail: dict[str, Any]) -> str:
        alert = detail.get("alert") or {}
        customer = detail.get("customer") or {}
        transaction = detail.get("transaction") or {}
        customer_name = customer.get("name") or "当前客户"
        customer_no = customer.get("customer_no") or "-"
        amount = transaction.get("amount") or "-"
        rules = "，".join(alert.get("rule_codes") or []) or "-"
        evidence = alert.get("evidence_summary") or "暂无证据摘要"
        if output_type == "回访话术":
            return (
                f"回访对象：{customer_name}（{customer_no}）\n"
                "开场说明：您好，我们在例行风控复核中关注到您近期账户交易存在需要核实的情况，"
                "本次沟通仅用于确认交易意愿并完善留痕。\n"
                f"核实问题：请确认近期交易金额 {amount} 元是否由您本人发起；"
                "请说明资金入金和赎回用途；请确认是否本人常用设备操作。\n"
                f"风险提示：本次预警命中 {rules}，核心证据为：{evidence}。\n"
                "留痕要求：请记录客户确认结果、异常解释、回访问答时间，并提交风控专员复核。"
            )
        if output_type == "工单摘要":
            return (
                f"工单标题：{alert.get('alert_type') or '风险预警'}人工复核工单\n"
                f"客户信息：{customer_name}（{customer_no}）\n"
                f"命中规则：{rules}\n"
                f"关键证据：{evidence}\n"
                f"风险等级：{alert.get('risk_level') or '-'}\n"
                "建议动作：分派风控专员核验证据链，补充客户回访记录，"
                "确认后选择继续调查、升级或关闭误报。\n"
                f"处理时限：{alert.get('due_time') or '按风险等级时限处理'}"
            )
        return (
            f"风险结论：{alert.get('alert_type') or '风险预警'}命中 {rules}，建议进入人工复核。\n"
            f"核心证据：{evidence}。\n"
            f"客户特征：{customer_name}（{customer_no}），"
            f"风险等级 {customer.get('risk_level') or '-'}。\n"
            "复核重点：核实资金来源和赎回用途，检查是否本人操作及设备是否异常，"
            "确认交易是否符合客户历史行为。\n"
            "处置建议：由风控专员人工复核并留痕，必要时发起客户回访或升级处理。"
        )


def _task_instruction(output_type: str) -> str:
    return {
        "预警研判": "请生成预警研判，输出风险结论、核心证据、复核重点和处置建议。",
        "回访话术": "请生成客户回访话术，输出开场说明、核实问题、风险提示和留痕提醒。",
        "工单摘要": "请生成工单摘要，输出工单标题、命中规则、关键证据、建议动作和处理时限。",
    }[output_type]