from flask import request from models import Event, EventType, db from typing import Optional, Dict, Any, List from datetime import datetime from flask_login import current_user from sqlalchemy import desc def log_event(event_type: str, details: Optional[Dict[str, Any]] = None, user_id: Optional[int] = None) -> Event: """ Log an event to the database. Args: event_type: The type of event (must match EventType enum) details: Optional dictionary containing event details user_id: Optional user ID (defaults to current user) Returns: The created Event object """ if user_id is None and current_user.is_authenticated: user_id = current_user.id event = Event( event_type=event_type, user_id=user_id, timestamp=datetime.utcnow(), details=details or {}, ip_address=request.remote_addr if request else None, user_agent=request.user_agent.string if request and request.user_agent else None ) db.session.add(event) db.session.commit() return event def get_user_events(user_id: int, limit: int = 50) -> List[Event]: """Get recent events for a specific user""" return Event.query.filter_by(user_id=user_id).order_by(desc(Event.timestamp)).limit(limit).all() def get_room_events(room_id: int, limit: int = 50) -> List[Event]: """Get recent events for a specific room""" return Event.query.filter( Event.details['room_id'].astext == str(room_id) ).order_by(desc(Event.timestamp)).limit(limit).all() def get_recent_events(limit: int = 50) -> List[Event]: """Get most recent events across all types""" return Event.query.order_by(desc(Event.timestamp)).limit(limit).all() def get_events_by_type(event_type: str, limit: int = 50) -> List[Event]: """Get recent events of a specific type""" return Event.query.filter_by(event_type=event_type).order_by(desc(Event.timestamp)).limit(limit).all() def get_events_by_date_range(start_date: datetime, end_date: datetime, limit: int = 50) -> List[Event]: """Get events within a date range""" return Event.query.filter( Event.timestamp >= start_date, Event.timestamp <= end_date ).order_by(desc(Event.timestamp)).limit(limit).all()