MobilityOps/backend/app/seed_loader.py

from __future__ import annotations

import csv
import uuid
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from difflib import SequenceMatcher
from pathlib import Path

from sqlalchemy import delete, insert, update
from sqlalchemy.orm import Session

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.idempotency import IdempotencyRecord
from app.models.inspection import Inspection
from app.models.maintenance import MaintenanceRecord
from app.models.outbox import OutboxEvent
from app.models.user import User
from app.models.vehicle import Vehicle
from app.services.audit import record_audit_event

settings = get_settings()

DEMO_USERS = [
    {
        "public_ref": "USR-OPS",
        "display_name": "Amelie De Ridder",
        "role": "operations_manager",
    },
    {
        "public_ref": "USR-EMP",
        "display_name": "Karim Boujaddaine",
        "role": "rental_employee",
    },
]

# seed/generate_seed.py authored the committed CSVs relative to this fixed date
# (`--anchor 2026-08-01`, matching Settings.demo_today). Every reset shifts every
# seeded date by (today - SEED_AUTHORED_ANCHOR) so "today" / "near-future" / "overlaps
# right now" scenarios stay true to the actual reset moment instead of decaying as real
# time passes between resets -- a fixed anchor with no shift goes stale within days.
SEED_AUTHORED_ANCHOR = date(2026, 8, 1)


def _seed_anchor_shift(today: date) -> timedelta:
    return today - SEED_AUTHORED_ANCHOR


