52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
# services/activity_log.py
|
|
"""Activity logging and user tracking system."""
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional
|
|
from collections import deque
|
|
import asyncio
|
|
|
|
class ActivityLogger:
|
|
def __init__(self, max_logs: int = 1000):
|
|
self.logs: deque = deque(maxlen=max_logs) # Keep last 1000 logs
|
|
self.active_users: Dict[str, datetime] = {} # username -> last_activity_time
|
|
self.lock = asyncio.Lock()
|
|
|
|
async def log_action(self, username: str, action: str, details: str = ""):
|
|
"""Log a user action"""
|
|
async with self.lock:
|
|
timestamp = datetime.now()
|
|
log_entry = {
|
|
"timestamp": timestamp.isoformat(),
|
|
"username": username,
|
|
"action": action,
|
|
"details": details
|
|
}
|
|
self.logs.append(log_entry)
|
|
|
|
# Update user activity
|
|
self.active_users[username] = timestamp
|
|
|
|
async def get_logs(self, limit: int = 100) -> List[Dict]:
|
|
"""Get recent logs"""
|
|
async with self.lock:
|
|
# Return most recent logs first
|
|
return list(self.logs)[-limit:][::-1]
|
|
|
|
async def get_active_users(self, timeout_minutes: int = 30) -> List[str]:
|
|
"""Get list of users active in the last N minutes"""
|
|
async with self.lock:
|
|
cutoff = datetime.now() - timedelta(minutes=timeout_minutes)
|
|
active = [
|
|
username for username, last_activity in self.active_users.items()
|
|
if last_activity >= cutoff
|
|
]
|
|
return sorted(active)
|
|
|
|
async def update_user_activity(self, username: str):
|
|
"""Update user's last activity timestamp"""
|
|
async with self.lock:
|
|
self.active_users[username] = datetime.now()
|
|
|
|
# Global instance
|
|
activity_logger = ActivityLogger()
|