"""风控只读查询服务，负责授权、范围、分页和对外投影。""" 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)