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

from __future__ import annotations

from typing import Any

from django.core.exceptions import ValidationError
from django.core.validators import URLValidator
from django.db import IntegrityError, transaction
from django.utils import timezone

from apps.modelhub.crypto import CredentialEncryptionError, encrypt_secret
from apps.modelhub.models import ModelDeployment, ModelProvider
from apps.modelhub.structured_output import (
    StructuredOutputError,
    validate_mode_for_capability,
)

CHAT_CAPABILITIES = {"text_chat", "vision_chat"}
CAPABILITIES = CHAT_CAPABILITIES | {"embedding", "rerank"}


class ModelGovernanceError(Exception):
    def __init__(self, code: str, message: str, *, status: int = 400):
        super().__init__(message)
        self.code = code
        self.status = status


def _clean_name(value: str, field_label: str) -> str:
    cleaned = value.strip()
    if not cleaned:
        raise ModelGovernanceError("invalid_model_config", f"{field_label}不能为空")
    return cleaned


def validate_provider_values(values: dict[str, Any], provider: ModelProvider | None = None) -> dict:
    cleaned = dict(values)
    if "name" in cleaned:
        cleaned["name"] = _clean_name(cleaned["name"], "渠道名称")
    if "base_url" in cleaned:
        base_url = cleaned["base_url"].strip().rstrip("/")
        try:
            URLValidator(schemes=["http", "https"])(base_url)
        except ValidationError as error:
            raise ModelGovernanceError(
                "invalid_provider_url", "渠道地址必须是有效的 HTTP(S) URL"
            ) from error
        cleaned["base_url"] = base_url
    timeout = cleaned.get(
        "default_timeout_seconds",
        provider.default_timeout_seconds if provider else 60,
    )
    if not 1 <= timeout <= 300:
        raise ModelGovernanceError(
            "invalid_provider_timeout", "请求超时必须在 1 到 300 秒之间"
        )
    if cleaned.get("clear_api_key") and cleaned.get("api_key"):
        raise ModelGovernanceError(
            "credential_action_conflict", "不能同时替换和清除渠道凭证"
        )
    return cleaned


def validate_deployment_values(
    values: dict[str, Any],
    deployment: ModelDeployment | None = None,
) -> dict:
    cleaned = dict(values)
    for field, label in (("display_name", "显示名称"), ("model_name", "实际模型名")):
        if field in cleaned:
            cleaned[field] = _clean_name(cleaned[field], label)
    capability = cleaned.get(
        "capability", deployment.capability if deployment else "text_chat"
    )
    if capability not in CAPABILITIES:
        raise ModelGovernanceError("invalid_model_capability", "未知的模型能力")
    if capability not in CHAT_CAPABILITIES:
        cleaned["structured_output_mode"] = "none"
    temperature = cleaned.get(
        "default_temperature",
        deployment.default_temperature if deployment else 0.2,
    )
    if not 0 <= temperature <= 2:
        raise ModelGovernanceError(
            "invalid_model_temperature", "Temperature 必须在 0 到 2 之间"
        )
    max_tokens = cleaned.get(
        "max_output_tokens", deployment.max_output_tokens if deployment else 2048
    )
    if not 1 <= max_tokens <= 32768:
        raise ModelGovernanceError(
            "invalid_model_max_tokens", "最大输出 Token 必须在 1 到 32768 之间"
        )
    mode = cleaned.get(
        "structured_output_mode",
        deployment.structured_output_mode if deployment else "unconfigured",
    )
    try:
        validate_mode_for_capability(capability, mode)
    except StructuredOutputError as error:
        raise ModelGovernanceError(error.code, str(error)) from error
    return cleaned


def create_provider_config(values: dict[str, Any], user) -> ModelProvider:
    cleaned = validate_provider_values(values)
    api_key = cleaned.pop("api_key").strip()
    if not api_key:
        raise ModelGovernanceError(
            "credential_required", "创建渠道时必须录入有效凭证"
        )
    cleaned.pop("clear_api_key", None)
    approved = cleaned.get("image_egress_approved", False)
    now = timezone.now()
    try:
        encrypted = encrypt_secret(api_key)
        return ModelProvider.objects.create(
            **cleaned,
            api_key_encrypted=encrypted,
            credential_updated_at=now,
            credential_updated_by=user,
            image_egress_approved_by=user if approved else None,
            image_egress_approved_at=now if approved else None,
            created_by=user,
        )
    except CredentialEncryptionError as error:
        raise ModelGovernanceError("credential_encryption_failed", str(error)) from error
    except IntegrityError as error:
        raise ModelGovernanceError(
            "provider_name_conflict", "活跃渠道名称已存在", status=409
        ) from error


