ai-ticket-platform/backend/apps/modelhub/api.py

from datetime import date, timedelta
from math import ceil

from django.conf import settings
from django.db.models import Count, Q
from django.utils import timezone
from ninja import Router
from ninja.errors import HttpError

from apps.accounts.api import require_login, require_permission
from apps.audit.models import OperationAuditLog
from apps.modelhub.crypto import mask_secret
from apps.modelhub.governance import (
    ModelGovernanceError,
    create_deployment_config,
    create_provider_config,
    deployment_impact,
    provider_impact,
    remove_deployment,
    remove_provider,
    restore_deployment,
    restore_provider,
    update_deployment_config,
    update_provider_config,
)
from apps.modelhub.models import ModelDeployment, ModelInvocation, ModelProvider
from apps.modelhub.schemas import (
    DeletionImpactOut,
    LifecycleResultOut,
    ModelDeploymentIn,
    ModelDeploymentOptionOut,
    ModelDeploymentOut,
    ModelDeploymentPageOut,
    ModelDeploymentUpdateIn,
    ModelInvocationDiagnosticsOut,
    ModelInvocationOut,
    ModelInvocationPageOut,
    ModelProviderIn,
    ModelProviderOut,
    ModelProviderPageOut,
    ModelProviderUpdateIn,
    ModelTestIn,
)
from apps.modelhub.services import create_model_test, create_structured_output_test
from apps.modelhub.structured_output import (
    CHAT_CAPABILITIES,
    StructuredOutputError,
    validate_mode_for_capability,
)
from apps.modelhub.tasks import run_model_test

router = Router(tags=["modelhub"])


def governance_http_error(error: ModelGovernanceError) -> HttpError:
    return HttpError(error.status, f"{error.code}: {error}")


def structured_http_error(error: StructuredOutputError) -> HttpError:
    return HttpError(400, f"{error.code}: {error}")


def require_superuser(request):
    user = require_login(request)
    if not user.is_superuser:
        raise HttpError(403, "superuser_required: 仅超级管理员可执行生命周期操作")
    return user


def require_any_permission(request, *codes: str):
    user = require_login(request)
    if user.is_superuser or any(user.has_perm(f"system.{code}") for code in codes):
        return user
    raise HttpError(403, "Permission denied")


def _user_label(user) -> str | None:
    if not user:
        return None
    return user.get_full_name() or user.username


def _provider_status(provider: ModelProvider) -> str:
    if provider.archived_at:
        return "archived"
    if not provider.enabled:
        return "disabled"
    if not provider.api_key_encrypted:
        return "credential_missing"
    return "ready"


def _deployment_status(deployment: ModelDeployment) -> str:
    if deployment.archived_at:
        return "archived"
    if deployment.provider.archived_at:
        return "provider_archived"
    if not deployment.enabled:
        return "disabled"
    provider_status = _provider_status(deployment.provider)
    if provider_status != "ready":
        return f"provider_{provider_status}"
    if (
        deployment.capability in CHAT_CAPABILITIES
        and deployment.structured_output_test_status != "passed"
    ):
        return "unverified"
    return "ready"


def serialize_provider(provider: ModelProvider) -> dict:
    return {
        "id": provider.id,
        "name": provider.name,
        "base_url": provider.base_url,
        "enabled": provider.enabled,
        "default_timeout_seconds": provider.default_timeout_seconds,
        "credential_set": bool(provider.api_key_encrypted),
        "credential_masked": mask_secret(provider.api_key_encrypted),
        "image_egress_approved": provider.image_egress_approved,
        "image_egress_approved_by": _user_label(provider.image_egress_approved_by),
        "image_egress_approved_at": provider.image_egress_approved_at,
        "credential_updated_at": provider.credential_updated_at,
        "credential_updated_by": _user_label(provider.credential_updated_by),
        "archived_at": provider.archived_at,
        "archived_by": _user_label(provider.archived_by),
        "deployment_count": (
            provider.deployment_count
            if hasattr(provider, "deployment_count")
            else provider.deployments.count()
        ),
        "active_deployment_count": (
            provider.active_deployment_count
            if hasattr(provider, "active_deployment_count")
            else provider.deployments.filter(archived_at__isnull=True).count()
        ),
        "effective_status": _provider_status(provider),
        "created_at": provider.created_at,
        "updated_at": provider.updated_at,
    }


