group_fqcd_jr/app/service/risk_query_service.py

"""风控只读查询服务，负责授权、范围、分页和对外投影。"""

from __future__ import annotations

from collections.abc import Mapping
from datetime import UTC, date, datetime
from decimal import Decimal
from typing import Any, cast

from sqlalchemy.ext.asyncio import AsyncSession

from app.api.schemas.risk import RiskAlertPageQuery, RiskEvidencePageQuery
from app.core.contracts import RequestContext
from app.core.errors import GenericResourceNotFoundError
from app.core.risk_cursor import decode_offset_cursor, encode_offset_cursor
from app.core.timeutil import from_local
from app.repository.fund_query_repository import CustomerScope, PageRequest
from app.repository.risk_repository import RiskRepository
from app.service.authorization_service import AuthorizationService

ID_FIELDS = {
    "id",
    "customer_id",
    "alert_id",
    "transaction_id",
    "account_id",
    "product_id",
    "related_transaction_id",
    "related_order_id",
    "related_work_order_id",
    "primary_risk_work_order_id",
    "handler_id",
    "submitter_id",
    "advisor_id",
}


class RiskQueryService:
    def __init__(
        self,
        session: AsyncSession,
        *,
        repository: RiskRepository | None = None,
    ) -> None:
        self.session = session
        self.repository = repository

    async def overview(self, context: RequestContext) -> dict[str, Any]:
        await AuthorizationService.require(context, "risk:alert:read")
        repository = self._repository(context)
        result = await repository.overview()
        levels = result["levels"]
        return {
            "total": result["total"],
            "levels": {
                "高风险": levels.get("高", 0),
                "中风险": levels.get("中", 0),
                "低风险": levels.get("低", 0),
            },
            "pending": result["pending"],
            "overdue": result["overdue"],
            "high_priority": [self._record(item) for item in result["high_priority"]],
        }

    async def list_alerts(
        self,
        context: RequestContext,
        query: RiskAlertPageQuery,
    ) -> dict[str, Any]:
        await AuthorizationService.require(context, "risk:alert:read")
        page = await self._repository(context).list_alerts(
            keyword=query.keyword,
            customer_no=query.customer_no,
            product_code=query.product_code,
            product_name=query.product_name,
            risk_level=query.risk_level,
            rule_code=query.rule_code,
            # REST 的时间参数是裸 datetime（不带时区），按**北京时间**解释后再换算成
            # 库内 UTC。Agent 路径本来就带时区（risk_natural_language.py:117），
            # 两条路径口径必须一致，否则同一个筛选条件在界面与对话里查出不同结果。
            start_time=from_local(query.start_time) if query.start_time else None,
            end_time=from_local(query.end_time) if query.end_time else None,
            page=PageRequest(limit=query.limit, offset=decode_offset_cursor(query.cursor)),
        )
        return self._page(page)

    async def get_alert_detail(
        self,
        context: RequestContext,
        alert_no: str,
    ) -> dict[str, Any]:
        await AuthorizationService.require(context, "risk:alert:read")
        record = await self._repository(context).get_alert_detail(alert_no)
        if record is None:
            raise GenericResourceNotFoundError("预警不存在")
        return self._record(record)

    async def list_evidence(
        self,
        context: RequestContext,
        source: str,
        query: RiskEvidencePageQuery,
    ) -> dict[str, Any]:
        await AuthorizationService.require(context, "risk:alert:read")
        page_request = PageRequest(
            limit=query.limit,
            offset=decode_offset_cursor(query.cursor),
        )
        repository = self._repository(context)
        if source == "customers":
            page = await repository.list_customers(
                keyword=query.keyword,
                behavior_level=query.behavior_level,
                page=page_request,
            )
        elif source == "products":
            page = await repository.list_products(keyword=query.keyword, page=page_request)
        elif source == "transactions":
            page = await repository.list_transactions(
                keyword=query.keyword,
                start_time=query.start_time,
                end_time=query.end_time,
                page=page_request,
            )
        elif source == "capital_flows":
            page = await repository.list_capital_flows(
                keyword=query.keyword,
                start_time=query.start_time,
                end_time=query.end_time,
                page=page_request,
            )
        elif source == "holdings":
            page = await repository.list_holdings(keyword=query.keyword, page=page_request)
        elif source == "login_records":
            page = await repository.list_login_records(
                keyword=query.keyword,
                start_time=query.start_time,
                end_time=query.end_time,
                page=page_request,
            )
        elif source == "alerts":
            alert_page = await repository.list_alerts(keyword=query.keyword, page=page_request)
            return self._page(alert_page)
        elif source == "notifications":
            page = await repository.list_notifications(
                keyword=query.keyword,
                send_status=query.send_status,
                start_time=query.start_time,
                end_time=query.end_time,
                page=page_request,
            )
        else:
            raise GenericResourceNotFoundError("证据类型不存在")
        return self._page(page)

    def _repository(self, context: RequestContext) -> RiskRepository:
        if self.repository is not None:
            return self.repository
        return RiskRepository(self.session, scope=scope_from_context(context))

    @classmethod
    def _page(cls, page: Any) -> dict[str, Any]:
        return {
            "items": [cls._record(item) for item in page.items],
            "next_cursor": (
                encode_offset_cursor(page.next_offset)
                if page.next_offset is not None
                else None
            ),
            "has_more": page.has_more,
        }

    @classmethod
    def _record(cls, record: Any) -> dict[str, Any]:
        values = record.to_dict() if hasattr(record, "to_dict") else dict(record)
        return cast(dict[str, Any], cls._plain(values))

    @classmethod
    def _plain(cls, value: Any, *, field: str | None = None) -> Any:
        if isinstance(value, Mapping):
            return {key: cls._plain(item, field=str(key)) for key, item in value.items()}
        if isinstance(value, (list, tuple)):
            return [cls._plain(item, field=field) for item in value]
        if isinstance(value, Decimal):
            return str(value)
        if isinstance(value, datetime):
            if value.tzinfo is None:
                value = value.replace(tzinfo=UTC)
            return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
        if isinstance(value, date):
            return value.isoformat()
        if field in ID_FIELDS and isinstance(value, int) and not isinstance(value, bool):
            return str(value)
        return value


def scope_from_context(context: RequestContext) -> CustomerScope:
    if context.data_scope == "all":
        return CustomerScope.unrestricted()
    if not context.customer_ids:
        return CustomerScope.denied()
    return CustomerScope.for_customers(int(customer_id) for customer_id in context.customer_ids)