from __future__ import annotations from typing import Any from django.db import models from .models import AiConfigAuditLog, AiModel, ModelAlias AI_MODEL_AUDIT_FIELDS = ( "name", "url", "model", "api_type", "api_key_encrypted", "capabilities", "timeout_seconds", "connect_timeout_seconds", "extra_body", "is_active", ) MODEL_ALIAS_AUDIT_FIELDS = ( "alias", "operation_type", "ai_model_id", "is_default", "is_active", ) FIELD_ALIASES = {"api_key_encrypted": "api_key", "ai_model_id": "ai_model"} def create_config_audit_log( *, actor: Any, action: str, instance: AiModel | ModelAlias, before: dict[str, Any] | None, after: dict[str, Any] | None, target_id: int | None = None, target_repr: str | None = None, ) -> AiConfigAuditLog | None: """Create a sanitized audit log for AI config changes.""" changed_fields, changes = _diff_snapshots(before, after) if action == AiConfigAuditLog.Action.UPDATE and not changes: return None actor_value = actor if getattr(actor, "is_authenticated", False) else None return AiConfigAuditLog.objects.create( actor=actor_value, action=action, target_type=_target_type(instance), target_id=target_id if target_id is not None else instance.pk, target_repr=(target_repr or str(instance))[:255], changed_fields=changed_fields, changes=changes, ) def snapshot_config(instance: AiModel | ModelAlias) -> dict[str, Any]: fields = _audit_fields(instance) return {field: _field_value(instance, field) for field in fields} def _diff_snapshots( before: dict[str, Any] | None, after: dict[str, Any] | None, ) -> tuple[list[str], dict[str, dict[str, Any]]]: fields = tuple((after or before or {}).keys()) changed_fields = [] changes = {} for field in fields: old_value = None if before is None else before.get(field) new_value = None if after is None else after.get(field) if old_value == new_value: continue public_field = FIELD_ALIASES.get(field, field) changed_fields.append(public_field) changes[public_field] = { "old": _sanitize_value(field, old_value), "new": _sanitize_value(field, new_value), } return changed_fields, changes def _target_type(instance: AiModel | ModelAlias) -> str: if isinstance(instance, AiModel): return AiConfigAuditLog.TargetType.AI_MODEL if isinstance(instance, ModelAlias): return AiConfigAuditLog.TargetType.MODEL_ALIAS raise TypeError(f"unsupported audit target: {type(instance)!r}") def _audit_fields(instance: AiModel | ModelAlias) -> tuple[str, ...]: if isinstance(instance, AiModel): return AI_MODEL_AUDIT_FIELDS if isinstance(instance, ModelAlias): return MODEL_ALIAS_AUDIT_FIELDS raise TypeError(f"unsupported audit target: {type(instance)!r}") def _field_value(instance: models.Model, field: str) -> Any: value = getattr(instance, field) if isinstance(value, (list, dict, str, int, bool)) or value is None: return value return str(value) def _sanitize_value(field: str, value: Any) -> Any: if field == "api_key_encrypted": return "set" if value else "empty" return value