from sqlalchemy.orm import Session from sqlalchemy import desc, or_ from typing import Optional, List, Dict, Any from datetime import datetime from app.models.operation_log import OperationLog from app.schemas.operation_log import ActionType class LogService: @staticmethod def create_log( db: Session, operator_id: int, action_type: ActionType, app_id: Optional[int] = None, target_user_id: Optional[int] = None, target_mobile: Optional[str] = None, ip_address: Optional[str] = None, details: Optional[Dict[str, Any]] = None ) -> OperationLog: log = OperationLog( app_id=app_id, operator_id=operator_id, action_type=action_type, target_user_id=target_user_id, target_mobile=target_mobile, ip_address=ip_address, details=details ) db.add(log) db.commit() db.refresh(log) return log @staticmethod def get_logs( db: Session, skip: int = 0, limit: int = 20, app_id: Optional[int] = None, action_type: Optional[ActionType] = None, operator_id: Optional[int] = None, keyword: Optional[str] = None, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None ): query = db.query(OperationLog) if app_id is not None: query = query.filter(OperationLog.app_id == app_id) if action_type: query = query.filter(OperationLog.action_type == action_type) if operator_id: query = query.filter(OperationLog.operator_id == operator_id) if keyword: # Search in target_mobile or operator mobile (need join for operator mobile) # For simplicity, let's search target_mobile first # If we want to search operator name/mobile, we'd join. # Let's assume keyword searches target_mobile for now. query = query.filter(OperationLog.target_mobile.ilike(f"%{keyword}%")) if start_date: query = query.filter(OperationLog.created_at >= start_date) if end_date: query = query.filter(OperationLog.created_at <= end_date) total = query.count() logs = query.order_by(desc(OperationLog.created_at)).offset(skip).limit(limit).all() return total, logs