281 lines
8.4 KiB
Python
281 lines
8.4 KiB
Python
|
"""
|
||
|
|
Audit Logging Service for Time Tracking Module
|
||
|
|
===============================================
|
||
|
|
|
||
|
|
Fuld sporbarhed af alle handlinger i modulet.
|
||
|
|
Alle events logges til tmodule_sync_log for compliance.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import json
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Optional, Dict, Any
|
||
|
|
from app.core.database import execute_insert
|
||
|
|
from app.timetracking.backend.models import TModuleSyncLogCreate
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
class AuditLogger:
|
||
|
|
"""
|
||
|
|
Service til at logge alle module events.
|
||
|
|
|
||
|
|
Event Types:
|
||
|
|
- sync_started, sync_completed, sync_failed
|
||
|
|
- approval, rejection, bulk_approval
|
||
|
|
- order_created, order_updated, order_cancelled
|
||
|
|
- export_started, export_completed, export_failed
|
||
|
|
- module_installed, module_uninstalled
|
||
|
|
"""
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_event(
|
||
|
|
event_type: str,
|
||
|
|
entity_type: Optional[str] = None,
|
||
|
|
entity_id: Optional[int] = None,
|
||
|
|
user_id: Optional[int] = None,
|
||
|
|
details: Optional[Dict[str, Any]] = None,
|
||
|
|
ip_address: Optional[str] = None,
|
||
|
|
user_agent: Optional[str] = None
|
||
|
|
) -> int:
|
||
|
|
"""
|
||
|
|
Log en event til tmodule_sync_log.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
event_type: Type af event (sync_started, approval, etc.)
|
||
|
|
entity_type: Type af entitet (time_entry, order, customer, case)
|
||
|
|
entity_id: ID på entiteten
|
||
|
|
user_id: ID på brugeren der foretog handlingen
|
||
|
|
details: Ekstra detaljer som JSON
|
||
|
|
ip_address: IP-adresse
|
||
|
|
user_agent: User agent string
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
ID på den oprettede log-entry
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
# Konverter details til JSON-string hvis det er en dict
|
||
|
|
details_json = json.dumps(details) if details else None
|
||
|
|
|
||
|
|
query = """
|
||
|
|
INSERT INTO tmodule_sync_log
|
||
|
|
(event_type, entity_type, entity_id, user_id, details, ip_address, user_agent)
|
||
|
|
VALUES (%s, %s, %s, %s, %s::jsonb, %s, %s)
|
||
|
|
RETURNING id
|
||
|
|
"""
|
||
|
|
|
||
|
|
log_id = execute_insert(
|
||
|
|
query,
|
||
|
|
(event_type, entity_type, entity_id, user_id, details_json, ip_address, user_agent)
|
||
|
|
)
|
||
|
|
|
||
|
|
logger.debug(f"📝 Logged event: {event_type} (ID: {log_id})")
|
||
|
|
return log_id
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"❌ Failed to log event {event_type}: {e}")
|
||
|
|
# Don't raise - audit logging failure shouldn't break main flow
|
||
|
|
return 0
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_sync_started(user_id: Optional[int] = None) -> int:
|
||
|
|
"""Log start af vTiger sync"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="sync_started",
|
||
|
|
user_id=user_id,
|
||
|
|
details={"timestamp": datetime.now().isoformat()}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_sync_completed(
|
||
|
|
stats: Dict[str, Any],
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log succesfuld sync med statistik"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="sync_completed",
|
||
|
|
user_id=user_id,
|
||
|
|
details=stats
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_sync_failed(
|
||
|
|
error: str,
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log fejlet sync"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="sync_failed",
|
||
|
|
user_id=user_id,
|
||
|
|
details={"error": error, "timestamp": datetime.now().isoformat()}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_approval(
|
||
|
|
time_id: int,
|
||
|
|
original_hours: float,
|
||
|
|
approved_hours: float,
|
||
|
|
rounded_to: Optional[float],
|
||
|
|
note: Optional[str],
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log godkendelse af tidsregistrering"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="approval",
|
||
|
|
entity_type="time_entry",
|
||
|
|
entity_id=time_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={
|
||
|
|
"original_hours": original_hours,
|
||
|
|
"approved_hours": approved_hours,
|
||
|
|
"rounded_to": rounded_to,
|
||
|
|
"note": note,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_rejection(
|
||
|
|
time_id: int,
|
||
|
|
reason: Optional[str],
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log afvisning af tidsregistrering"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="rejection",
|
||
|
|
entity_type="time_entry",
|
||
|
|
entity_id=time_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={
|
||
|
|
"reason": reason,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_order_created(
|
||
|
|
order_id: int,
|
||
|
|
customer_id: int,
|
||
|
|
total_hours: float,
|
||
|
|
total_amount: float,
|
||
|
|
line_count: int,
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log oprettelse af ordre"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="order_created",
|
||
|
|
entity_type="order",
|
||
|
|
entity_id=order_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={
|
||
|
|
"customer_id": customer_id,
|
||
|
|
"total_hours": total_hours,
|
||
|
|
"total_amount": total_amount,
|
||
|
|
"line_count": line_count,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_order_updated(
|
||
|
|
order_id: int,
|
||
|
|
changes: Dict[str, Any],
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log opdatering af ordre"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="order_updated",
|
||
|
|
entity_type="order",
|
||
|
|
entity_id=order_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={
|
||
|
|
"changes": changes,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_order_cancelled(
|
||
|
|
order_id: int,
|
||
|
|
reason: Optional[str],
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log annullering af ordre"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="order_cancelled",
|
||
|
|
entity_type="order",
|
||
|
|
entity_id=order_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={
|
||
|
|
"reason": reason,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_export_started(
|
||
|
|
order_id: int,
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log start af e-conomic export"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="export_started",
|
||
|
|
entity_type="order",
|
||
|
|
entity_id=order_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={"timestamp": datetime.now().isoformat()}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_export_completed(
|
||
|
|
order_id: int,
|
||
|
|
economic_draft_id: Optional[int],
|
||
|
|
economic_order_number: Optional[str],
|
||
|
|
dry_run: bool,
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log succesfuld export til e-conomic"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="export_completed",
|
||
|
|
entity_type="order",
|
||
|
|
entity_id=order_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={
|
||
|
|
"economic_draft_id": economic_draft_id,
|
||
|
|
"economic_order_number": economic_order_number,
|
||
|
|
"dry_run": dry_run,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_export_failed(
|
||
|
|
order_id: int,
|
||
|
|
error: str,
|
||
|
|
user_id: Optional[int] = None
|
||
|
|
) -> int:
|
||
|
|
"""Log fejlet export til e-conomic"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="export_failed",
|
||
|
|
entity_type="order",
|
||
|
|
entity_id=order_id,
|
||
|
|
user_id=user_id,
|
||
|
|
details={
|
||
|
|
"error": error,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def log_module_uninstalled(user_id: Optional[int] = None) -> int:
|
||
|
|
"""Log modul-uninstall"""
|
||
|
|
return AuditLogger.log_event(
|
||
|
|
event_type="module_uninstalled",
|
||
|
|
user_id=user_id,
|
||
|
|
details={"timestamp": datetime.now().isoformat()}
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# Singleton instance
|
||
|
|
audit = AuditLogger()
|