def _parse_dt(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def _parse_bool(value: str) -> bool:
    return value.strip().lower() == "true"


def _parse_optional_int(value: str) -> int | None:
    value = value.strip()
    return int(value) if value else None


@dataclass
class SeedResult:
    counts: dict[str, int]
    anchor_date: date
    seeded_at: datetime


def _seed_dir() -> Path:
    return Path(settings.seed_dir)


def _read_csv(name: str) -> list[dict[str, str]]:
    path = _seed_dir() / name
    with path.open(newline="", encoding="utf-8") as handle:
        return list(csv.DictReader(handle))


def clear_all(db: Session) -> None:
    for model in (
        AuditEvent,
        OutboxEvent,
        IdempotencyRecord,
        DataQualityIssue,
        Inspection,
        MaintenanceRecord,
        Booking,
        Vehicle,
        Customer,
        User,
    ):
        db.execute(delete(model))


def load_seed(db: Session) -> SeedResult:
    counts: dict[str, int] = {}
    today = datetime.now(UTC).date()
    shift = _seed_anchor_shift(today)

    user_rows = [
        {"id": uuid.uuid4(), **user, "active": True} for user in DEMO_USERS
    ]
    db.execute(insert(User), user_rows)
    counts["users"] = len(user_rows)

    customer_id_by_ref: dict[str, uuid.UUID] = {}
    customer_rows = []
    customer_row_by_ref: dict[str, dict] = {}
    for row in _read_csv("customers.csv"):
        cid = uuid.uuid4()
        customer_id_by_ref[row["public_ref"]] = cid
        customer_row = {
            "id": cid,
            "public_ref": row["public_ref"],
            "first_name": row["first_name"],
            "last_name": row["last_name"],
            "email": row["email"] or None,
            "phone": row["phone"] or None,
            "postal_code": row["postal_code"] or None,
            "city": row["city"] or None,
        }
        customer_rows.append(customer_row)
        customer_row_by_ref[row["public_ref"]] = customer_row
    db.execute(insert(Customer), customer_rows)
    counts["customers"] = len(customer_rows)
    # Second pass for merged_into (self-referencing FK) since target must exist first.
    for row in _read_csv("customers.csv"):
        merged_ref = row.get("merged_into") or ""
        if merged_ref:
            db.execute(
                update(Customer)
                .where(Customer.id == customer_id_by_ref[row["public_ref"]])
                .values(merged_into_customer_id=customer_id_by_ref[merged_ref])
            )

    vehicle_id_by_ref: dict[str, uuid.UUID] = {}
    vehicle_rows = []
    for row in _read_csv("vehicles.csv"):
        vid = uuid.uuid4()
        vehicle_id_by_ref[row["public_ref"]] = vid
        vehicle_rows.append(
            {
                "id": vid,
                "public_ref": row["public_ref"],
                "make": row["make"],
                "model": row["model"],
                "model_year": int(row["model_year"]),
                "registration_number": row["registration_number"],
                "location": row["location"],
                "operational_status": row["operational_status"],
                "odometer_km": int(row["odometer_km"]),
                "next_service_km": int(row["next_service_km"]),
                "active": _parse_bool(row["active"]),
                "version": 1,
            }
        )
    db.execute(insert(Vehicle), vehicle_rows)
    counts["vehicles"] = len(vehicle_rows)

    booking_id_by_ref: dict[str, uuid.UUID] = {}
    booking_rows = []
    for row in _read_csv("bookings.csv"):
        bid = uuid.uuid4()
        booking_id_by_ref[row["public_ref"]] = bid
        booking_rows.append(
            {
                "id": bid,
                "public_ref": row["public_ref"],
                "customer_id": customer_id_by_ref[row["customer_ref"]],
                "vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
                "starts_at": _parse_dt(row["starts_at"]) + shift,
                "ends_at": _parse_dt(row["ends_at"]) + shift,
                "status": row["status"],
                "start_odometer_km": _parse_optional_int(row["start_odometer_km"]),
                "end_odometer_km": _parse_optional_int(row["end_odometer_km"]),
                "requirements_complete": _parse_bool(row["requirements_complete"]),
            }
        )
    db.execute(insert(Booking), booking_rows)
    counts["bookings"] = len(booking_rows)

    inspection_rows = []
    for row in _read_csv("inspections.csv"):
        inspection_rows.append(
            {
                "id": uuid.uuid4(),
                "public_ref": row["public_ref"],
                "booking_id": booking_id_by_ref[row["booking_ref"]],
                "vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
                "type": row["type"],
                "fuel_level_percent": int(row["fuel_level_percent"]),
                "cleanliness_ok": _parse_bool(row["cleanliness_ok"]),
                "damage_reported": _parse_bool(row["damage_reported"]),
                "technical_warning": _parse_bool(row["technical_warning"]),
                "odometer_km": int(row["odometer_km"]),
                "completed_at": _parse_dt(row["completed_at"]) + shift,
                "completed_by": None,
            }
        )
    db.execute(insert(Inspection), inspection_rows)
    counts["inspections"] = len(inspection_rows)

    maintenance_rows = []
    for row in _read_csv("maintenance.csv"):
        maintenance_rows.append(
            {
                "id": uuid.uuid4(),
                "public_ref": row["public_ref"],
                "vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
                "occurred_at": _parse_dt(row["occurred_at"]) + shift,
                "odometer_km": int(row["odometer_km"]),
                "category": row["category"],
                "summary": row["summary"],
            }
        )
    db.execute(insert(MaintenanceRecord), maintenance_rows)
    counts["maintenance"] = len(maintenance_rows)

    def resolve_entity(entity_ref: str) -> tuple[str, uuid.UUID]:
        if entity_ref.startswith("CUS-"):
            return "customer", customer_id_by_ref[entity_ref]
        return "vehicle", vehicle_id_by_ref[entity_ref]

    def _seed_signals(public_ref: str, entity_ref: str, related_refs: list[str]) -> list[dict]:
        # The four named DQ-DEMO-* rows anchor the guided demo's scripted scenarios, so
        # they carry real, accurate structured signals (not just a legacy English
        # sentence) -- the frontend renders these as the primary, localized evidence;
        # see docs/fleet-ops-correction/current-gap-audit.md §6.
        if public_ref == "DQ-DEMO-DUPLICATE":
            a = customer_row_by_ref[entity_ref]
            b = customer_row_by_ref[related_refs[0]]
            name_a = f"{a['first_name']} {a['last_name']}".strip().lower()
            name_b = f"{b['first_name']} {b['last_name']}".strip().lower()
            ratio = SequenceMatcher(None, name_a, name_b).ratio()
            return [
                {"code": "duplicate.exact_email"},
                {"code": "duplicate.exact_phone"},
                {"code": "duplicate.same_postal_code"},
                {"code": "duplicate.similar_name", "params": {"score": round(ratio, 2)}},
            ]
        if public_ref == "DQ-DEMO-OVERLAP":
            return [{"code": "overlap.reserved_bookings", "params": {"refs": related_refs}}]
        if public_ref == "DQ-DEMO-STATUS":
            return [{"code": "vehicle.booking_conflict"}]
        if public_ref == "DQ-DEMO-ATTENTION":
            return [
                {
                    "code": "attention.upcoming_booking_missing_inspection",
                    "params": {"booking_ref": related_refs[0] if related_refs else ""},
                }
            ]
        return []

    dq_rows = []
    now = datetime.now(UTC)
    for row in _read_csv("data_quality_issues.csv"):
        entity_type, entity_id = resolve_entity(row["entity_ref"])
        related_ref = row.get("related_ref") or ""
        related_refs = related_ref.split("|") if related_ref else []
        dq_rows.append(
            {
                "id": uuid.uuid4(),
                "public_ref": row["public_ref"],
                "rule_type": row["rule_type"],
                "entity_type": entity_type,
                "entity_id": entity_id,
                "severity": row["severity"],
                "status": row["status"],
                "evidence_json": {
                    "summary": row["evidence"],
                    "entity_ref": row["entity_ref"],
                    "related_refs": related_refs,
                    "signals": _seed_signals(row["public_ref"], row["entity_ref"], related_refs),
                },
                "proposed_action_json": {},
                "detected_at": now,
                "resolved_at": now if row["status"] == "resolved" else None,
                "resolved_by": "USR-OPS" if row["status"] == "resolved" else None,
            }
        )
    db.execute(insert(DataQualityIssue), dq_rows)
    counts["data_quality_issues"] = len(dq_rows)

    vehicle_ref_by_booking_ref = {
        row["public_ref"]: row["vehicle_ref"] for row in _read_csv("bookings.csv")
    }

    outbox_rows = []
    for row in _read_csv("workflow_runs.csv"):
        booking_id = booking_id_by_ref.get(row["aggregate_ref"])
        # Build the same schema-complete envelope the live return workflow (M2) produces,
        # so a seeded/historical event is redeliverable (e.g. via manual retry) without the
        # dispatcher crashing on a missing key. See PROJECT_STATE.md M4 notes.
        outbox_rows.append(
            {
                "event_id": uuid.UUID(row["event_id"]),
                "event_type": row["event_type"],
                "aggregate_type": "booking",
                "aggregate_id": booking_id or uuid.uuid4(),
                "payload_json": {
                    "correlation_id": str(uuid.uuid4()),
                    "aggregate": {
                        "type": "booking",
                        "id": str(booking_id or uuid.uuid4()),
                        "public_ref": row["aggregate_ref"],
                    },
                    "data": {
                        "vehicle_ref": vehicle_ref_by_booking_ref.get(row["aggregate_ref"], ""),
                        "inspection_ref": "",
                        "resulting_vehicle_status": "cleaning",
                        "attention_reasons": [],
                    },
                    "aggregate_ref": row["aggregate_ref"],
                },
                "occurred_at": _parse_dt(row["occurred_at"]) + shift,
                "delivery_status": row["status"],
                "attempts": int(row["attempts"]),
                "next_attempt_at": None,
                "last_error": row["last_error"] or None,
                # The seed dataset's one synthetic failure (BK-H-0020) models a
                # connection-timeout-style delivery failure -- see workflow_runs.csv.
                "last_error_code": "connectionError" if row["last_error"] else None,
                "external_run_id": None,
            }
        )
    db.execute(insert(OutboxEvent), outbox_rows)
    counts["workflow_runs"] = len(outbox_rows)

    seeded_at = datetime.now(UTC)
    record_audit_event(
        db,
        actor_type="system",
        actor_label="seed loader",
        action="demo_data_seeded",
        entity_type="system",
        metadata={
            "anchor_date": today.isoformat(),
            "seed_authored_anchor": SEED_AUTHORED_ANCHOR.isoformat(),
            "counts": counts,
        },
    )

    return SeedResult(counts=counts, anchor_date=today, seeded_at=seeded_at)


def reset_and_seed(db: Session) -> SeedResult:
    from app.services.data_quality import run_scan

    clear_all(db)
    result = load_seed(db)
    db.commit()
    scan = run_scan(db)
    result.counts["data_quality_issues"] += sum(scan.created.values())
    return result