""" Audit Log Service """ from datetime import datetime 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 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=old_values, new_values=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 ): """ 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 ) @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 - datetime.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: print(f"[AUDIT DEBUG] Duplicate login skipped for {username} within 5s") return existing except Exception as e: print(f"[AUDIT WARNING] Failed to check duplicate login: {e}") 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 print(f"Error logging logout: {e}")