""" Document Operation Logs API Query document operation history for audit purposes """ from datetime import datetime from flask import request from flask_login import current_user from flask_restx import Resource, fields, marshal_with, reqparse from werkzeug.exceptions import BadRequest, NotFound from controllers.console import api from controllers.console.wraps import account_initialization_required, setup_required from libs.login import login_required from services.dataset_operation_log_service import DatasetOperationLogService, DatasetOperationType from services.dataset_service import DatasetService, DocumentService # Response fields for operation log operation_log_fields = { "id": fields.String, "action": fields.String, "created_at": fields.DateTime(dt_format="iso8601"), "created_ip": fields.String, "content": fields.Raw, } operation_log_list_fields = { "logs": fields.List(fields.Nested(operation_log_fields)), "total": fields.Integer, "page": fields.Integer, "per_page": fields.Integer, "pages": fields.Integer, } class DocumentOperationLogsResource(Resource): """Base resource for document operation logs""" pass class DocumentOperationLogsApi(DocumentOperationLogsResource): """API for querying document operation logs""" @setup_required @login_required @account_initialization_required @marshal_with(operation_log_list_fields) def get(self, dataset_id: str, document_id: str): """ Get operation logs for a specific document Args: dataset_id: Dataset ID document_id: Document ID Query Parameters: - action: Filter by action type (e.g., document.enable, document.disable) - start_date: Filter by start date (ISO 8601 format) - end_date: Filter by end date (ISO 8601 format) - page: Page number (default: 1) - per_page: Items per page (default: 20, max: 100) Returns: dict: Paginated list of operation logs """ dataset_id_str = str(dataset_id) document_id_str = str(document_id) # Get dataset and check permissions dataset = DatasetService.get_dataset(dataset_id_str) if not dataset: raise NotFound("Dataset not found.") DatasetService.check_dataset_permission(dataset, current_user) # Get document to verify it exists document = DocumentService.get_document(dataset_id_str, document_id_str) if not document: raise NotFound("Document not found.") # Parse query parameters parser = reqparse.RequestParser() parser.add_argument("action", type=str, location="args") parser.add_argument("start_date", type=str, location="args") parser.add_argument("end_date", type=str, location="args") parser.add_argument("page", type=int, default=1, location="args") parser.add_argument("per_page", type=int, default=20, location="args") args = parser.parse_args() # Validate and parse dates start_date = None end_date = None if args["start_date"]: try: start_date = datetime.fromisoformat(args["start_date"].replace("Z", "+00:00")) except ValueError: raise BadRequest("Invalid start_date format. Use ISO 8601 format (e.g., 2024-01-01T00:00:00Z)") if args["end_date"]: try: end_date = datetime.fromisoformat(args["end_date"].replace("Z", "+00:00")) except ValueError: raise BadRequest("Invalid end_date format. Use ISO 8601 format (e.g., 2024-01-01T00:00:00Z)") # Validate action type action_type = None if args["action"]: try: action_type = DatasetOperationType(args["action"]) except ValueError: raise BadRequest(f"Invalid action type: {args['action']}") # Limit per_page to prevent abuse per_page = min(args["per_page"], 100) # Query operation logs result = DatasetOperationLogService.get_dataset_operation_logs( tenant_id=str(dataset.tenant_id), dataset_id=dataset_id_str, resource_type="document", resource_id=document_id_str, action=action_type, start_date=start_date, end_date=end_date, page=args["page"], per_page=per_page, ) return result class DatasetOperationLogsApi(DocumentOperationLogsResource): """API for querying all operation logs in a dataset""" @setup_required @login_required @account_initialization_required @marshal_with(operation_log_list_fields) def get(self, dataset_id: str): """ Get operation logs for an entire dataset Args: dataset_id: Dataset ID Query Parameters: - resource_type: Filter by resource type (document, segment, metadata) - action: Filter by action type - start_date: Filter by start date (ISO 8601 format) - end_date: Filter by end date (ISO 8601 format) - page: Page number (default: 1) - per_page: Items per page (default: 20, max: 100) Returns: dict: Paginated list of operation logs """ dataset_id_str = str(dataset_id) # Get dataset and check permissions dataset = DatasetService.get_dataset(dataset_id_str) if not dataset: raise NotFound("Dataset not found.") DatasetService.check_dataset_permission(dataset, current_user) # Parse query parameters parser = reqparse.RequestParser() parser.add_argument("resource_type", type=str, location="args") parser.add_argument("action", type=str, location="args") parser.add_argument("start_date", type=str, location="args") parser.add_argument("end_date", type=str, location="args") parser.add_argument("page", type=int, default=1, location="args") parser.add_argument("per_page", type=int, default=20, location="args") args = parser.parse_args() # Validate resource type if args["resource_type"] and args["resource_type"] not in ["document", "segment", "metadata", "dataset"]: raise BadRequest(f"Invalid resource_type: {args['resource_type']}") # Validate and parse dates start_date = None end_date = None if args["start_date"]: try: start_date = datetime.fromisoformat(args["start_date"].replace("Z", "+00:00")) except ValueError: raise BadRequest("Invalid start_date format. Use ISO 8601 format (e.g., 2024-01-01T00:00:00Z)") if args["end_date"]: try: end_date = datetime.fromisoformat(args["end_date"].replace("Z", "+00:00")) except ValueError: raise BadRequest("Invalid end_date format. Use ISO 8601 format (e.g., 2024-01-01T00:00:00Z)") # Validate action type action_type = None if args["action"]: try: action_type = DatasetOperationType(args["action"]) except ValueError: raise BadRequest(f"Invalid action type: {args['action']}") # Limit per_page to prevent abuse per_page = min(args["per_page"], 100) # Query operation logs result = DatasetOperationLogService.get_dataset_operation_logs( tenant_id=str(dataset.tenant_id), dataset_id=dataset_id_str, resource_type=args["resource_type"], action=action_type, start_date=start_date, end_date=end_date, page=args["page"], per_page=per_page, ) return result # Register API routes api.add_resource( DocumentOperationLogsApi, "/datasets//documents//operation-logs", ) api.add_resource( DatasetOperationLogsApi, "/datasets//operation-logs", )