from collections.abc import Awaitable, Callable from typing import Any from fastapi import APIRouter, Depends, Header, Path, Query, Response from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from app.api.dependencies.auth import build_request_context from app.api.dependencies.database import get_session from app.api.dependencies.rate_limit import enforce_rate_limit from app.api.schemas.admin import ( EmptyPayload, EndpointPayload, HandoverAssignPayload, HandoverCancelPayload, HandoverClosePayload, HandoverResolvePayload, HandoverStatus, IntentPayload, ItemPayload, NegativePayload, PromptPayload, ReleasePayload, ReplyPayload, ReviewPayload, RoutingPayload, ) from app.api.views.envelope import envelope as _envelope from app.core.contracts import RequestContext from app.core.profile_governance_contracts import ProfileDriftReviewRequest from app.service.admin_service import AdminService from app.service.api_transaction_service import ApiTransactionService from app.service.customer_profile_candidate_service import CustomerProfileCandidateService from app.service.customer_service_handover_action_service import ( CustomerServiceHandoverActionService, ) from app.service.customer_service_handover_admin_service import CustomerServiceHandoverAdminService from app.service.profile_governance_service import ProfileGovernanceService router = APIRouter( prefix="/api/v1/admin", tags=["platform-admin"], dependencies=[Depends(enforce_rate_limit)] ) @router.get("/advisor/profile-tags") async def list_profile_tags( customer_id: int = Query(gt=0), context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, object]: return await ProfileGovernanceService().tags(customer_id, context) @router.get("/advisor/profile-drift-reviews") async def list_profile_drift_reviews( context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, object]: return await ProfileGovernanceService().pending_reviews(context) @router.post("/advisor/profile-drift-reviews/{review_id}/reviews") async def review_profile_drift( payload: ProfileDriftReviewRequest, review_id: int = Path(gt=0), context: RequestContext = Depends(build_request_context), # noqa: B008 key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: return await ProfileGovernanceService().review(review_id, payload, context, key) def register_resource( resource: str, schema: type[BaseModel], id_name: str, *, scoped: bool = False, update: bool = True, detail: bool = True, ) -> None: prefix = f"/config-releases/{{release_id}}/{resource}" if scoped else f"/{resource}" async def create( payload: BaseModel, response: Response, release_id: int | None = None, context: RequestContext = Depends(build_request_context), # noqa: B008 key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, Any]: result = await AdminService().mutate( resource, context, payload.model_dump(mode="json"), key, None, release_id=release_id ) response.headers["ETag"] = f'"{result["meta"]["etag"]}"' return result create.__annotations__["payload"] = schema router.add_api_route( prefix, create, methods=["POST"], status_code=201, operation_id=f"create_{resource}" ) async def list_rows( release_id: int | None = None, limit: int = Query(default=20, ge=1, le=100), cursor: str | None = Query(default=None), context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, Any]: """列表查询(文档 §3.8 统一游标)。游标校验在 Service 的权限闸门之后进行。""" return await AdminService().query( resource, context, release_id=release_id, limit=limit, cursor=cursor ) router.add_api_route(prefix, list_rows, methods=["GET"], operation_id=f"list_{resource}") async def get( response: Response, row_id: int = Path(alias=id_name, gt=0), context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, Any]: result = await AdminService().query(resource, context, row_id=row_id) response.headers["ETag"] = f'"{result["meta"]["etag"]}"' return result if detail: router.add_api_route( f"{prefix}/{{{id_name}}}", get, methods=["GET"], operation_id=f"get_{resource}" ) async def put( payload: BaseModel, response: Response, row_id: int = Path(alias=id_name, gt=0), release_id: int | None = None, context: RequestContext = Depends(build_request_context), # noqa: B008 key: str | None = Header(default=None, alias="Idempotency-Key"), if_match: str | None = Header(default=None, alias="If-Match"), ) -> dict[str, Any]: result = await AdminService().mutate( resource, context, payload.model_dump(mode="json"), key, if_match, row_id=row_id, release_id=release_id, ) response.headers["ETag"] = f'"{result["meta"]["etag"]}"' return result put.__annotations__["payload"] = schema if update: router.add_api_route( f"{prefix}/{{{id_name}}}", put, methods=["PUT"], operation_id=f"update_{resource}" ) def register_transition(resource: str, id_name: str, action: str) -> None: async def transition( payload: BaseModel, response: Response, row_id: int = Path(alias=id_name, gt=0), context: RequestContext = Depends(build_request_context), # noqa: B008 key: str | None = Header(default=None, alias="Idempotency-Key"), if_match: str | None = Header(default=None, alias="If-Match"), ) -> dict[str, Any]: result = await AdminService().mutate( resource, context, payload.model_dump(mode="json"), key, if_match, row_id=row_id, action=action, ) response.headers["ETag"] = f'"{result["meta"]["etag"]}"' return result transition.__annotations__["payload"] = ReviewPayload if action == "reviews" else EmptyPayload router.add_api_route( f"/{resource}/{{{id_name}}}/{action}", transition, methods=["POST"], status_code=201 if action == "rollbacks" else 200, operation_id=f"{action}_{resource}", ) register_resource("config-releases", ReleasePayload, "release_id", update=False) # ⚠️ `platform-config-items` 与 `model-routing-rules` 必须是 `detail=True`: # 它们的更新端点(PUT)**硬性要求 `If-Match`**,而校验用的是**该行内容的 digest** # (`admin_service.mutate`:`if_match is None or if_match != digest(existing) → 409`)。 # 此前 `detail=False` 意味着**没有任何端点能返回这个 digest** —— 列表的 `meta` 只有 # trace_id、也没有详情端点,于是**首次编辑必然 409**:乐观并发成了死锁, # 编辑功能实际不可用(2026-09-13 前端等价测试发现)。 register_resource("platform-config-items", ItemPayload, "item_id", scoped=True, detail=True) register_resource("model-endpoints", EndpointPayload, "endpoint_id") register_resource("model-routing-rules", RoutingPayload, "rule_id", scoped=True, detail=True) register_resource("prompt-templates", PromptPayload, "prompt_id", update=False) register_resource("agent-intent-configs", IntentPayload, "config_id") register_resource("reply-templates", ReplyPayload, "template_id", detail=False) register_resource("negative-word-rules", NegativePayload, "rule_id", detail=False) for action in ("validations", "reviews", "activations", "rollbacks"): register_transition("config-releases", "release_id", action) for action in ("reviews", "activations", "disablements"): register_transition("model-endpoints", "endpoint_id", action) # 意图配置沿用同一套"审核 → 生效 → 归档"流转;归档落 `archived` # (该表 CHECK 约束只允许 draft/approved/active/archived,没有 disabled)。 for action in ("reviews", "activations", "archivals"): register_transition("agent-intent-configs", "config_id", action) @router.get("/audit-records") async def audit_records( limit: int = Query(default=20, ge=1, le=100), cursor: str | None = Query(default=None), context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, Any]: """审计查询(文档 §9.6 支持游标过滤)。游标非法时返回 `400 INVALID_CURSOR`。""" return await AdminService().query("audit-records", context, limit=limit, cursor=cursor) @router.get("/customer-service/handover-tickets") async def list_customer_service_handover_tickets( limit: int = Query(default=20, ge=1, le=100), cursor: str | None = Query(default=None), status: HandoverStatus | None = Query(default=None), # noqa: B008 context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, Any]: """只读查看客服转人工队列;不暴露原始会话。可按状态筛。 处置动作在下面五个端点里,需要 `handover:write`(本端点只需 `handover:read`)。 """ return await CustomerServiceHandoverAdminService().list_tickets( context, limit=limit, cursor=cursor, status=status ) @router.get("/customer-service/handover-tickets/{ticket_no}") async def get_customer_service_handover_ticket( ticket_no: str, context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, Any]: """只读查看单个工单的脱敏转接摘要与流转状态。""" return await CustomerServiceHandoverAdminService().get_ticket(ticket_no, context) # ---- 转人工工单处置(分配 / 接单 / 解决 / 关闭 / 取消)---- # # 状态机与权限口径见 `docs/02` §7.2 与 `app/service/customer_service_handover_action_service.py`: # pending -> assigned -> processing -> resolved -> closed,未解决可 cancelled。 # 五个端点都要求 `handover:write` + admin 角色,并且都要求 `Idempotency-Key` # (重复提交直接回放上次结果,不会二次推进状态机)。 @router.post("/customer-service/handover-tickets/{ticket_no}/assignments") async def assign_customer_service_handover_ticket( payload: HandoverAssignPayload, ticket_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, Any]: """把待处理工单分配给一个坐席(`pending -> assigned`)。""" return await _handover_action( session, context, key, f"POST /api/v1/admin/customer-service/handover-tickets/{ticket_no}/assignments", payload.model_dump(), lambda inner: CustomerServiceHandoverActionService(inner).assign( ticket_no, payload.assignee_id, context ), ) @router.post("/customer-service/handover-tickets/{ticket_no}/acceptances") async def accept_customer_service_handover_ticket( ticket_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, Any]: """坐席接单(`pending` 自助接管 / `assigned -> processing`)。""" return await _handover_action( session, context, key, f"POST /api/v1/admin/customer-service/handover-tickets/{ticket_no}/acceptances", {}, lambda inner: CustomerServiceHandoverActionService(inner).accept(ticket_no, context), ) @router.post("/customer-service/handover-tickets/{ticket_no}/resolutions") async def resolve_customer_service_handover_ticket( payload: HandoverResolvePayload, ticket_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, Any]: """给出解决结论(`processing -> resolved`)。""" return await _handover_action( session, context, key, f"POST /api/v1/admin/customer-service/handover-tickets/{ticket_no}/resolutions", payload.model_dump(), lambda inner: CustomerServiceHandoverActionService(inner).resolve( ticket_no, payload.resolution, context ), ) @router.post("/customer-service/handover-tickets/{ticket_no}/closures") async def close_customer_service_handover_ticket( ticket_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"), payload: HandoverClosePayload | None = None, 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, Any]: """归档关闭(`resolved -> closed`);可选补一条说明。""" note = payload.note if payload is not None else "" return await _handover_action( session, context, key, f"POST /api/v1/admin/customer-service/handover-tickets/{ticket_no}/closures", {"note": note}, lambda inner: CustomerServiceHandoverActionService(inner).close( ticket_no, note, context ), ) @router.post("/customer-service/handover-tickets/{ticket_no}/cancellations") async def cancel_customer_service_handover_ticket( payload: HandoverCancelPayload, ticket_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, Any]: """取消未解决的工单(`pending|assigned|processing -> cancelled`)。""" return await _handover_action( session, context, key, f"POST /api/v1/admin/customer-service/handover-tickets/{ticket_no}/cancellations", payload.model_dump(), lambda inner: CustomerServiceHandoverActionService(inner).cancel( ticket_no, payload.reason, context ), ) async def _handover_action( session: AsyncSession, context: RequestContext, key: str | None, scope: str, body: dict[str, Any], action: Callable[[AsyncSession], Awaitable[dict[str, Any]]], ) -> dict[str, Any]: """工单处置的统一外壳:幂等 + 统一信封。 幂等记录与业务写入**同事务**(`ApiTransactionService.execute_in`), 重复请求直接回放 `response_json`,不会二次驱动状态机。 幂等范围含路径参数 —— 同一个键用在两张不同工单上会被当成两次不同操作 (与风控处置同一口径:把路径参数折成模板会让不同资源互相回放)。 """ data = await ApiTransactionService().execute_in( session, context, scope, key, body, action ) return _envelope(data, context) @router.get("/customer-profile-candidates") async def list_customer_profile_candidates( limit: int = Query(default=20, ge=1, le=100), context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, Any]: """管理员查看待确认或待审核的画像候选。""" return await CustomerProfileCandidateService().list_for_admin(context, limit=limit) @router.post("/customer-profile-candidates/{candidate_id}/reviews", status_code=200) async def review_customer_profile_candidate( candidate_id: int, payload: ReviewPayload, context: RequestContext = Depends(build_request_context), # noqa: B008 ) -> dict[str, Any]: """管理员批准或驳回候选;批准会处理同键旧正式记忆。""" return await CustomerProfileCandidateService().review_by_admin( candidate_id, payload.decision, context, comment=payload.comment )