harness_zbt/backend/app/api/v1/admin_business.py

from datetime import datetime
from typing import Annotated, Any

from fastapi import APIRouter, Depends, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field

from app.api.responses import success_response
from app.core.errors import AppError
from app.domains.attribution.service import AttributionService
from app.domains.catalog.service import ProductCatalogService
from app.domains.enrollment.service import EnrollmentService
from app.domains.identity.models import AdminUser
from app.domains.identity.service import AdminAuthService

admin_business_bearer = HTTPBearer(auto_error=False)


class RoleUpdateRequest(BaseModel):
    permissions: list[str] = Field(default_factory=list, max_length=50)
    data_scope: str = Field(min_length=2, max_length=32)


class AdminUserCreateRequest(BaseModel):
    username: str = Field(min_length=3, max_length=32)
    password: str = Field(min_length=8, max_length=128)
    display_name: str = Field(min_length=1, max_length=64)
    role_code: str = Field(min_length=2, max_length=64)


class AdminUserUpdateRequest(BaseModel):
    display_name: str = Field(min_length=1, max_length=64)
    status: str = Field(min_length=2, max_length=32)
    role_code: str = Field(min_length=2, max_length=64)


class PasswordResetRequest(BaseModel):
    new_password: str = Field(min_length=8, max_length=128)


class UserStatusRequest(BaseModel):
    status: str = Field(min_length=2, max_length=32)


class ProductStatusRequest(BaseModel):
    status: str = Field(pattern="^(ACTIVE|INACTIVE)$")


class ProductUpdateRequest(BaseModel):
    name: str = Field(min_length=1, max_length=128)
    category: str = Field(pattern="^(MEDICAL|ACCIDENT)$")
    summary: str = Field(min_length=1, max_length=500)


class ProductVersionCreateRequest(BaseModel):
    version_no: str = Field(min_length=1, max_length=32)
    effective_from: datetime
    source_version_id: str | None = Field(default=None, max_length=26)


class ProductPlanRequest(BaseModel):
    id: str | None = Field(default=None, max_length=26)
    code: str = Field(min_length=1, max_length=32)
    name: str = Field(min_length=1, max_length=128)
    summary: str = Field(default="", max_length=500)
    status: str = Field(pattern="^(ACTIVE|INACTIVE)$")
    premium_cents: int = Field(ge=0, le=100_000_000)
    coverage_amount_cents: int = Field(ge=0, le=100_000_000_000)
    min_age: int = Field(ge=0, le=120)
    max_age: int = Field(ge=0, le=120)


class ProductCreateRequest(BaseModel):
    product_code: str = Field(min_length=3, max_length=32)
    name: str = Field(min_length=1, max_length=128)
    category: str = Field(pattern="^(MEDICAL|ACCIDENT)$")
    summary: str = Field(min_length=1, max_length=500)
    version_no: str = Field(default="1.0.0", min_length=1, max_length=32)
    effective_from: datetime
    rule_version: str = Field(default="eligibility-v1", min_length=1, max_length=32)
    rate_version: str = Field(default="rate-v1", min_length=1, max_length=32)
    terms_summary: str = Field(default="", max_length=5000)
    plans: list[ProductPlanRequest] = Field(min_length=1, max_length=12)


class ProductVersionUpdateRequest(BaseModel):
    effective_from: datetime
    effective_to: datetime | None = None
    rule_version: str = Field(min_length=1, max_length=32)
    rate_version: str = Field(min_length=1, max_length=32)
    terms_summary: str = Field(default="", max_length=5000)
    plans: list[ProductPlanRequest] = Field(min_length=1, max_length=12)


