Ryan Kegel
3dee8b80a2
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 24s
- Added tracking events for tasks, penalties, and rewards with timestamps. - Created new TinyDB table for tracking records to maintain audit history. - Developed backend API for querying tracking events with filters and pagination. - Implemented logging for tracking events with per-user rotating log files. - Added unit tests for tracking event creation, querying, and anonymization. - Deferred frontend changes for future implementation. - Established acceptance criteria and documentation for the tracking feature. feat: Introduce account deletion scheduler - Implemented a scheduler to delete accounts marked for deletion after a configurable threshold. - Added new fields to the User model to manage deletion status and attempts. - Created admin API endpoints for managing deletion thresholds and viewing the deletion queue. - Integrated error handling and logging for the deletion process. - Developed unit tests for the deletion scheduler and related API endpoints. - Documented the deletion process and acceptance criteria.
85 lines
2.3 KiB
Python
85 lines
2.3 KiB
Python
"""Per-user rotating audit logger for tracking events."""
|
|
import logging
|
|
import os
|
|
from logging.handlers import RotatingFileHandler
|
|
from config.paths import get_logs_dir
|
|
from models.tracking_event import TrackingEvent
|
|
|
|
|
|
# Store handlers per user_id to avoid recreating
|
|
_user_loggers = {}
|
|
|
|
|
|
def get_tracking_logger(user_id: str) -> logging.Logger:
|
|
"""
|
|
Get or create a per-user rotating file logger for tracking events.
|
|
|
|
Args:
|
|
user_id: User ID for the log file
|
|
|
|
Returns:
|
|
Logger instance configured for the user
|
|
"""
|
|
if user_id in _user_loggers:
|
|
return _user_loggers[user_id]
|
|
|
|
logs_dir = get_logs_dir()
|
|
os.makedirs(logs_dir, exist_ok=True)
|
|
|
|
log_file = os.path.join(logs_dir, f'tracking_user_{user_id}.log')
|
|
|
|
logger = logging.getLogger(f'tracking.user.{user_id}')
|
|
logger.setLevel(logging.INFO)
|
|
logger.propagate = False # Don't propagate to root logger
|
|
|
|
# Rotating file handler: 10MB max, keep 5 backups
|
|
handler = RotatingFileHandler(
|
|
log_file,
|
|
maxBytes=10 * 1024 * 1024, # 10MB
|
|
backupCount=5,
|
|
encoding='utf-8'
|
|
)
|
|
|
|
formatter = logging.Formatter(
|
|
'%(asctime)s | %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
handler.setFormatter(formatter)
|
|
|
|
logger.addHandler(handler)
|
|
_user_loggers[user_id] = logger
|
|
|
|
return logger
|
|
|
|
|
|
def log_tracking_event(event: TrackingEvent) -> None:
|
|
"""
|
|
Log a tracking event to the user's audit log file.
|
|
|
|
Args:
|
|
event: TrackingEvent to log
|
|
"""
|
|
if not event.user_id:
|
|
# If user was deleted (anonymized), skip logging
|
|
return
|
|
|
|
logger = get_tracking_logger(event.user_id)
|
|
|
|
log_msg = (
|
|
f"user_id={event.user_id} | "
|
|
f"child_id={event.child_id} | "
|
|
f"entity_type={event.entity_type} | "
|
|
f"entity_id={event.entity_id} | "
|
|
f"action={event.action} | "
|
|
f"points_before={event.points_before} | "
|
|
f"points_after={event.points_after} | "
|
|
f"delta={event.delta:+d} | "
|
|
f"occurred_at={event.occurred_at}"
|
|
)
|
|
|
|
if event.metadata:
|
|
metadata_str = ' | '.join(f"{k}={v}" for k, v in event.metadata.items())
|
|
log_msg += f" | {metadata_str}"
|
|
|
|
logger.info(log_msg)
|