AI/ontology_platform/ont_platform/audit/models.py

"""감사 로그 모델.

Phase 8: 감시 및 규정 준수
"""

from dataclasses import dataclass, field
from datetime import datetime, UTC
from enum import Enum
from typing import Optional, Any, Dict
import uuid


class AuditAction(str, Enum):
    """감시 작업 타입."""

    # CRUD 작업
    CREATE = "CREATE"
    READ = "READ"
    UPDATE = "UPDATE"
    DELETE = "DELETE"

    # 분석 작업
    ANALYZE = "ANALYZE"
    QUERY = "QUERY"

    # LLM 작업
    LLM_CALL = "LLM_CALL"
    LLM_STREAM = "LLM_STREAM"

    # 사용자 관리
    USER_LOGIN = "USER_LOGIN"
    USER_LOGOUT = "USER_LOGOUT"
    USER_CREATED = "USER_CREATED"
    USER_UPDATED = "USER_UPDATED"
    USER_DELETED = "USER_DELETED"

    # API 키
    API_KEY_CREATED = "API_KEY_CREATED"
    API_KEY_DELETED = "API_KEY_DELETED"
    API_KEY_USED = "API_KEY_USED"

    # 조직
    ORG_CREATED = "ORG_CREATED"
    ORG_UPDATED = "ORG_UPDATED"


class ResourceType(str, Enum):
    """리소스 타입."""

    ENTITY = "ENTITY"
    RELATION = "RELATION"
    GRAPH = "GRAPH"
    USER = "USER"
    API_KEY = "API_KEY"
    ORGANIZATION = "ORGANIZATION"
    QUERY = "QUERY"


@dataclass
class Change:
    """변경 사항."""

    field_name: str
    old_value: Any = None
    new_value: Any = None

    def to_dict(self) -> dict:
        """딕셔너리로 변환."""
        return {
            "field": self.field_name,
            "old": str(self.old_value),
            "new": str(self.new_value),
        }


@dataclass
class AuditLog:
    """감시 로그."""

    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    org_id: str = ""  # 조직 ID
    user_id: str = ""  # 사용자 ID
    action: AuditAction = AuditAction.READ
    resource_type: ResourceType = ResourceType.ENTITY
    resource_id: str = ""  # 대상 엔티티 ID
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    status: str = "success"  # "success", "failed"
    error_message: Optional[str] = None
    changes: list = field(default_factory=list)  # Change 객체 리스트
    metadata: dict = field(default_factory=dict)

    def to_dict(self) -> dict:
        """딕셔너리로 변환."""
        return {
            "id": self.id,
            "org_id": self.org_id,
            "user_id": self.user_id,
            "action": self.action.value,
            "resource_type": self.resource_type.value,
            "resource_id": self.resource_id,
            "timestamp": self.timestamp.isoformat(),
            "ip_address": self.ip_address,
            "status": self.status,
            "error_message": self.error_message,
            "changes": [c.to_dict() if isinstance(c, Change) else c for c in self.changes],
            "metadata": self.metadata,
        }

    def to_neo4j_dict(self) -> dict:
        """Neo4j 저장용 딕셔너리."""
        return {
            "audit_id": self.id,
            "org_id": self.org_id,
            "user_id": self.user_id,
            "action": self.action.value,
            "resource_type": self.resource_type.value,
            "resource_id": self.resource_id,
            "timestamp": self.timestamp.timestamp(),
            "ip_address": self.ip_address or "",
            "status": self.status,
            "error_message": self.error_message or "",
            "changes": str(self.changes),
        }


@dataclass
class AuditQuery:
    """감사 로그 쿼리."""

    org_id: str = ""
    user_id: Optional[str] = None
    action: Optional[AuditAction] = None
    resource_type: Optional[ResourceType] = None
    resource_id: Optional[str] = None
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None
    status: Optional[str] = None
    limit: int = 100
    offset: int = 0

    def to_cypher_filters(self) -> tuple[str, dict]:
        """Cypher 필터 생성."""
        filters = []
        params = {}

        if self.org_id:
            filters.append("log.org_id = $org_id")
            params["org_id"] = self.org_id

        if self.user_id:
            filters.append("log.user_id = $user_id")
            params["user_id"] = self.user_id

        if self.action:
            filters.append("log.action = $action")
            params["action"] = self.action.value

        if self.resource_type:
            filters.append("log.resource_type = $resource_type")
            params["resource_type"] = self.resource_type.value

        if self.resource_id:
            filters.append("log.resource_id = $resource_id")
            params["resource_id"] = self.resource_id

        if self.start_time:
            filters.append("log.timestamp >= $start_time")
            params["start_time"] = self.start_time.timestamp()

        if self.end_time:
            filters.append("log.timestamp <= $end_time")
            params["end_time"] = self.end_time.timestamp()

        if self.status:
            filters.append("log.status = $status")
            params["status"] = self.status

        where_clause = " AND ".join(filters) if filters else "1=1"
        return where_clause, params