"""风控只读查询接口。"""

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")),
        },
    }