163 lines
5.3 KiB
Python
163 lines
5.3 KiB
Python
"""
|
|
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 set to Mexico City as requested implicitly by legacy format example
|
|
tz = pytz.timezone('America/Mexico_City')
|
|
now = datetime.now(tz)
|
|
|
|
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
|
|
|
|
|
|
# 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):
|
|
return AuditService.create_audit_log(
|
|
db=db,
|
|
reference="LOGIN",
|
|
procedure="SYSTEM SCAF",
|
|
movement="SYSTEM LOGIN",
|
|
username=username,
|
|
operation_type="LOGIN",
|
|
ip_address=ip_address
|
|
)
|