from __future__ import annotations import uuid from dataclasses import dataclass, field from datetime import UTC, datetime from difflib import SequenceMatcher from sqlalchemy import select from sqlalchemy.orm import Session from app.core.errors import AppError 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 CurrentUser from app.services.audit import record_audit_event REQUIRED_CUSTOMER_FIELDS = ("first_name", "last_name") REQUIRED_VEHICLE_FIELDS = ("registration_number", "make", "model", "location") DUPLICATE_THRESHOLD = 70 @dataclass class ScanResult: created: dict[str, int] = field(default_factory=dict) def bump(self, rule_type: str) -> None: self.created[rule_type] = self.created.get(rule_type, 0) + 1 def _normalize(value: str | None) -> str: return (value or "").strip().lower() def _has_open_issue(db: Session, rule_type: str, entity_type: str, entity_id: uuid.UUID) -> bool: return ( db.scalar( select(DataQualityIssue.id).where( DataQualityIssue.rule_type == rule_type, DataQualityIssue.entity_type == entity_type, DataQualityIssue.entity_id == entity_id, DataQualityIssue.status == "open", ) ) is not None ) def _next_public_ref(db: Session, prefix: str) -> str: existing = db.execute(select(DataQualityIssue.public_ref)).scalars().all() numbers = [ int(ref.rsplit("-", 1)[-1]) for ref in existing if ref.startswith(f"{prefix}-") and ref.rsplit("-", 1)[-1].isdigit() ] next_number = (max(numbers) + 1) if numbers else 1 return f"{prefix}-{next_number:04d}" def _open_issue( db: Session, scan: ScanResult, *, rule_type: str, entity_type: str, entity_id: uuid.UUID, severity: str, summary: str, entity_ref: str, related_refs: list[str], ) -> None: if _has_open_issue(db, rule_type, entity_type, entity_id): return now = datetime.now(UTC) issue = DataQualityIssue( public_ref=_next_public_ref(db, "DQ-SCAN"), rule_type=rule_type, entity_type=entity_type, entity_id=entity_id, severity=severity, status="open", evidence_json={ "summary": summary, "entity_ref": entity_ref, "related_refs": related_refs, }, proposed_action_json={}, detected_at=now, ) db.add(issue) db.flush() scan.bump(rule_type) def _scan_duplicate_customers(db: Session, scan: ScanResult) -> None: customers = db.scalars( select(Customer).where(Customer.merged_into_customer_id.is_(None)) ).all() customers.sort(key=lambda c: c.public_ref) for i, a in enumerate(customers): for b in customers[i + 1 :]: score = 0 signals = [] if _normalize(a.email) and _normalize(a.email) == _normalize(b.email): score += 60 signals.append("exact email") if _normalize(a.phone) and _normalize(a.phone) == _normalize(b.phone): score += 50 signals.append("exact phone") if _normalize(a.postal_code) and _normalize(a.postal_code) == _normalize(b.postal_code): score += 10 signals.append("exact postal code") name_a = f"{_normalize(a.first_name)} {_normalize(a.last_name)}" name_b = f"{_normalize(b.first_name)} {_normalize(b.last_name)}" ratio = SequenceMatcher(None, name_a, name_b).ratio() if ratio >= 0.5: score += round(ratio * 30) signals.append("similar name") if score >= DUPLICATE_THRESHOLD: _open_issue( db, scan, rule_type="possible_duplicate_customer", entity_type="customer", entity_id=a.id, severity="high", summary="; ".join(signals) + f" (score {score})", entity_ref=a.public_ref, related_refs=[b.public_ref], ) def _scan_missing_required_fields(db: Session, scan: ScanResult) -> None: for customer in db.scalars( select(Customer).where(Customer.merged_into_customer_id.is_(None)) ).all(): missing = [f for f in REQUIRED_CUSTOMER_FIELDS if not getattr(customer, f)] if not customer.email and not customer.phone: missing.append("email_or_phone") if missing: _open_issue( db, scan, rule_type="missing_required_field", entity_type="customer", entity_id=customer.id, severity="low", summary=f"Missing: {', '.join(missing)}", entity_ref=customer.public_ref, related_refs=[], ) for vehicle in db.scalars(select(Vehicle).where(Vehicle.active.is_(True))).all(): missing = [f for f in REQUIRED_VEHICLE_FIELDS if not getattr(vehicle, f)] if missing: _open_issue( db, scan, rule_type="missing_required_field", entity_type="vehicle", entity_id=vehicle.id, severity="low", summary=f"Missing: {', '.join(missing)}", entity_ref=vehicle.public_ref, related_refs=[], ) def _scan_booking_overlaps(db: Session, scan: ScanResult) -> None: vehicles = db.scalars(select(Vehicle)).all() bookings_by_vehicle: dict[uuid.UUID, list[Booking]] = {} for booking in db.scalars( select(Booking).where(Booking.status.in_(["reserved", "active"])) ).all(): bookings_by_vehicle.setdefault(booking.vehicle_id, []).append(booking) vehicle_by_id = {v.id: v for v in vehicles} for vehicle_id, bookings in bookings_by_vehicle.items(): bookings.sort(key=lambda b: b.starts_at) for i, first in enumerate(bookings): for second in bookings[i + 1 :]: if second.starts_at < first.ends_at and first.starts_at < second.ends_at: vehicle = vehicle_by_id[vehicle_id] _open_issue( db, scan, rule_type="booking_overlap", entity_type="vehicle", entity_id=vehicle_id, severity="high", summary=f"Overlapping bookings {first.public_ref} and {second.public_ref}", entity_ref=vehicle.public_ref, related_refs=[first.public_ref, second.public_ref], ) def _scan_vehicle_status_conflicts(db: Session, scan: ScanResult) -> None: vehicles = db.scalars(select(Vehicle)).all() active_by_vehicle: dict[uuid.UUID, list[Booking]] = {} for booking in db.scalars(select(Booking).where(Booking.status == "active")).all(): active_by_vehicle.setdefault(booking.vehicle_id, []).append(booking) open_high_by_vehicle = { row[0] for row in db.execute( select(DataQualityIssue.entity_id).where( DataQualityIssue.entity_type == "vehicle", DataQualityIssue.status == "open", DataQualityIssue.severity == "high", ) ).all() } for vehicle in vehicles: has_active_booking = vehicle.id in active_by_vehicle reason = None if vehicle.operational_status == "available" and has_active_booking: reason = "marked available while an active booking exists" elif vehicle.operational_status == "rented" and not has_active_booking: reason = "marked rented without an active booking" elif vehicle.operational_status == "available" and vehicle.id in open_high_by_vehicle: reason = "marked available while a high-severity quality issue is open" elif vehicle.operational_status == "maintenance" and has_active_booking: reason = "marked maintenance while an active booking exists" if reason: _open_issue( db, scan, rule_type="vehicle_status_conflict", entity_type="vehicle", entity_id=vehicle.id, severity="high", summary=f"Vehicle {reason}", entity_ref=vehicle.public_ref, related_refs=[], ) def _scan_odometer_regressions(db: Session, scan: ScanResult) -> None: # The seed dataset's vehicle.odometer_km is generated independently of booking # history, so comparing every historical booking against it produces near-universal # false positives. Instead check the booking sequence's own internal consistency: # each vehicle's completed bookings should show a non-decreasing odometer reading. vehicles = {v.id: v for v in db.scalars(select(Vehicle)).all()} bookings_by_vehicle: dict[uuid.UUID, list[Booking]] = {} for booking in db.scalars( select(Booking).where( Booking.status == "returned", Booking.end_odometer_km.is_not(None) ) ).all(): bookings_by_vehicle.setdefault(booking.vehicle_id, []).append(booking) for vehicle_id, bookings in bookings_by_vehicle.items(): bookings.sort(key=lambda b: b.ends_at) for earlier, later in zip(bookings, bookings[1:], strict=False): if later.end_odometer_km < earlier.end_odometer_km: vehicle = vehicles[vehicle_id] _open_issue( db, scan, rule_type="odometer_regression", entity_type="vehicle", entity_id=vehicle_id, severity="medium", summary=( f"Booking {later.public_ref} recorded {later.end_odometer_km} km, " f"below the {earlier.end_odometer_km} km recorded by earlier " f"booking {earlier.public_ref}." ), entity_ref=vehicle.public_ref, related_refs=[earlier.public_ref, later.public_ref], ) break def run_scan(db: Session) -> ScanResult: scan = ScanResult() _scan_duplicate_customers(db, scan) _scan_missing_required_fields(db, scan) _scan_odometer_regressions(db, scan) _scan_booking_overlaps(db, scan) _scan_vehicle_status_conflicts(db, scan) db.commit() return scan def _load_open_issue(db: Session, public_ref: str) -> DataQualityIssue: issue = db.scalar( select(DataQualityIssue).where(DataQualityIssue.public_ref == public_ref) ) if issue is None: raise AppError("ISSUE_NOT_FOUND", "Data quality issue not found.", status_code=404) if issue.status != "open": raise AppError( "ISSUE_NOT_OPEN", f"Issue is '{issue.status}', not 'open'.", status_code=409, ) return issue def defer_issue(db: Session, public_ref: str, actor: CurrentUser) -> DataQualityIssue: issue = _load_open_issue(db, public_ref) issue.status = "deferred" issue.resolved_at = datetime.now(UTC) issue.resolved_by = actor.display_name record_audit_event( db, actor_type="user", actor_label=actor.display_name, action="data_quality_issue_deferred", entity_type="data_quality_issue", entity_id=issue.id, before={"status": "open"}, after={"status": "deferred"}, ) db.commit() return issue def reject_issue(db: Session, public_ref: str, actor: CurrentUser) -> DataQualityIssue: issue = _load_open_issue(db, public_ref) issue.status = "rejected" issue.resolved_at = datetime.now(UTC) issue.resolved_by = actor.display_name record_audit_event( db, actor_type="user", actor_label=actor.display_name, action="data_quality_issue_rejected", entity_type="data_quality_issue", entity_id=issue.id, before={"status": "open"}, after={"status": "rejected"}, ) db.commit() return issue MERGEABLE_FIELDS = ("first_name", "last_name", "email", "phone", "postal_code", "city") def merge_customers( db: Session, public_ref: str, survivor_ref: str, field_overrides: dict[str, str] | None, actor: CurrentUser, ) -> dict: issue = _load_open_issue(db, public_ref) if issue.rule_type != "possible_duplicate_customer": raise AppError( "NOT_A_DUPLICATE_ISSUE", "This issue is not a possible-duplicate-customer issue.", status_code=409, ) entity_ref = issue.evidence_json.get("entity_ref") related_refs = issue.evidence_json.get("related_refs", []) candidate_refs = {entity_ref, *related_refs} if survivor_ref not in candidate_refs: raise AppError( "INVALID_SURVIVOR", "The survivor reference must be one of the two customers in this issue.", status_code=422, details={"candidates": sorted(candidate_refs)}, ) loser_ref = next(ref for ref in candidate_refs if ref != survivor_ref) survivor = db.scalar(select(Customer).where(Customer.public_ref == survivor_ref)) loser = db.scalar(select(Customer).where(Customer.public_ref == loser_ref)) if survivor is None or loser is None: raise AppError( "CUSTOMER_NOT_FOUND", "One of the customers could not be found.", status_code=404 ) before = { "survivor": {f: getattr(survivor, f) for f in MERGEABLE_FIELDS}, "loser": {f: getattr(loser, f) for f in MERGEABLE_FIELDS}, } for field_name, value in (field_overrides or {}).items(): if field_name not in MERGEABLE_FIELDS: raise AppError( "INVALID_FIELD_OVERRIDE", f"Field '{field_name}' cannot be merged.", status_code=422 ) setattr(survivor, field_name, value) rewired = db.execute( Booking.__table__.update() .where(Booking.customer_id == loser.id) .values(customer_id=survivor.id) ) loser.merged_into_customer_id = survivor.id issue.status = "resolved" issue.resolved_at = datetime.now(UTC) issue.resolved_by = actor.display_name record_audit_event( db, actor_type="user", actor_label=actor.display_name, action="customer_merged", entity_type="customer", entity_id=survivor.id, before=before, after={"survivor": {f: getattr(survivor, f) for f in MERGEABLE_FIELDS}}, metadata={ "loser_ref": loser_ref, "survivor_ref": survivor_ref, "rewired_bookings": rewired.rowcount, }, ) db.commit() return { "issue_ref": issue.public_ref, "survivor_ref": survivor_ref, "loser_ref": loser_ref, "rewired_bookings": rewired.rowcount, }