dify-1.8.1/api/services/dataset_operation_log_service.py

"""
Dataset Operation Log Service
Records all user operations on datasets for audit and compliance
"""

from datetime import datetime
from enum import Enum
from typing import Any, Optional

from flask import request
from sqlalchemy import and_, or_

from extensions.ext_database import db
from models.model import OperationLog


class DatasetOperationType(str, Enum):
    """Dataset operation types"""

    # Dataset operations
    DATASET_CREATE = "dataset.create"
    DATASET_UPDATE = "dataset.update"
    DATASET_DELETE = "dataset.delete"

    # Document operations
    DOCUMENT_CREATE = "document.create"
    DOCUMENT_UPDATE = "document.update"
    DOCUMENT_DELETE = "document.delete"
    DOCUMENT_BATCH_DELETE = "document.batch_delete"
    DOCUMENT_RENAME = "document.rename"
    # Document status operations
    DOCUMENT_ENABLE = "document.enable"
    DOCUMENT_DISABLE = "document.disable"
    DOCUMENT_ARCHIVE = "document.archive"
    DOCUMENT_UNARCHIVE = "document.unarchive"
    DOCUMENT_BATCH_ENABLE = "document.batch_enable"
    DOCUMENT_BATCH_DISABLE = "document.batch_disable"
    DOCUMENT_BATCH_ARCHIVE = "document.batch_archive"
    DOCUMENT_BATCH_UNARCHIVE = "document.batch_unarchive"
    # Document schedule operations
    DOCUMENT_SCHEDULE_CREATE = "document.schedule.create"
    DOCUMENT_SCHEDULE_CANCEL = "document.schedule.cancel"
    DOCUMENT_SCHEDULE_EXECUTE = "document.schedule.execute"

    # Segment operations
    SEGMENT_CREATE = "segment.create"
    SEGMENT_UPDATE = "segment.update"
    SEGMENT_DELETE = "segment.delete"
    SEGMENT_BATCH_DELETE = "segment.batch_delete"
    SEGMENT_ENABLE = "segment.enable"
    SEGMENT_DISABLE = "segment.disable"

    # Metadata operations
    METADATA_CREATE = "metadata.create"
    METADATA_UPDATE = "metadata.update"
    METADATA_DELETE = "metadata.delete"


class DatasetOperationLogService:
    """Service for logging dataset operations"""

    @staticmethod
    def log_operation(
        tenant_id: str,
        account_id: str,
        action: DatasetOperationType,
        dataset_id: str,
        resource_type: str,  # 'dataset', 'document', 'segment'
        resource_id: str,
        changes: Optional[dict] = None,
        metadata: Optional[dict] = None,
    ) -> OperationLog:
        """
        Record a dataset operation

        Args:
            tenant_id: Tenant ID
            account_id: Account ID of the operator
            action: Operation type
            dataset_id: Dataset ID
            resource_type: Resource type
            resource_id: Resource ID
            changes: Change details (e.g., {"old": {...}, "new": {...}})
            metadata: Additional metadata

        Returns:
            Created OperationLog instance
        """
        content = {
            "dataset_id": dataset_id,
            "resource_type": resource_type,
            "resource_id": resource_id,
            "action": action.value,
            "timestamp": datetime.utcnow().isoformat(),
        }

        if changes:
            content["changes"] = changes

        if metadata:
            content["metadata"] = metadata

        log = OperationLog(
            tenant_id=tenant_id,
            account_id=account_id,
            action=action.value,
            content=content,
            created_ip=request.remote_addr or "0.0.0.0" if request else "0.0.0.0",
        )

        db.session.add(log)
        db.session.commit()

        return log

    @staticmethod
    def get_dataset_operation_logs(
        tenant_id: str,
        dataset_id: Optional[str] = None,
        resource_type: Optional[str] = None,
        resource_id: Optional[str] = None,
        action: Optional[DatasetOperationType] = None,
        account_id: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        page: int = 1,
        per_page: int = 20,
    ) -> dict[str, Any]:
        """
        Query dataset operation logs

        Args:
            tenant_id: Tenant ID
            dataset_id: Filter by dataset ID
            resource_type: Filter by resource type
            resource_id: Filter by resource ID
            action: Filter by action type
            account_id: Filter by account ID
            start_date: Filter by start date
            end_date: Filter by end date
            page: Page number
            per_page: Items per page

        Returns:
            Dict with logs and pagination info
        """
        query = db.session.query(OperationLog).filter(
            OperationLog.tenant_id == tenant_id,
            or_(
                OperationLog.action.like("dataset.%"),
                OperationLog.action.like("document.%"),
                OperationLog.action.like("segment.%"),
                OperationLog.action.like("metadata.%"),
            ),
        )

        if dataset_id:
            query = query.filter(OperationLog.content["dataset_id"].as_string() == dataset_id)

        if resource_type:
            query = query.filter(OperationLog.content["resource_type"].as_string() == resource_type)

        if resource_id:
            query = query.filter(OperationLog.content["resource_id"].as_string() == resource_id)

        if action:
            query = query.filter(OperationLog.action == action.value)

        if account_id:
            query = query.filter(OperationLog.account_id == account_id)

        if start_date:
            query = query.filter(OperationLog.created_at >= start_date)

        if end_date:
            query = query.filter(OperationLog.created_at <= end_date)

        query = query.order_by(OperationLog.created_at.desc())

        paginated = query.paginate(page=page, per_page=per_page, error_out=False)

        return {
            "logs": paginated.items,
            "total": paginated.total,
            "page": page,
            "per_page": per_page,
            "pages": paginated.pages,
        }