def _reference_summary(deployment: ModelDeployment) -> dict[str, int]:
    return {
        "agent_version_count": (
            deployment.agent_version_count
            if hasattr(deployment, "agent_version_count")
            else deployment.agent_versions.count()
        ),
        "step_binding_count": (
            deployment.step_binding_count
            if hasattr(deployment, "step_binding_count")
            else deployment.agent_step_bindings.count()
        ),
        "run_step_count": (
            deployment.run_step_count
            if hasattr(deployment, "run_step_count")
            else deployment.agent_run_steps.count()
        ),
        "invocation_count": (
            deployment.invocation_count
            if hasattr(deployment, "invocation_count")
            else deployment.invocations.count()
        ),
    }


def serialize_deployment(deployment: ModelDeployment, metrics: dict | None = None) -> dict:
    if metrics is None:
        from apps.agents.structured_metrics import build_structured_output_metrics

        metrics = build_structured_output_metrics(deployment_id=deployment.id)
    return {
        "id": deployment.id,
        "provider_id": deployment.provider_id,
        "provider_name": deployment.provider.name,
        "display_name": deployment.display_name,
        "model_name": deployment.model_name,
        "capability": deployment.capability,
        "default_temperature": deployment.default_temperature,
        "max_output_tokens": deployment.max_output_tokens,
        "structured_output_mode": deployment.structured_output_mode,
        "structured_output_test_status": deployment.structured_output_test_status,
        "structured_output_verified_at": deployment.structured_output_verified_at,
        "structured_output_test_message": deployment.structured_output_test_message,
        "structured_output_metrics": metrics,
        "enabled": deployment.enabled,
        "effective_status": _deployment_status(deployment),
        "provider_effective_status": _provider_status(deployment.provider),
        "reference_summary": _reference_summary(deployment),
        "archived_at": deployment.archived_at,
        "archived_by": _user_label(deployment.archived_by),
        "archived_via_provider": deployment.archived_via_provider,
    }


def _diagnostics_expires_at(invocation: ModelInvocation):
    return invocation.created_at + timedelta(
        days=settings.MODEL_INVOCATION_DIAGNOSTIC_RETENTION_DAYS
    )


def serialize_invocation(
    invocation: ModelInvocation,
    *,
    include_diagnostics: bool = False,
) -> dict:
    snapshot = invocation.model_snapshot or {}
    deployment = invocation.deployment
    expires_at = _diagnostics_expires_at(invocation)
    available = (
        invocation.diagnostics_scrubbed_at is None and expires_at > timezone.now()
    )
    summary = {
        key: snapshot.get(key)
        for key in (
            "provider_id",
            "provider_name",
            "deployment_id",
            "model_name",
            "capability",
            "structured_output",
        )
        if key in snapshot
    }
    return {
        "public_id": invocation.public_id,
        "deployment_id": invocation.deployment_id,
        "deployment_display_name": deployment.display_name if deployment else None,
        "provider_name": (
            deployment.provider.name if deployment else snapshot.get("provider_name")
        ),
        "invocation_type": invocation.invocation_type,
        "status": invocation.status,
        "model_snapshot": snapshot if include_diagnostics else {},
        "model_snapshot_summary": summary,
        "output_text": invocation.output_text if include_diagnostics and available else "",
        "input_tokens": invocation.input_tokens,
        "output_tokens": invocation.output_tokens,
        "duration_ms": invocation.duration_ms,
        "retry_count": invocation.retry_count,
        "error_code": invocation.error_code,
        "error_message": invocation.error_message[:500],
        "created_at": invocation.created_at,
        "started_at": invocation.started_at,
        "finished_at": invocation.finished_at,
        "diagnostics_scrubbed_at": invocation.diagnostics_scrubbed_at,
        "diagnostics_expires_at": expires_at,
        "diagnostics_available": available,
    }


def audit(user, action: str, target_type: str, target_id: str, snapshot: dict) -> None:
    OperationAuditLog.objects.create(
        user=user if getattr(user, "is_authenticated", False) else None,
        module="modelhub",
        action=action,
        target_type=target_type,
        target_id=target_id,
        request_snapshot=snapshot,
        result="success",
    )


def _page(queryset, page: int, page_size: int):
    page = max(page, 1)
    page_size = min(max(page_size, 1), 100)
    total = queryset.count()
    start = (page - 1) * page_size
    return list(queryset[start : start + page_size]), {
        "page": page,
        "page_size": page_size,
        "total": total,
        "total_pages": ceil(total / page_size) if total else 0,
    }


