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

from django.db import IntegrityError, transaction
from django.utils import timezone
from ninja import Router
from ninja.errors import HttpError

from apps.accounts.api import require_permission
from apps.audit.models import OperationAuditLog
from apps.skills.models import SkillDefinition, SkillVersion
from apps.skills.schemas import (
    SkillCreateIn,
    SkillDefinitionOut,
    SkillDefinitionUpdateIn,
    SkillDetailOut,
    SkillDraftUpdateIn,
    SkillVersionOut,
)
from apps.skills.services import SkillValidationError, ensure_draft, validate_skill_version

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


def serialize_definition(definition: SkillDefinition, detailed: bool = False) -> dict:
    draft = definition.versions.filter(status="draft").first()
    payload = {
        "id": definition.id,
        "code": definition.code,
        "name": definition.name,
        "description": definition.description,
        "active_version": definition.active_version.version
        if definition.active_version_id
        else None,
        "draft_version": draft.version if draft else None,
        "updated_at": definition.updated_at,
    }
    if detailed:
        payload["versions"] = list(definition.versions.all())
    return payload


def audit(user, action: str, definition: SkillDefinition, snapshot: dict) -> None:
    OperationAuditLog.objects.create(
        user=user,
        module="skills",
        action=action,
        target_type="skill",
        target_id=str(definition.id),
        request_snapshot=snapshot,
        result="success",
    )


@router.get("", response=list[SkillDefinitionOut])
def list_skills(request):
    require_permission(request, "skill.manage")
    queryset = SkillDefinition.objects.select_related("active_version").prefetch_related("versions")
    return [serialize_definition(item) for item in queryset]


@router.post("", response=SkillDetailOut)
def create_skill(request, payload: SkillCreateIn):
    require_permission(request, "skill.manage")
    try:
        with transaction.atomic():
            definition = SkillDefinition.objects.create(
                code=payload.code,
                name=payload.name,
                description=payload.description,
                created_by=request.user,
            )
            SkillVersion.objects.create(
                definition=definition,
                version=1,
                status="draft",
                prompt_template=payload.prompt_template,
                input_schema=payload.input_schema,
                output_schema=payload.output_schema,
                output_enforcement=payload.output_enforcement,
                example_input=payload.example_input,
                created_by=request.user,
            )
    except IntegrityError as error:
        raise HttpError(400, "Skill code already exists") from error
    audit(request.user, "create", definition, {"code": definition.code})
    return serialize_definition(definition, detailed=True)


@router.get("/{skill_id}", response=SkillDetailOut)
def get_skill(request, skill_id: int):
    require_permission(request, "skill.manage")
    definition = SkillDefinition.objects.select_related("active_version").get(id=skill_id)
    return serialize_definition(definition, detailed=True)


@router.patch("/{skill_id}", response=SkillDetailOut)
def update_skill(request, skill_id: int, payload: SkillDefinitionUpdateIn):
    require_permission(request, "skill.manage")
    definition = SkillDefinition.objects.select_related("active_version").get(id=skill_id)
    values = payload.dict(exclude_unset=True)
    for field, value in values.items():
        setattr(definition, field, value)
    definition.save()
    audit(request.user, "update", definition, {"fields": sorted(values)})
    return serialize_definition(definition, detailed=True)


@router.post("/{skill_id}/draft", response=SkillVersionOut)
def create_skill_draft(request, skill_id: int):
    require_permission(request, "skill.manage")
    definition = SkillDefinition.objects.select_related("active_version").get(id=skill_id)
    return ensure_draft(definition, request.user)


@router.get("/{skill_id}/draft", response=SkillVersionOut)
def get_skill_draft(request, skill_id: int):
    require_permission(request, "skill.manage")
    return SkillVersion.objects.get(definition_id=skill_id, status="draft")


@router.patch("/{skill_id}/draft", response=SkillVersionOut)
def update_skill_draft(request, skill_id: int, payload: SkillDraftUpdateIn):
    require_permission(request, "skill.manage")
    definition = SkillDefinition.objects.select_related("active_version").get(id=skill_id)
    draft = ensure_draft(definition, request.user)
    values = payload.dict(exclude_unset=True)
    for field, value in values.items():
        setattr(draft, field, value)
    try:
        validate_skill_version(draft)
    except SkillValidationError as error:
        raise HttpError(400, str(error)) from error
    draft.save()
    audit(request.user, "update_draft", definition, {"version": draft.version})
    return draft


@router.get("/{skill_id}/versions", response=list[SkillVersionOut])
def list_skill_versions(request, skill_id: int):
    require_permission(request, "skill.manage")
    return SkillVersion.objects.filter(definition_id=skill_id)


@router.post("/{skill_id}/publish", response=SkillVersionOut)
def publish_skill(request, skill_id: int):
    require_permission(request, "skill.manage")
    with transaction.atomic():
        definition = SkillDefinition.objects.select_for_update().get(id=skill_id)
        draft = SkillVersion.objects.select_for_update().get(
            definition=definition,
            status="draft",
        )
        try:
            validate_skill_version(draft)
        except SkillValidationError as error:
            raise HttpError(400, str(error)) from error
        draft.status = "published"
        draft.published_by = request.user
        draft.published_at = timezone.now()
        draft.save(update_fields=["status", "published_by", "published_at", "updated_at"])
        definition.active_version = draft
        definition.save(update_fields=["active_version", "updated_at"])
    audit(request.user, "publish", definition, {"version": draft.version})
    return draft


@router.post("/{skill_id}/rollback/{version}", response=SkillDetailOut)
def rollback_skill(request, skill_id: int, version: int):
    require_permission(request, "skill.manage")
    definition = SkillDefinition.objects.select_related("active_version").get(id=skill_id)
    target = SkillVersion.objects.get(
        definition=definition,
        version=version,
        status="published",
    )
    definition.active_version = target
    definition.save(update_fields=["active_version", "updated_at"])
    audit(request.user, "rollback", definition, {"version": target.version})
    return serialize_definition(definition, detailed=True)