112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
|
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
|