501 lines
20 KiB
Python
501 lines
20 KiB
Python
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.core.db import begin_exclusive_demo_reset, end_exclusive_demo_reset
|
|
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 DEMO_SCENARIO_ERROR_CODE, 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))
|
|
|
|
|
|
_PERSISTENT_TELEMETRY_ACTIONS = (
|
|
"mcp_tool_request",
|
|
"n8n_return_followup_recorded",
|
|
"n8n_workflow_failure_registered",
|
|
"n8n_procedures_synced",
|
|
"n8n_workflow_heartbeat",
|
|
"knowledge_question_asked",
|
|
)
|
|
|
|
|
|
def clear_all(db: Session, *, preserve_integration_telemetry: bool = False) -> None:
|
|
# RevokedSession is intentionally NOT cleared: it has no FK to users and wiping it
|
|
# would silently re-validate cookies that were logged out before the reset.
|
|
for model in (
|
|
OutboxEvent,
|
|
IdempotencyRecord,
|
|
DataQualityIssue,
|
|
Inspection,
|
|
MaintenanceRecord,
|
|
Booking,
|
|
Vehicle,
|
|
Customer,
|
|
User,
|
|
):
|
|
db.execute(delete(model))
|
|
if preserve_integration_telemetry:
|
|
db.execute(
|
|
delete(AuditEvent).where(AuditEvent.action.not_in(_PERSISTENT_TELEMETRY_ACTIONS))
|
|
)
|
|
else:
|
|
db.execute(delete(AuditEvent))
|
|
|
|
|
|
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 = []
|
|
vehicle_row_by_ref: dict[str, dict] = {}
|
|
for row in _read_csv("vehicles.csv"):
|
|
vid = uuid.uuid4()
|
|
vehicle_id_by_ref[row["public_ref"]] = vid
|
|
vehicle_row = {
|
|
"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,
|
|
}
|
|
vehicle_rows.append(vehicle_row)
|
|
vehicle_row_by_ref[row["public_ref"]] = vehicle_row
|
|
db.execute(insert(Vehicle), vehicle_rows)
|
|
counts["vehicles"] = len(vehicle_rows)
|
|
|
|
booking_id_by_ref: dict[str, uuid.UUID] = {}
|
|
booking_rows = []
|
|
booking_row_by_ref: dict[str, dict] = {}
|
|
for row in _read_csv("bookings.csv"):
|
|
bid = uuid.uuid4()
|
|
booking_id_by_ref[row["public_ref"]] = bid
|
|
booking_row = {
|
|
"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"]),
|
|
}
|
|
booking_rows.append(booking_row)
|
|
booking_row_by_ref[row["public_ref"]] = booking_row
|
|
db.execute(insert(Booking), booking_rows)
|
|
counts["bookings"] = len(booking_rows)
|
|
|
|
inspection_rows = []
|
|
inspection_source_rows = _read_csv("inspections.csv")
|
|
return_inspection_row_by_booking_ref = {
|
|
row["booking_ref"]: row for row in inspection_source_rows if row["type"] == "return"
|
|
}
|
|
for row in inspection_source_rows:
|
|
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 _vehicle_conflict_facts(vehicle_ref: str, *, service_threshold_reached: bool) -> dict:
|
|
# Mirrors app.services.vehicle_status.VehicleStatusFacts.as_dict() for the
|
|
# handful of seed-only rows below -- none of them carry an active rental or a
|
|
# real booking conflict (verified against the fixed seed dataset), only a
|
|
# genuinely-crossed service threshold or none at all, so those two fields are
|
|
# the only ones that vary per vehicle.
|
|
vehicle = vehicle_row_by_ref[vehicle_ref]
|
|
return {
|
|
"active_booking_refs": [],
|
|
"overlapping_booking_pairs": [],
|
|
"service_threshold_reached": service_threshold_reached,
|
|
"odometer_km": vehicle["odometer_km"],
|
|
"next_service_km": vehicle["next_service_km"],
|
|
"open_booking_overlap_issue_ref": None,
|
|
}
|
|
|
|
def _odometer_regression_signal(later_ref: str, earlier_ref: str) -> list[dict]:
|
|
later = return_inspection_row_by_booking_ref[later_ref]
|
|
earlier = return_inspection_row_by_booking_ref[earlier_ref]
|
|
return [
|
|
{
|
|
"code": "odometer.regression",
|
|
"source_type": "return",
|
|
"params": {
|
|
"later_ref": later["public_ref"],
|
|
"later_km": int(later["odometer_km"]),
|
|
"earlier_ref": earlier["public_ref"],
|
|
"earlier_km": int(earlier["odometer_km"]),
|
|
"booking_ref": later_ref,
|
|
},
|
|
}
|
|
]
|
|
|
|
def _missing_field_signal(field: str) -> list[dict]:
|
|
return [{"code": "missing_field", "params": {"field": field}}]
|
|
|
|
# Every seed-only row below (i.e. not one of the four named DQ-DEMO-* scenarios)
|
|
# used to carry no structured signal at all -- just the placeholder summary
|
|
# "Synthetic deterministic seed issue". Each now cites a real fact about its actual
|
|
# entity (a genuinely-crossed service threshold, a genuinely-blank field, or a real
|
|
# pair of booking odometer readings engineered into seed/bookings.csv), using the
|
|
# exact same signal vocabulary the live scan (app.services.data_quality) already
|
|
# renders through -- see docs/fleet-ops-correction/current-gap-audit.md §6.
|
|
_SEED_SIGNALS_BY_REF: dict[str, list[dict]] = {
|
|
"DQ-0005": [
|
|
{
|
|
"code": "vehicle.service_threshold_reached",
|
|
"params": _vehicle_conflict_facts("MO-036", service_threshold_reached=True),
|
|
}
|
|
],
|
|
"DQ-0006": _missing_field_signal("location"),
|
|
"DQ-0007": _odometer_regression_signal("BK-H-0007", "BK-H-0057"),
|
|
"DQ-0008": [
|
|
{
|
|
"code": "vehicle.rental_ended",
|
|
"params": _vehicle_conflict_facts("MO-007", service_threshold_reached=False),
|
|
}
|
|
],
|
|
"DQ-0009": _missing_field_signal("location"),
|
|
"DQ-0010": _odometer_regression_signal("BK-H-0010", "BK-H-0060"),
|
|
"DQ-0011": [
|
|
{
|
|
"code": "vehicle.service_threshold_reached",
|
|
"params": _vehicle_conflict_facts("MO-028", service_threshold_reached=True),
|
|
}
|
|
],
|
|
"DQ-0012": _missing_field_signal("registration_number"),
|
|
"DQ-0013": _missing_field_signal("location"),
|
|
"DQ-0014": _missing_field_signal("registration_number"),
|
|
"DQ-0015": _missing_field_signal("location"),
|
|
"DQ-0016": _missing_field_signal("location"),
|
|
"DQ-0017": _missing_field_signal("location"),
|
|
"DQ-0018": _missing_field_signal("registration_number"),
|
|
"DQ-0019": _missing_field_signal("location"),
|
|
"DQ-0020": _missing_field_signal("location"),
|
|
"DQ-0021": _missing_field_signal("location"),
|
|
}
|
|
|
|
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 _SEED_SIGNALS_BY_REF.get(public_ref, [])
|
|
|
|
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 []
|
|
signals = _seed_signals(row["public_ref"], row["entity_ref"], related_refs)
|
|
evidence_extra: dict[str, object] = {}
|
|
if row["rule_type"] == "odometer_regression" and signals:
|
|
params = signals[0].get("params", {})
|
|
later_ref = params.get("later_ref")
|
|
earlier_ref = params.get("earlier_ref")
|
|
booking_ref = params.get("booking_ref")
|
|
if (
|
|
isinstance(earlier_ref, str)
|
|
and isinstance(later_ref, str)
|
|
and isinstance(booking_ref, str)
|
|
):
|
|
# The authored CSV prose predates structured references. Preserve the
|
|
# two real source bookings so the detail page can show evidence, while
|
|
# only the later (regressing) reading is eligible for correction.
|
|
related_refs = [earlier_ref, booking_ref, later_ref]
|
|
evidence_extra = {
|
|
"source_type": signals[0].get("source_type", "return"),
|
|
"source_types": [signals[0].get("source_type", "return")],
|
|
"correctable_booking_refs": [booking_ref],
|
|
}
|
|
severity_due_delta = {
|
|
"high": timedelta(hours=4),
|
|
"medium": timedelta(days=1),
|
|
"low": timedelta(days=3),
|
|
}.get(row["severity"], timedelta(days=1))
|
|
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": signals,
|
|
**evidence_extra,
|
|
},
|
|
"proposed_action_json": {},
|
|
"detected_at": now,
|
|
"due_at": now + severity_due_delta if row["status"] == "open" else None,
|
|
"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.
|
|
# It is coded as a *prepared demo scenario*, not as a real
|
|
# connectionError, so integration health never degrades because of a
|
|
# prop and a viewer is told plainly that this failure is staged.
|
|
"last_error_code": DEMO_SCENARIO_ERROR_CODE 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,
|
|
*,
|
|
preserve_integration_telemetry: bool = False,
|
|
commit: bool = True,
|
|
) -> SeedResult:
|
|
from app.services.data_quality import run_scan
|
|
|
|
if not db.info.get("mobilityops_demo_reset_exclusive"):
|
|
begin_exclusive_demo_reset(db)
|
|
clear_all(db, preserve_integration_telemetry=preserve_integration_telemetry)
|
|
result = load_seed(db)
|
|
scan = run_scan(db, commit=False)
|
|
result.counts["data_quality_issues"] += sum(scan.created.values())
|
|
if commit:
|
|
db.commit()
|
|
end_exclusive_demo_reset(db)
|
|
else:
|
|
db.flush()
|
|
return result
|