lzf_0626
15866d4564
docs/05 §5.1 把"业务写接口"列为必须携带 Idempotency-Key 的接口，风控 6 个 POST （手工扫描、确认接收、进入调查、关闭误报、完成结案、升级处理）此前一个都没带， 重复提交会二次驱动状态机。 复用平台的 api_request_receipt 与 ApiTransactionService，但新增 execute_in：原来的 execute 自己开 SessionFactory() 和 session.begin()，而 RiskActionService._finish 会在内部 commit，套进去就成了"内层提交外层事务"。execute_in 改为在调用方传入的 session 上读写幂等记录，幂等记录因此与业务写入同处一个事务（§5.2）。 scope 用实际路径（含 alert_no）而不是路由模板：§5.1 的幂等范围是 user_id + method + normalized_path + idempotency_key，把路径参数折成模板会让同一个键 在不同预警之间互相回放 —— 那是把两次不同资源的操作当成一次。 测试：单测加幂等透传替身与"缺键即 422"用例；新增 tests/integration/test_risk_idempotency_mysql.py，覆盖同键回放不重复执行、同键不同 正文 409、缺键/非 ASCII 拒绝，以及端到端"重复 POST 只调用一次处置逻辑"。
332 lines
13 KiB
Python
332 lines
13 KiB
Python
"""风控只读查询接口。"""
|
||
|
||
import json
|
||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||
from datetime import datetime, time
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Depends, File, Header, Path, Request, UploadFile
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from starlette.responses import StreamingResponse
|
||
|
||
from app.api.dependencies.auth import build_request_context
|
||
from app.api.dependencies.database import get_session
|
||
from app.api.dependencies.negotiation import accepts_event_stream
|
||
from app.api.dependencies.rate_limit import enforce_rate_limit
|
||
from app.api.schemas.risk import (
|
||
RiskAlertEscalationRequest,
|
||
RiskAlertExclusionRequest,
|
||
RiskAlertPageQuery,
|
||
RiskAlertResolutionRequest,
|
||
RiskDailyReportGenerateRequest,
|
||
RiskDailyReportMailRequest,
|
||
RiskEvidencePageQuery,
|
||
RiskEvidenceSource,
|
||
RiskNotificationPageQuery,
|
||
)
|
||
from app.core.contracts import RequestContext
|
||
from app.core.errors import SseNotAcceptableError
|
||
from app.infrastructure.db import mysql_scan_lock
|
||
from app.service.api_transaction_service import ApiTransactionService
|
||
from app.service.risk_action_service import RiskActionService
|
||
from app.service.risk_daily_report_mail_service import RiskDailyReportMailService
|
||
from app.service.risk_daily_report_service import RiskDailyReportService
|
||
from app.service.risk_evidence_archive_service import RiskEvidenceArchiveService
|
||
from app.service.risk_notification_service import RiskNotificationService
|
||
from app.service.risk_query_service import RiskQueryService
|
||
from app.service.risk_scan_service import RiskScanBusyError, RiskScanService
|
||
|
||
router = APIRouter(
|
||
prefix="/api/v1/risk",
|
||
tags=["risk"],
|
||
dependencies=[Depends(enforce_rate_limit)],
|
||
)
|
||
|
||
|
||
async def _idempotent_write(
|
||
session: AsyncSession,
|
||
context: RequestContext,
|
||
key: str | None,
|
||
scope: str,
|
||
body: Any,
|
||
action: Callable[[AsyncSession], Awaitable[dict[str, Any]]],
|
||
) -> dict[str, Any]:
|
||
"""风控写接口的统一幂等入口（`docs/05` §5.1、§5.2、§13.2）。
|
||
|
||
`scope` 用**实际**路径（含 `alert_no`）：§5.1 的幂等范围是
|
||
`user_id + method + normalized_path + idempotency_key`，把路径参数折成模板会让
|
||
同一个键在不同预警之间互相回放 —— 那是把两次不同资源的操作当成一次。
|
||
|
||
幂等记录与业务写入同事务（见 `ApiTransactionService.execute_in`），重复请求直接
|
||
回放 `response_json`，不会二次驱动状态机。
|
||
"""
|
||
return await ApiTransactionService().execute_in(session, context, scope, key, body, action)
|
||
|
||
|
||
@router.get("/overview")
|
||
async def risk_overview(
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> dict[str, object]:
|
||
data = await RiskQueryService(session).overview(context)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.get("/alerts")
|
||
async def list_risk_alerts(
|
||
query: RiskAlertPageQuery = Depends(), # noqa: B008
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> dict[str, object]:
|
||
data = await RiskQueryService(session).list_alerts(context, query)
|
||
return _list_envelope(data, context)
|
||
|
||
|
||
@router.post("/alerts/scan")
|
||
async def scan_risk_alerts(
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||
) -> dict[str, object]:
|
||
async def run(inner: AsyncSession) -> dict[str, Any]:
|
||
# 手工触发的扫描必须与定时扫描互斥，否则两条路径会同时查不到重复、同时插入。
|
||
# 锁加在**入口层**而不是 `RiskScanService.scan()` 内部：`GET_LOCK` 是连接级的，
|
||
# 而调度器已在它自己的 session 上持锁 —— 被两个入口共用的服务方法若再取同一把锁，
|
||
# 取锁的连接不是持锁的那一个、必然失败，会**把定时扫描自己挡死**。
|
||
async with mysql_scan_lock() as acquired:
|
||
if not acquired:
|
||
raise RiskScanBusyError("规则扫描正在执行，请稍后重试")
|
||
return await RiskScanService(inner).scan(context)
|
||
|
||
data = await _idempotent_write(
|
||
session, context, key, "POST /api/v1/risk/alerts/scan", {}, run
|
||
)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.post("/alerts/{alert_no}/acknowledgements")
|
||
async def acknowledge_risk_alert(
|
||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||
) -> dict[str, object]:
|
||
data = await _idempotent_write(
|
||
session,
|
||
context,
|
||
key,
|
||
f"POST /api/v1/risk/alerts/{alert_no}/acknowledgements",
|
||
{},
|
||
lambda inner: RiskActionService(inner).acknowledge(alert_no, context),
|
||
)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.post("/alerts/{alert_no}/investigations")
|
||
async def investigate_risk_alert(
|
||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||
) -> dict[str, object]:
|
||
data = await _idempotent_write(
|
||
session,
|
||
context,
|
||
key,
|
||
f"POST /api/v1/risk/alerts/{alert_no}/investigations",
|
||
{},
|
||
lambda inner: RiskActionService(inner).investigate(alert_no, context),
|
||
)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.post("/alerts/{alert_no}/exclusions")
|
||
async def exclude_risk_alert(
|
||
payload: RiskAlertExclusionRequest,
|
||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||
) -> dict[str, object]:
|
||
data = await _idempotent_write(
|
||
session,
|
||
context,
|
||
key,
|
||
f"POST /api/v1/risk/alerts/{alert_no}/exclusions",
|
||
{"reason": payload.reason},
|
||
lambda inner: RiskActionService(inner).exclude(alert_no, payload.reason, context),
|
||
)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.post("/alerts/{alert_no}/resolutions")
|
||
async def resolve_risk_alert(
|
||
payload: RiskAlertResolutionRequest,
|
||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||
) -> dict[str, object]:
|
||
data = await _idempotent_write(
|
||
session,
|
||
context,
|
||
key,
|
||
f"POST /api/v1/risk/alerts/{alert_no}/resolutions",
|
||
{"resolution": payload.resolution},
|
||
lambda inner: RiskActionService(inner).resolve(alert_no, payload.resolution, context),
|
||
)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.post("/alerts/{alert_no}/escalations")
|
||
async def escalate_risk_alert(
|
||
payload: RiskAlertEscalationRequest,
|
||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||
) -> dict[str, object]:
|
||
data = await _idempotent_write(
|
||
session,
|
||
context,
|
||
key,
|
||
f"POST /api/v1/risk/alerts/{alert_no}/escalations",
|
||
{"reason": payload.reason},
|
||
lambda inner: RiskActionService(inner).escalate(alert_no, payload.reason, context),
|
||
)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.post("/alerts/{alert_no}/evidence")
|
||
async def archive_risk_evidence(
|
||
evidence_file: UploadFile = File(...), # noqa: B008
|
||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> dict[str, object]:
|
||
try:
|
||
data = await RiskEvidenceArchiveService(session).archive(alert_no, evidence_file, context)
|
||
return _envelope(data, context)
|
||
finally:
|
||
await evidence_file.close()
|
||
|
||
|
||
@router.get("/alerts/{alert_no}")
|
||
async def get_risk_alert(
|
||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> dict[str, object]:
|
||
data = await RiskQueryService(session).get_alert_detail(context, alert_no.strip())
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.get("/evidence/{source}")
|
||
async def list_risk_evidence(
|
||
source: RiskEvidenceSource = Path(), # noqa: B008
|
||
query: RiskEvidencePageQuery = Depends(), # noqa: B008
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> dict[str, object]:
|
||
data = await RiskQueryService(session).list_evidence(context, source, query)
|
||
return _list_envelope(data, context)
|
||
|
||
|
||
@router.get("/notifications")
|
||
async def list_risk_notifications(
|
||
query: RiskNotificationPageQuery = Depends(), # noqa: B008
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> dict[str, object]:
|
||
data = await RiskNotificationService(session).list_notifications(context, query)
|
||
return _list_envelope(data, context)
|
||
|
||
|
||
@router.post("/daily-report")
|
||
async def generate_risk_daily_report(
|
||
payload: RiskDailyReportGenerateRequest,
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> dict[str, object]:
|
||
report_time = (
|
||
datetime.combine(payload.report_date, time.min)
|
||
if payload.report_date is not None
|
||
else None
|
||
)
|
||
data = await RiskDailyReportService(session).generate(context, report_time)
|
||
return _envelope(data, context)
|
||
|
||
|
||
@router.post("/daily-report/stream")
|
||
async def stream_risk_daily_report(
|
||
payload: RiskDailyReportGenerateRequest,
|
||
request: Request,
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> StreamingResponse:
|
||
report_time = (
|
||
datetime.combine(payload.report_date, time.min)
|
||
if payload.report_date is not None
|
||
else None
|
||
)
|
||
service = RiskDailyReportService(session)
|
||
# 鉴权与内容协商都必须在返回 StreamingResponse **之前**完成：`stream()` 是 async
|
||
# generator，函数体到第一次迭代才执行，而那时响应头已经发出去了 —— 403/406 只能
|
||
# 变成"200 + 半截流"（docs/25 P3 #24）。顺序与 §6.4 一致：先鉴权，后 Accept。
|
||
await service.authorize(context)
|
||
if not accepts_event_stream(request.headers.get("Accept")):
|
||
raise SseNotAcceptableError("Accept 必须接受 text/event-stream")
|
||
|
||
async def events() -> AsyncIterator[str]:
|
||
async for event in service.stream(context, report_time):
|
||
event_type = str(event.get("type", "message"))
|
||
payload = json.dumps(event, ensure_ascii=False, default=str)
|
||
yield f"event: {event_type}\ndata: {payload}\n\n"
|
||
|
||
return StreamingResponse(
|
||
events(),
|
||
media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
||
)
|
||
|
||
|
||
@router.post("/daily-report/mail")
|
||
async def send_risk_daily_report_mail(
|
||
payload: RiskDailyReportMailRequest,
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
) -> dict[str, object]:
|
||
data = await RiskDailyReportMailService().send(
|
||
payload.recipients,
|
||
payload.subject,
|
||
payload.content,
|
||
context=context,
|
||
)
|
||
return _envelope(data, context)
|
||
|
||
|
||
def _envelope(data: object, context: RequestContext) -> dict[str, object]:
|
||
return {
|
||
"data": data,
|
||
"meta": {"trace_id": context.trace_id},
|
||
}
|
||
|
||
|
||
def _list_envelope(page: dict[str, Any], context: RequestContext) -> dict[str, object]:
|
||
"""列表资源的信封（docs/05 §3.3）。
|
||
|
||
§3.3 的列表样例是 `data` 为**纯数组**、游标与 `has_more` 放在 `meta` 里，并且明确
|
||
「业务接口不得增加其他顶层字段」。而 `RiskQueryService._page` 返回的是
|
||
`{items, next_cursor, has_more}` —— 整体塞进 `data` 后，游标跑进了**业务数据**里、
|
||
`meta` 只剩 trace_id，两处都不符合契约。
|
||
|
||
这里统一拆包；service 侧不必改（它继续返回那个内部结构，只是不再直接当 `data` 用）。
|
||
"""
|
||
return {
|
||
"data": page.get("items") or [],
|
||
"meta": {
|
||
"trace_id": context.trace_id,
|
||
"next_cursor": page.get("next_cursor"),
|
||
"has_more": bool(page.get("has_more")),
|
||
},
|
||
}
|