""" Audit Log Service """ from datetime import datetime, timedelta, date, time from decimal import Decimal import uuid import pytz from typing import Optional, List, Dict, Any from sqlalchemy.orm import Session from ..models import AuditLog from .core import AuditMapper, ReferenceGenerator from core.security import verify_token # keep if needed or simpler just remove if unused # We don't need security import here anymore as context is passed explicitly or handled by events def _make_json_safe(obj: Any) -> Any: """Recursively convert non-JSON-serializable types to serializable equivalents.""" if obj is None: return None if isinstance(obj, dict): return {k: _make_json_safe(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_make_json_safe(v) for v in obj] if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, date): return obj.isoformat() if isinstance(obj, time): return obj.isoformat() if isinstance(obj, Decimal): return float(obj) if isinstance(obj, uuid.UUID): return str(obj) if isinstance(obj, bytes): return obj.decode("utf-8", errors="replace") if hasattr(obj, "__dict__"): d = dict(obj.__dict__) d.pop("_sa_instance_state", None) return {k: _make_json_safe(v) for k, v in d.items()} if not isinstance(obj, (int, float, str, bool)): return str(obj) return obj class AuditService: @staticmethod def create_audit_log( db: Session, reference: str, procedure: str, movement: str, username: str, system: str = "SCAF", # Extra context table_name: Optional[str] = None, record_id: Optional[str] = None, operation_type: Optional[str] = None, old_values: Optional[Dict] = None, new_values: Optional[Dict] = None, changed_fields: Optional[List[str]] = None, # HTTP Context ip_address: Optional[str] = None, user_agent: Optional[str] = None, endpoint: Optional[str] = None, request_method: Optional[str] = None, session_id: Optional[str] = None, company_id: Optional[int] = None, tenant_id: Optional[int] = None, ) -> AuditLog: """ Low-level creation of an Audit Log entry """ # Timezone handling: Use UTC for consistency across regions. # Frontend will convert to user's local time. now = datetime.now(pytz.UTC) log = AuditLog( reference=reference, procedure=procedure, movement=movement, username=username, date=now.date(), time=now.time(), timestamp=now, system=system, table_name=table_name, record_id=record_id, operation_type=operation_type, old_values=_make_json_safe(old_values), new_values=_make_json_safe(new_values), changed_fields=changed_fields, ip_address=ip_address, user_agent=user_agent, endpoint=endpoint, request_method=request_method, session_id=session_id, company_id=company_id, tenant_id=tenant_id ) db.add(log) db.commit() db.refresh(log) return log @staticmethod def log_crud_operation( db: Session, table_name: str, operation_type: str, record_data: Dict[str, Any], username: str, record_id: Optional[str] = None, old_values: Optional[Dict] = None, new_values: Optional[Dict] = None, # Context ip_address: Optional[str] = None, user_agent: Optional[str] = None, company_id: Optional[int] = None, tenant_id: Optional[int] = None ): """ High-level wrapper to log CRUD operations automatically mapping to Legacy format """ # 1. Map to Legacy Base Format legacy_data = AuditMapper.map_to_legacy_format( table_name=table_name, record_id=record_id, operation_type=operation_type, username=username ) # 2. Refine Reference based on specific table logic reference = legacy_data["reference"] if table_name == "invoice_header": generated_ref = ReferenceGenerator.generate_invoice_reference(record_data) # Use generated ref only if meaningful, else keep default if generated_ref != "FUSE0-000-00": reference = generated_ref elif table_name == "pedimentos": reference = ReferenceGenerator.generate_pedimento_reference(record_data) elif table_name == "clients_and_providers": reference = record_data.get("rfc") or reference elif table_name == "parts": reference = record_data.get("part_number") or reference elif table_name == "companies": reference = record_data.get("rfc") or reference elif table_name == "classes": reference = record_data.get("class_code") or reference # 3. Detect Changed Fields (for Update) changed_fields = None if operation_type == "UPDATE" and old_values and new_values: changed_fields = [ k for k in new_values.keys() if old_values.get(k) != new_values.get(k) ] # 4. Create Log return AuditService.create_audit_log( db=db, reference=reference, procedure=legacy_data["procedure"], movement=legacy_data["movement"], username=username, system=legacy_data["system"], table_name=table_name, record_id=record_id, operation_type=operation_type, old_values=old_values, new_values=new_values, changed_fields=changed_fields, ip_address=ip_address, user_agent=user_agent, company_id=company_id, tenant_id=tenant_id ) @staticmethod def log_login(db: Session, username: str, ip_address: str = None, user_agent: str = None): # Prevent duplicate login logs (debounce 5 seconds) # This handles cases where frontend might submit twice or redirects trigger re-auth try: # Timezone handling: Use UTC for consistency now = datetime.now(pytz.UTC) five_seconds_ago = now - timedelta(seconds=5) # Check for recent login from same user existing = db.query(AuditLog).filter( AuditLog.username == username, AuditLog.operation_type == "LOGIN", # Compare against timestamp (timezone aware) AuditLog.timestamp >= five_seconds_ago ).first() if existing: return existing except Exception as e: import traceback traceback.print_exc() return AuditService.create_audit_log( db=db, reference="LOGIN", procedure="SYSTEM SCAF", movement="SYSTEM LOGIN", username=username, operation_type="LOGIN", ip_address=ip_address, user_agent=user_agent ) @staticmethod def log_logout(db: Session, username: str, ip_address: str = None, user_agent: str = None): """ Registra un evento de cierre de sesión """ try: # Reutilizamos create_audit_log para mantener consistencia AuditService.create_audit_log( db=db, reference="LOGOUT", procedure="SYSTEM AUTH", movement="SYSTEM LOGOUT", username=username, operation_type="LOGOUT", ip_address=ip_address, user_agent=user_agent ) except Exception as e: # No re-lanzamos la excepción para no interrumpir el flujo de logout pass