from __future__ import annotations import csv import io import json import uuid from collections.abc import Sequence from datetime import UTC, datetime, timedelta from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import Response from sqlalchemy import func, select from sqlalchemy.orm import Session from app.api.deps import get_db, require_operations_manager from app.core.config import get_settings from app.models.audit import AuditEvent from app.models.booking import Booking from app.models.customer import Customer from app.models.data_quality import DataQualityIssue from app.models.vehicle import Vehicle from app.schemas import AuditEventOut, AuditEventPageOut, CurrentUser from app.services.audit import record_audit_event router = APIRouter(prefix="/api/v1/audit", tags=["audit"]) settings = get_settings() # Only entity types with a stable public reference and (optionally) a real frontend route # are resolved here. Types like "system", "knowledge" or "mcp_tool" carry no linkable # entity_id and are left as plain labels. _ENTITY_MODELS: dict[str, Any] = { "vehicle": Vehicle, "booking": Booking, "customer": Customer, "data_quality_issue": DataQualityIssue, } _ROUTE_TEMPLATES: dict[str, str] = { "vehicle": "/vehicles/{ref}", "booking": "/bookings/{ref}", "data_quality_issue": "/data-quality/{ref}", # No customer detail route exists in this proof of concept; still resolve the # reference for display, just without a link. } def _as_utc(value: datetime | None) -> datetime | None: """Treat naive query datetimes as UTC so they compare safely with aware values.""" if value is None: return None return value if value.tzinfo is not None else value.replace(tzinfo=UTC) @router.get("/export.csv") def export_audit_csv( occurred_from: datetime | None = Query(default=None), occurred_to: datetime | None = Query(default=None), db: Session = Depends(get_db), actor: CurrentUser = Depends(require_operations_manager), ) -> Response: end = _as_utc(occurred_to) or datetime.now(UTC) start = _as_utc(occurred_from) or end - timedelta(days=30) if end <= start or end - start > timedelta(days=90): raise HTTPException(status_code=422, detail="Audit export range must be 1 to 90 days") events = db.scalars( select(AuditEvent) .where(AuditEvent.occurred_at >= start, AuditEvent.occurred_at <= end) .order_by(AuditEvent.occurred_at) .limit(settings.privacy_audit_export_max_rows + 1) ).all() if len(events) > settings.privacy_audit_export_max_rows: raise HTTPException(status_code=413, detail="Audit export exceeds configured row limit") output = io.StringIO(newline="") writer = csv.writer(output) writer.writerow( ( "id", "occurred_at", "actor_type", "actor_label", "action", "entity_type", "entity_id", "correlation_id", "before", "after", "metadata", ) ) for event in events: writer.writerow( ( event.id, event.occurred_at.isoformat(), event.actor_type, event.actor_label, event.action, event.entity_type, event.entity_id or "", event.correlation_id, json.dumps(event.before_json, separators=(",", ":"), default=str), json.dumps(event.after_json, separators=(",", ":"), default=str), json.dumps(event.metadata_json, separators=(",", ":"), default=str), ) ) record_audit_event( db, actor_type="user", actor_label=actor.display_name, action="audit_exported", entity_type="audit", metadata={"from": start.isoformat(), "to": end.isoformat(), "rows": len(events)}, ) db.commit() return Response( output.getvalue(), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="mobilityops-audit.csv"'}, ) def _resolve_entity_refs(db: Session, events: Sequence[AuditEvent]) -> dict[uuid.UUID, str]: ids_by_type: dict[str, set[uuid.UUID]] = {} for event in events: if event.entity_id is not None and event.entity_type in _ENTITY_MODELS: ids_by_type.setdefault(event.entity_type, set()).add(event.entity_id) refs: dict[uuid.UUID, str] = {} for entity_type, ids in ids_by_type.items(): model = _ENTITY_MODELS[entity_type] rows: Sequence[Any] = db.scalars(select(model).where(model.id.in_(ids))).all() for row in rows: refs[row.id] = row.public_ref return refs @router.get("", response_model=list[AuditEventOut] | AuditEventPageOut) def list_audit_events( actor_label: str | None = Query(default=None), action: str | None = Query(default=None), entity_type: str | None = Query(default=None), entity_ref: str | None = Query(default=None, min_length=1, max_length=100), correlation_id: uuid.UUID | None = Query(default=None), occurred_from: datetime | None = Query(default=None), occurred_to: datetime | None = Query(default=None), page: int | None = Query(default=None, ge=1), page_size: int = Query(default=25, ge=1, le=25), db: Session = Depends(get_db), _user: CurrentUser = Depends(require_operations_manager), ) -> list[AuditEventOut] | AuditEventPageOut: stmt = select(AuditEvent).order_by(AuditEvent.occurred_at.desc()) if actor_label: stmt = stmt.where(AuditEvent.actor_label == actor_label) if action: stmt = stmt.where(AuditEvent.action == action) if entity_type: stmt = stmt.where(AuditEvent.entity_type == entity_type) if entity_ref: matched_ids: set[uuid.UUID] = set() for model in _ENTITY_MODELS.values(): matched_ids.update( db.scalars( select(model.id).where(model.public_ref.ilike(f"%{entity_ref.strip()}%")) ).all() ) if not matched_ids: if page is None: return [] return AuditEventPageOut(items=[], page=1, page_size=page_size, total=0, total_pages=1) stmt = stmt.where(AuditEvent.entity_id.in_(matched_ids)) if correlation_id: stmt = stmt.where(AuditEvent.correlation_id == correlation_id) if occurred_from: stmt = stmt.where(AuditEvent.occurred_at >= occurred_from) if occurred_to: stmt = stmt.where(AuditEvent.occurred_at <= occurred_to) total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0 page_number = page or 1 events = db.scalars( stmt if page is None else stmt.offset((page_number - 1) * page_size).limit(page_size) ).all() entity_refs = _resolve_entity_refs(db, events) out = [] for e in events: ref = entity_refs.get(e.entity_id) if e.entity_id else None route = _ROUTE_TEMPLATES.get(e.entity_type) out.append( AuditEventOut( id=str(e.id), actor_type=e.actor_type, actor_label=e.actor_label, action=e.action, entity_type=e.entity_type, entity_id=str(e.entity_id) if e.entity_id else None, entity_ref=ref, entity_link=route.format(ref=ref) if route and ref else None, correlation_id=str(e.correlation_id), occurred_at=e.occurred_at, before=e.before_json, after=e.after_json, metadata=e.metadata_json, ) ) if page is None: return out total_pages = max(1, (total + page_size - 1) // page_size) return AuditEventPageOut( items=out, page=min(page_number, total_pages), page_size=page_size, total=total, total_pages=total_pages, )