215 lines
7.9 KiB
Python
215 lines
7.9 KiB
Python
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,
|
|
)
|