@router.get("/providers", response=ModelProviderPageOut)
def list_providers(
    request,
    q: str = "",
    enabled: str = "all",
    archived: str = "active",
    page: int = 1,
    page_size: int = 20,
):
    require_permission(request, "model.manage")
    queryset = ModelProvider.objects.select_related(
        "image_egress_approved_by", "credential_updated_by", "archived_by"
    ).annotate(
        deployment_count=Count("deployments", distinct=True),
        active_deployment_count=Count(
            "deployments",
            filter=Q(deployments__archived_at__isnull=True),
            distinct=True,
        ),
    )
    if q:
        queryset = queryset.filter(Q(name__icontains=q) | Q(base_url__icontains=q))
    if enabled in {"true", "false"}:
        queryset = queryset.filter(enabled=enabled == "true")
    if archived == "active":
        queryset = queryset.filter(archived_at__isnull=True)
    elif archived == "archived":
        queryset = queryset.filter(archived_at__isnull=False)
    rows, pagination = _page(queryset, page, page_size)
    return {"items": [serialize_provider(item) for item in rows], **pagination}


@router.post("/providers", response=ModelProviderOut)
def create_provider(request, payload: ModelProviderIn):
    require_permission(request, "model.manage")
    try:
        provider = create_provider_config(payload.dict(), request.user)
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    audit(request.user, "create", "model_provider", str(provider.id), {"name": provider.name})
    return serialize_provider(provider)


@router.patch("/providers/{provider_id}", response=ModelProviderOut)
def update_provider(request, provider_id: int, payload: ModelProviderUpdateIn):
    require_permission(request, "model.manage")
    provider = ModelProvider.objects.select_related(
        "image_egress_approved_by", "credential_updated_by", "archived_by"
    ).get(id=provider_id)
    try:
        provider, snapshot = update_provider_config(
            provider, payload.dict(exclude_unset=True), request.user
        )
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    audit(request.user, "update", "model_provider", str(provider.id), snapshot)
    return serialize_provider(provider)


@router.get("/providers/{provider_id}/deletion-impact", response=DeletionImpactOut)
def get_provider_deletion_impact(request, provider_id: int):
    require_superuser(request)
    return provider_impact(ModelProvider.objects.get(id=provider_id))


@router.delete("/providers/{provider_id}", response=LifecycleResultOut)
def delete_provider(request, provider_id: int):
    user = require_superuser(request)
    try:
        result, impact = remove_provider(provider_id, user)
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    audit(user, result, "model_provider", str(provider_id), {"impact": impact})
    return {"result": result, "impact": impact}


@router.post("/providers/{provider_id}/restore", response=ModelProviderOut)
def restore_provider_api(request, provider_id: int):
    user = require_superuser(request)
    try:
        provider = restore_provider(provider_id)
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    audit(user, "restore", "model_provider", str(provider_id), {"enabled": False})
    return serialize_provider(provider)


@router.get("/deployments", response=ModelDeploymentPageOut)
def list_deployments(
    request,
    provider_id: int | None = None,
    capability: str = "",
    effective_status: str = "",
    structured_test_status: str = "",
    archived: str = "active",
    q: str = "",
    page: int = 1,
    page_size: int = 20,
):
    require_permission(request, "model.manage")
    queryset = ModelDeployment.objects.select_related(
        "provider", "archived_by"
    ).annotate(
        agent_version_count=Count("agent_versions", distinct=True),
        step_binding_count=Count("agent_step_bindings", distinct=True),
        run_step_count=Count("agent_run_steps", distinct=True),
        invocation_count=Count("invocations", distinct=True),
    )
    if provider_id:
        queryset = queryset.filter(provider_id=provider_id)
    if capability:
        queryset = queryset.filter(capability=capability)
    if structured_test_status:
        queryset = queryset.filter(
            structured_output_test_status=structured_test_status
        )
    if q:
        queryset = queryset.filter(
            Q(display_name__icontains=q)
            | Q(model_name__icontains=q)
            | Q(provider__name__icontains=q)
        )
    if archived == "active":
        queryset = queryset.filter(archived_at__isnull=True)
    elif archived == "archived":
        queryset = queryset.filter(archived_at__isnull=False)
    if effective_status:
        if effective_status == "archived":
            queryset = queryset.filter(archived_at__isnull=False)
        elif effective_status == "provider_archived":
            queryset = queryset.filter(
                archived_at__isnull=True,
                provider__archived_at__isnull=False,
            )
        elif effective_status == "disabled":
            queryset = queryset.filter(archived_at__isnull=True, enabled=False)
        elif effective_status == "provider_disabled":
            queryset = queryset.filter(
                archived_at__isnull=True,
                enabled=True,
                provider__archived_at__isnull=True,
                provider__enabled=False,
            )
        elif effective_status == "unverified":
            queryset = queryset.filter(
                archived_at__isnull=True,
                provider__archived_at__isnull=True,
                enabled=True,
                provider__enabled=True,
                capability__in=CHAT_CAPABILITIES,
            ).exclude(structured_output_test_status="passed")
        elif effective_status == "ready":
            queryset = queryset.filter(
                archived_at__isnull=True,
                provider__archived_at__isnull=True,
                enabled=True,
                provider__enabled=True,
            ).exclude(provider__api_key_encrypted="").filter(
                Q(capability__in={"embedding", "rerank"})
                | Q(structured_output_test_status="passed")
            )
    rows, pagination = _page(queryset, page, page_size)
    from apps.agents.structured_metrics import build_deployment_structured_output_metrics

    metrics = build_deployment_structured_output_metrics([item.id for item in rows])
    return {
        "items": [serialize_deployment(item, metrics[item.id]) for item in rows],
        **pagination,
    }