def update_provider_config(
    provider: ModelProvider,
    values: dict[str, Any],
    user,
) -> tuple[ModelProvider, dict[str, Any]]:
    if provider.archived_at:
        raise ModelGovernanceError(
            "model_config_archived", "归档渠道不允许编辑、测试或启用"
        )
    cleaned = validate_provider_values(values, provider)
    api_key = cleaned.pop("api_key", None)
    clear_api_key = cleaned.pop("clear_api_key", False)
    approved = cleaned.pop("image_egress_approved", None)
    if (
        cleaned.get("enabled") is True
        and not clear_api_key
        and not api_key
        and not provider.api_key_encrypted
    ):
        raise ModelGovernanceError(
            "credential_required", "启用渠道前必须录入有效凭证"
        )
    base_url_changed = (
        "base_url" in cleaned and cleaned["base_url"] != provider.base_url.rstrip("/")
    )
    credential_action = "unchanged"
    now = timezone.now()
    try:
        if api_key:
            provider.api_key_encrypted = encrypt_secret(api_key)
            provider.credential_updated_at = now
            provider.credential_updated_by = user
            credential_action = "replaced"
        elif clear_api_key:
            provider.api_key_encrypted = ""
            provider.credential_updated_at = now
            provider.credential_updated_by = user
            provider.enabled = False
            credential_action = "cleared"
        for field, value in cleaned.items():
            setattr(provider, field, value)
        if clear_api_key:
            provider.enabled = False
        if approved is not None:
            provider.image_egress_approved = approved
            provider.image_egress_approved_by = user if approved else None
            provider.image_egress_approved_at = now if approved else None
        provider.save()
    except CredentialEncryptionError as error:
        raise ModelGovernanceError("credential_encryption_failed", str(error)) from error
    except IntegrityError as error:
        raise ModelGovernanceError(
            "provider_name_conflict", "活跃渠道名称已存在", status=409
        ) from error
    if base_url_changed:
        provider.deployments.filter(
            archived_at__isnull=True,
            capability__in=CHAT_CAPABILITIES,
        ).update(
            structured_output_test_status="untested",
            structured_output_verified_at=None,
            structured_output_test_message="渠道地址已变更，请重新验证结构化输出",
        )
    return provider, {
        "fields": sorted([*cleaned, *(["image_egress_approved"] if approved is not None else [])]),
        "credential_action": credential_action,
    }


def create_deployment_config(values: dict[str, Any]) -> ModelDeployment:
    cleaned = validate_deployment_values(values)
    provider = ModelProvider.objects.get(id=cleaned.pop("provider_id"))
    if provider.archived_at:
        raise ModelGovernanceError(
            "model_config_archived", "不能在已归档渠道下创建模型"
        )
    try:
        return ModelDeployment.objects.create(provider=provider, **cleaned)
    except IntegrityError as error:
        raise ModelGovernanceError(
            "deployment_name_conflict",
            "同一活跃渠道下实际模型名已存在",
            status=409,
        ) from error


def update_deployment_config(
    deployment: ModelDeployment,
    values: dict[str, Any],
) -> tuple[ModelDeployment, dict[str, Any]]:
    if deployment.archived_at or deployment.provider.archived_at:
        raise ModelGovernanceError(
            "model_config_archived", "归档模型不允许编辑、测试或启用"
        )
    cleaned = validate_deployment_values(values, deployment)
    reset_test = any(
        field in cleaned and cleaned[field] != getattr(deployment, field)
        for field in {"model_name", "structured_output_mode", "capability"}
    )
    for field, value in cleaned.items():
        setattr(deployment, field, value)
    if reset_test:
        deployment.structured_output_test_status = "untested"
        deployment.structured_output_verified_at = None
        deployment.structured_output_test_message = (
            "模型配置已变更，请重新验证结构化输出"
            if deployment.capability in CHAT_CAPABILITIES
            else ""
        )
    try:
        deployment.save()
    except IntegrityError as error:
        raise ModelGovernanceError(
            "deployment_name_conflict",
            "同一活跃渠道下实际模型名已存在",
            status=409,
        ) from error
    return deployment, {"fields": sorted(cleaned), "structured_test_reset": reset_test}


def deployment_impact(deployment: ModelDeployment) -> dict[str, int | str]:
    counts = {
        "agent_version_count": deployment.agent_versions.count(),
        "step_binding_count": deployment.agent_step_bindings.count(),
        "run_step_count": deployment.agent_run_steps.count(),
        "invocation_count": deployment.invocations.count(),
    }
    return {
        "suggested_action": "archive" if any(counts.values()) else "delete",
        "deployment_count": 1,
        **counts,
    }


def provider_impact(provider: ModelProvider) -> dict[str, int | str]:
    from django.db.models import Count, Sum

    deployments = provider.deployments.annotate(
        agent_version_refs=Count("agent_versions", distinct=True),
        step_binding_refs=Count("agent_step_bindings", distinct=True),
        run_step_refs=Count("agent_run_steps", distinct=True),
        invocation_refs=Count("invocations", distinct=True),
    )
    totals = deployments.aggregate(
        agent_version_count=Sum("agent_version_refs"),
        step_binding_count=Sum("step_binding_refs"),
        run_step_count=Sum("run_step_refs"),
        invocation_count=Sum("invocation_refs"),
    )
    counts = {key: value or 0 for key, value in totals.items()}
    return {
        "suggested_action": "archive" if any(counts.values()) else "delete",
        "deployment_count": deployments.count(),
        **counts,
    }


