Unified_Authentication_Platform/backend/app/services/log_service.py

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