@router.get("/deployment-options", response=list[ModelDeploymentOptionOut])
def list_deployment_options(request):
    require_any_permission(request, "agent.manage", "model.manage")
    queryset = ModelDeployment.objects.select_related("provider").filter(
        archived_at__isnull=True,
        provider__archived_at__isnull=True,
        enabled=True,
        provider__enabled=True,
    ).exclude(provider__api_key_encrypted="")
    items = [item for item in queryset if _deployment_status(item) == "ready"]
    return [
        {
            "id": item.id,
            "provider_id": item.provider_id,
            "provider_name": item.provider.name,
            "display_name": item.display_name,
            "model_name": item.model_name,
            "capability": item.capability,
            "default_temperature": item.default_temperature,
            "max_output_tokens": item.max_output_tokens,
            "structured_output_mode": item.structured_output_mode,
            "structured_output_test_status": item.structured_output_test_status,
            "structured_output_verified_at": item.structured_output_verified_at,
            "structured_output_test_message": item.structured_output_test_message,
            "enabled": item.enabled,
        }
        for item in items
    ]


@router.post("/deployments", response=ModelDeploymentOut)
def create_deployment(request, payload: ModelDeploymentIn):
    require_permission(request, "model.manage")
    try:
        deployment = create_deployment_config(payload.dict())
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    deployment = ModelDeployment.objects.select_related("provider").get(id=deployment.id)
    audit(
        request.user,
        "create",
        "model_deployment",
        str(deployment.id),
        {"model_name": deployment.model_name},
    )
    return serialize_deployment(deployment)


@router.patch("/deployments/{deployment_id}", response=ModelDeploymentOut)
def update_deployment(request, deployment_id: int, payload: ModelDeploymentUpdateIn):
    require_permission(request, "model.manage")
    deployment = ModelDeployment.objects.select_related("provider", "archived_by").get(
        id=deployment_id
    )
    try:
        deployment, snapshot = update_deployment_config(
            deployment, payload.dict(exclude_unset=True)
        )
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    audit(request.user, "update", "model_deployment", str(deployment.id), snapshot)
    return serialize_deployment(deployment)


def _ensure_testable(deployment: ModelDeployment):
    if deployment.archived_at or deployment.provider.archived_at:
        raise HttpError(
            400, "model_config_archived: 归档模型不允许编辑、测试或启用"
        )


@router.post("/deployments/{deployment_id}/test", response=ModelInvocationOut)
def test_deployment(request, deployment_id: int, payload: ModelTestIn):
    require_permission(request, "model.manage")
    deployment = ModelDeployment.objects.select_related("provider").get(id=deployment_id)
    _ensure_testable(deployment)
    if deployment.capability not in CHAT_CAPABILITIES:
        raise HttpError(400, "Embedding 或重排模型请通过对应能力入口验证")
    invocation = create_model_test(deployment, payload.prompt, request.user)
    run_model_test.delay(invocation.id)
    invocation.refresh_from_db()
    audit(
        request.user,
        "test",
        "model_deployment",
        str(deployment.id),
        {"invocation_id": str(invocation.public_id)},
    )
    return serialize_invocation(invocation, include_diagnostics=True)