@transaction.atomic
def remove_deployment(deployment_id: int, user) -> tuple[str, dict]:
    deployment = (
        ModelDeployment.objects.select_for_update()
        .select_related("provider")
        .get(id=deployment_id)
    )
    impact = deployment_impact(deployment)
    if deployment.archived_at:
        return "archived", impact
    if impact["suggested_action"] == "delete":
        deployment.delete()
        return "deleted", impact
    deployment.enabled = False
    deployment.archived_at = timezone.now()
    deployment.archived_by = user
    deployment.archived_via_provider = False
    deployment.save(
        update_fields=[
            "enabled",
            "archived_at",
            "archived_by",
            "archived_via_provider",
            "updated_at",
        ]
    )
    return "archived", impact


@transaction.atomic
def remove_provider(provider_id: int, user) -> tuple[str, dict]:
    provider = ModelProvider.objects.select_for_update().get(id=provider_id)
    list(provider.deployments.select_for_update())
    impact = provider_impact(provider)
    if provider.archived_at:
        return "archived", impact
    if impact["suggested_action"] == "delete":
        provider.deployments.all().delete()
        provider.delete()
        return "deleted", impact
    now = timezone.now()
    provider.deployments.filter(archived_at__isnull=True).update(
        enabled=False,
        archived_at=now,
        archived_by=user,
        archived_via_provider=True,
    )
    provider.enabled = False
    provider.archived_at = now
    provider.archived_by = user
    provider.api_key_encrypted = ""
    provider.credential_updated_at = now
    provider.credential_updated_by = user
    provider.image_egress_approved = False
    provider.image_egress_approved_by = None
    provider.image_egress_approved_at = None
    provider.save()
    return "archived", impact


@transaction.atomic
def restore_deployment(deployment_id: int) -> ModelDeployment:
    deployment = (
        ModelDeployment.objects.select_for_update()
        .select_related("provider")
        .get(id=deployment_id)
    )
    if deployment.archived_at is None:
        deployment.enabled = False
        deployment.save(update_fields=["enabled", "updated_at"])
        return deployment
    if deployment.provider.archived_at:
        raise ModelGovernanceError(
            "provider_archived", "请先恢复模型所属渠道", status=409
        )
    if ModelDeployment.objects.filter(
        provider_id=deployment.provider_id,
        model_name=deployment.model_name,
        archived_at__isnull=True,
    ).exclude(id=deployment.id).exists():
        raise ModelGovernanceError(
            "lifecycle_name_conflict", "恢复失败：活跃模型已复用该实际模型名", status=409
        )
    deployment.archived_at = None
    deployment.archived_by = None
    deployment.archived_via_provider = False
    deployment.enabled = False
    deployment.structured_output_test_status = "untested"
    deployment.structured_output_verified_at = None
    deployment.structured_output_test_message = (
        "模型已恢复，请重新验证"
        if deployment.capability in CHAT_CAPABILITIES
        else ""
    )
    if deployment.capability not in CHAT_CAPABILITIES:
        deployment.structured_output_mode = "none"
    deployment.save()
    return deployment


@transaction.atomic
def restore_provider(provider_id: int) -> ModelProvider:
    provider = ModelProvider.objects.select_for_update().get(id=provider_id)
    children = list(provider.deployments.select_for_update())
    if ModelProvider.objects.filter(
        name=provider.name, archived_at__isnull=True
    ).exclude(id=provider.id).exists():
        raise ModelGovernanceError(
            "lifecycle_name_conflict", "恢复失败：活跃渠道已复用该名称", status=409
        )
    for deployment in children:
        if not deployment.archived_via_provider:
            continue
        if ModelDeployment.objects.filter(
            provider_id=provider.id,
            model_name=deployment.model_name,
            archived_at__isnull=True,
        ).exclude(id=deployment.id).exists():
            raise ModelGovernanceError(
                "lifecycle_name_conflict",
                f"恢复失败：模型名 {deployment.model_name} 已被复用",
                status=409,
            )
    provider.archived_at = None
    provider.archived_by = None
    provider.enabled = False
    provider.save(update_fields=["archived_at", "archived_by", "enabled", "updated_at"])
    restored_ids = list(
        provider.deployments.filter(archived_via_provider=True).values_list(
            "id",
            flat=True,
        )
    )
    restored_deployments = provider.deployments.filter(id__in=restored_ids)
    restored_deployments.update(
        archived_at=None,
        archived_by=None,
        archived_via_provider=False,
        enabled=False,
        structured_output_test_status="untested",
        structured_output_verified_at=None,
    )
    restored_deployments.filter(capability__in=CHAT_CAPABILITIES).update(
        structured_output_test_message="渠道已恢复，请重新录入凭证并验证",
    )
    restored_deployments.exclude(capability__in=CHAT_CAPABILITIES).update(
        structured_output_mode="none",
        structured_output_test_message="",
    )
    return provider