def create_admin_business_router(
    service: EnrollmentService,
    auth_service: AdminAuthService,
    catalog_service: ProductCatalogService,
    attribution_service: AttributionService,
) -> APIRouter:
    router = APIRouter(tags=["admin-business"])

    def current_admin(
        credentials: Annotated[
            HTTPAuthorizationCredentials | None,
            Depends(admin_business_bearer),
        ],
    ) -> AdminUser:
        if credentials is None:
            raise AppError("AUTH_REQUIRED", "请先登录后台", 401)
        return auth_service.authenticate_access(credentials.credentials)

    def require_read(user: AdminUser, permission: str) -> None:
        if "*" not in user.permissions and permission not in user.permissions:
            raise AppError("PERMISSION_DENIED", "当前角色无权访问该数据", 403)

    def scoped_orders(user: AdminUser) -> dict[str, Any]:
        result = service.list_orders()
        if user.data_scope == "SELF":
            allowed = attribution_service.order_ids(user.id)
            items = [item for item in result["items"] if item["order_id"] in allowed]
            result = {"items": items, "total": len(items)}
        if user.data_scope != "ALL":
            result = {
                "items": [_mask_order(item) for item in result["items"]],
                "total": result["total"],
            }
        return result

    def scoped_policies(user: AdminUser) -> dict[str, Any]:
        result = service.list_all_policies()
        if user.data_scope == "SELF":
            allowed = attribution_service.order_ids(user.id)
            items = [item for item in result["items"] if item["order_id"] in allowed]
            result = {"items": items, "total": len(items)}
        if user.data_scope != "ALL":
            result = {
                "items": [_mask_policy(item) for item in result["items"]],
                "total": result["total"],
            }
        return result

    @router.get("/admin/dashboard")
    def dashboard(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "dashboard:read")
        orders = scoped_orders(user)
        policies = scoped_policies(user)
        active_policies = [item for item in policies["items"] if item["status"] == "ACTIVE"]
        return success_response(
            request,
            {
                "order_count": orders["total"],
                "policy_count": len(active_policies),
                "premium_cents": sum(int(item["premium_cents"]) for item in active_policies),
            },
        )

    @router.get("/admin/orders")
    def orders(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "order:read")
        return success_response(request, scoped_orders(user))

    @router.get("/admin/orders/{order_id}")
    def order_detail(
        order_id: str,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "order:read")
        item = next(
            (order for order in scoped_orders(user)["items"] if order["order_id"] == order_id),
            None,
        )
        if item is None:
            raise AppError("ORDER_NOT_FOUND", "未找到订单或无权查看", 404)
        return success_response(request, item)

    @router.get("/admin/policies")
    def policies(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "policy:read")
        return success_response(request, scoped_policies(user))

    @router.get("/admin/policies/{policy_id}")
    def policy_detail(
        policy_id: str,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "policy:read")
        item = next(
            (
                policy
                for policy in scoped_policies(user)["items"]
                if policy["policy_id"] == policy_id
            ),
            None,
        )
        if item is None:
            raise AppError("POLICY_NOT_FOUND", "未找到保单或无权查看", 404)
        return success_response(request, item)

    @router.get("/admin/products")
    def products(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:read")
        return success_response(request, catalog_service.list_managed())

    @router.post("/admin/products", status_code=201)
    def create_product(
        payload: ProductCreateRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.create_product(
                product_code=payload.product_code,
                name=payload.name,
                category=payload.category,
                summary=payload.summary,
                version_no=payload.version_no,
                effective_from=payload.effective_from,
                rule_version=payload.rule_version,
                rate_version=payload.rate_version,
                terms_summary=payload.terms_summary,
                plans=[plan.model_dump() for plan in payload.plans],
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.put("/admin/products/{product_id}/status")
    def update_product_status(
        product_id: str,
        payload: ProductStatusRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.update_product_status(
                product_id,
                payload.status,
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.put("/admin/products/{product_id}")
    def update_product(
        product_id: str,
        payload: ProductUpdateRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.update_product(
                product_id,
                name=payload.name,
                category=payload.category,
                summary=payload.summary,
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.post("/admin/products/{product_id}/versions", status_code=201)
    def create_product_version(
        product_id: str,
        payload: ProductVersionCreateRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.create_draft_version(
                product_id,
                version_no=payload.version_no,
                effective_from=payload.effective_from,
                source_version_id=payload.source_version_id,
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.put("/admin/products/{product_id}/versions/{version_id}")
    def update_product_version(
        product_id: str,
        version_id: str,
        payload: ProductVersionUpdateRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.update_draft_version(
                product_id,
                version_id,
                effective_from=payload.effective_from,
                effective_to=payload.effective_to,
                rule_version=payload.rule_version,
                rate_version=payload.rate_version,
                terms_summary=payload.terms_summary,
                plans=[plan.model_dump() for plan in payload.plans],
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.post("/admin/products/{product_id}/versions/{version_id}/publish")
    def publish_product_version(
        product_id: str,
        version_id: str,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.publish_version(
                product_id,
                version_id,
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.post("/admin/products/{product_id}/versions/{version_id}/retire")
    def retire_product_version(
        product_id: str,
        version_id: str,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.retire_version(
                product_id,
                version_id,
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.delete("/admin/products/{product_id}/versions/{version_id}")
    def delete_product_version(
        product_id: str,
        version_id: str,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:write")
        return success_response(
            request,
            catalog_service.delete_draft_version(
                product_id,
                version_id,
                actor_id=user.id,
                actor_name=user.display_name,
            ),
        )

    @router.get("/admin/products/{product_id}/logs")
    def product_change_logs(
        product_id: str,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "product:read")
        return success_response(
            request,
            catalog_service.list_change_logs(product_id),
        )

    @router.get("/admin/attribution/overview")
    def attribution_overview(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "attribution:read")
        owner_id = user.id if user.data_scope == "SELF" else None
        return success_response(
            request,
            attribution_service.performance(owner_id),
        )

    @router.get("/admin/attribution/codes")
    def attribution_codes(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "attribution:read")
        owner_id = user.id if user.data_scope == "SELF" else None
        return success_response(
            request,
            attribution_service.list_codes(owner_id),
        )

    @router.get("/admin/roles")
    def roles(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "role:read")
        return success_response(request, auth_service.list_roles())

    @router.put("/admin/roles/{role_code}")
    def update_role(
        role_code: str,
        payload: RoleUpdateRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "role:write")
        return success_response(
            request,
            auth_service.update_role(
                role_code,
                permissions=payload.permissions,
                data_scope=payload.data_scope,
            ),
        )

    @router.get("/admin/users")
    def managed_users(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "user:read")
        return success_response(request, auth_service.list_managed_users())

    @router.post("/admin/users/admins", status_code=201)
    def create_admin_user(
        payload: AdminUserCreateRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "user:write")
        return success_response(
            request,
            auth_service.create_admin_user(
                username=payload.username,
                password=payload.password,
                display_name=payload.display_name,
                role_code=payload.role_code,
            ),
        )

    @router.put("/admin/users/admins/{user_id}")
    def update_admin_user(
        user_id: str,
        payload: AdminUserUpdateRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "user:write")
        return success_response(
            request,
            auth_service.update_admin_user(
                user,
                user_id,
                display_name=payload.display_name,
                status=payload.status,
                role_code=payload.role_code,
            ),
        )

    @router.post("/admin/users/admins/{user_id}/password")
    def reset_admin_password(
        user_id: str,
        payload: PasswordResetRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "user:write")
        return success_response(
            request,
            auth_service.reset_admin_password(
                user_id,
                new_password=payload.new_password,
            ),
        )

    @router.put("/admin/users/customers/{user_id}/status")
    def update_customer_status(
        user_id: str,
        payload: UserStatusRequest,
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        require_read(user, "user:write")
        return success_response(
            request,
            auth_service.update_customer_status(
                user_id,
                status=payload.status,
            ),
        )

    return router


def _mask_secret(value: str | None) -> str:
    if not value:
        return ""
    if len(value) == 11:
        return f"{value[:3]}****{value[-4:]}"
    if len(value) >= 10:
        return f"{value[:6]}********{value[-4:]}"
    return value


def _mask_person(person: dict[str, Any]) -> dict[str, Any]:
    masked = dict(person)
    if "id_no" in masked:
        masked["id_no"] = _mask_secret(str(masked["id_no"]))
    return masked


def _mask_order(item: dict[str, Any]) -> dict[str, Any]:
    masked = dict(item)
    masked["applicant"] = _mask_person(dict(item.get("applicant", {})))
    masked["insured"] = _mask_person(dict(item.get("insured", {})))
    contact = dict(item.get("contact", {}))
    if "mobile" in contact:
        contact["mobile"] = _mask_secret(str(contact["mobile"]))
    masked["contact"] = contact
    return masked


def _mask_policy(item: dict[str, Any]) -> dict[str, Any]:
    masked = dict(item)
    masked["applicant"] = _mask_person(dict(item.get("applicant", {})))
    masked["insured"] = _mask_person(dict(item.get("insured", {})))
    return masked