@router.post(
    "/deployments/{deployment_id}/structured-output-test",
    response=ModelInvocationOut,
)
def test_structured_output(request, deployment_id: int):
    require_permission(request, "model.manage")
    deployment = ModelDeployment.objects.select_related("provider").get(id=deployment_id)
    _ensure_testable(deployment)
    if deployment.capability not in CHAT_CAPABILITIES:
        raise HttpError(400, "该模型能力不适用结构化输出测试")
    try:
        validate_mode_for_capability(
            deployment.capability, deployment.structured_output_mode
        )
    except StructuredOutputError as error:
        raise structured_http_error(error) from error
    deployment.structured_output_test_status = "untested"
    deployment.structured_output_verified_at = None
    deployment.structured_output_test_message = "结构化输出测试执行中"
    deployment.save(
        update_fields=[
            "structured_output_test_status",
            "structured_output_verified_at",
            "structured_output_test_message",
            "updated_at",
        ]
    )
    invocation = create_structured_output_test(deployment, request.user)
    run_model_test.delay(invocation.id)
    invocation.refresh_from_db()
    audit(
        request.user,
        "structured_output_test",
        "model_deployment",
        str(deployment.id),
        {
            "invocation_id": str(invocation.public_id),
            "mode": deployment.structured_output_mode,
        },
    )
    return serialize_invocation(invocation, include_diagnostics=True)


@router.get("/deployments/{deployment_id}/deletion-impact", response=DeletionImpactOut)
def get_deployment_deletion_impact(request, deployment_id: int):
    require_superuser(request)
    return deployment_impact(ModelDeployment.objects.get(id=deployment_id))


@router.delete("/deployments/{deployment_id}", response=LifecycleResultOut)
def delete_deployment(request, deployment_id: int):
    user = require_superuser(request)
    try:
        result, impact = remove_deployment(deployment_id, user)
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    audit(user, result, "model_deployment", str(deployment_id), {"impact": impact})
    return {"result": result, "impact": impact}


@router.post("/deployments/{deployment_id}/restore", response=ModelDeploymentOut)
def restore_deployment_api(request, deployment_id: int):
    user = require_superuser(request)
    try:
        deployment = restore_deployment(deployment_id)
    except ModelGovernanceError as error:
        raise governance_http_error(error) from error
    audit(user, "restore", "model_deployment", str(deployment_id), {"enabled": False})
    return serialize_deployment(deployment)


@router.get("/invocations", response=ModelInvocationPageOut)
def list_invocations(
    request,
    deployment_id: int | None = None,
    provider_id: int | None = None,
    invocation_type: str = "",
    status: str = "",
    date_from: date | None = None,
    date_to: date | None = None,
    error_code: str = "",
    page: int = 1,
    page_size: int = 20,
):
    require_permission(request, "model.manage")
    queryset = ModelInvocation.objects.select_related("deployment__provider")
    if deployment_id:
        queryset = queryset.filter(deployment_id=deployment_id)
    if provider_id:
        queryset = queryset.filter(deployment__provider_id=provider_id)
    if invocation_type:
        queryset = queryset.filter(invocation_type=invocation_type)
    if status:
        queryset = queryset.filter(status=status)
    if date_from:
        queryset = queryset.filter(created_at__date__gte=date_from)
    if date_to:
        queryset = queryset.filter(created_at__date__lte=date_to)
    if error_code:
        queryset = queryset.filter(error_code=error_code)
    rows, pagination = _page(queryset, page, page_size)
    return {"items": [serialize_invocation(item) for item in rows], **pagination}


@router.get("/invocations/{public_id}", response=ModelInvocationOut)
def get_invocation(request, public_id: str):
    require_permission(request, "model.manage")
    invocation = ModelInvocation.objects.select_related("deployment__provider").get(
        public_id=public_id
    )
    return serialize_invocation(invocation)


@router.get(
    "/invocations/{public_id}/diagnostics",
    response=ModelInvocationDiagnosticsOut,
)
def get_invocation_diagnostics(request, public_id: str):
    require_permission(request, "model.invocation.view")
    invocation = ModelInvocation.objects.get(public_id=public_id)
    expires_at = _diagnostics_expires_at(invocation)
    available = (
        invocation.diagnostics_scrubbed_at is None and expires_at > timezone.now()
    )
    if not available:
        raise HttpError(
            410, "diagnostics_expired: 模型调用诊断内容已超过保留期"
        )
    return {
        "public_id": invocation.public_id,
        "input_snapshot": invocation.input_snapshot,
        "output_text": invocation.output_text,
        "diagnostics_scrubbed_at": invocation.diagnostics_scrubbed_at,
        "diagnostics_expires_at": expires_at,
        "diagnostics_available": True,
    }


@router.get("/model-tests/{public_id}", response=ModelInvocationOut)
def get_model_test(request, public_id: str):
    require_permission(request, "model.manage")
    invocation = ModelInvocation.objects.select_related("deployment__provider").get(
        public_id=public_id,
        invocation_type__in={"model_test", "structured_output_test"},
    )
    return serialize_invocation(invocation, include_diagnostics=True)