508 lines
20 KiB
Python
508 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import AppError
|
|
from app.models.booking import Booking
|
|
from app.models.data_quality import DataQualityIssue
|
|
from app.models.inspection import Inspection
|
|
from app.models.vehicle import Vehicle
|
|
from app.schemas import CurrentUser, ResolveOdometerRegressionRequest
|
|
from app.services.audit import record_audit_event
|
|
from app.services.data_quality_common import issue_due_at, load_open_issue, new_scan_ref
|
|
|
|
|
|
def _odometer_fingerprint(
|
|
*,
|
|
source_type: str,
|
|
later_ref: str,
|
|
later_km: int,
|
|
) -> dict[str, str | int]:
|
|
"""Stable identity for one reviewed regression, independent of issue IDs."""
|
|
return {
|
|
"source_type": source_type,
|
|
"later_ref": later_ref,
|
|
"later_km": later_km,
|
|
}
|
|
|
|
|
|
def _was_odometer_regression_retained(
|
|
db: Session,
|
|
*,
|
|
vehicle_id: uuid.UUID,
|
|
fingerprint: dict[str, str | int],
|
|
) -> bool:
|
|
reviewed = db.scalars(
|
|
select(DataQualityIssue).where(
|
|
DataQualityIssue.rule_type == "odometer_regression",
|
|
DataQualityIssue.entity_type == "vehicle",
|
|
DataQualityIssue.entity_id == vehicle_id,
|
|
DataQualityIssue.status == "resolved",
|
|
)
|
|
).all()
|
|
return any(
|
|
issue.evidence_json.get("resolution_decision") == "retain_canonical"
|
|
and fingerprint in issue.evidence_json.get("retained_odometer_fingerprints", [])
|
|
for issue in reviewed
|
|
)
|
|
|
|
|
|
def open_odometer_regression_issue(
|
|
db: Session,
|
|
*,
|
|
vehicle: Vehicle,
|
|
reading_ref: str,
|
|
reading_km: int,
|
|
canonical_km: int,
|
|
source_type: str,
|
|
related_refs: list[str],
|
|
correctable_booking_refs: list[str] | None = None,
|
|
canonical_ref: str | None = None,
|
|
detected_at: datetime | None = None,
|
|
public_ref: str | None = None,
|
|
actor_label: str | None = None,
|
|
actor_type: str = "user",
|
|
correlation_id: uuid.UUID | None = None,
|
|
) -> DataQualityIssue | None:
|
|
"""Open one explainable DQ-03 issue while the caller holds the vehicle lock."""
|
|
if reading_km >= canonical_km:
|
|
return None
|
|
earlier_ref = canonical_ref or vehicle.public_ref
|
|
fingerprint = _odometer_fingerprint(
|
|
source_type=source_type,
|
|
later_ref=reading_ref,
|
|
later_km=reading_km,
|
|
)
|
|
# A manager's explicit "retain canonical" decision acknowledges this exact source
|
|
# fact. Reopening it on every scheduled scan would create churn; only changed or new
|
|
# evidence (and therefore a different fingerprint) is actionable again.
|
|
if _was_odometer_regression_retained(
|
|
db,
|
|
vehicle_id=vehicle.id,
|
|
fingerprint=fingerprint,
|
|
):
|
|
return None
|
|
existing = db.scalar(
|
|
select(DataQualityIssue)
|
|
.where(
|
|
DataQualityIssue.rule_type == "odometer_regression",
|
|
DataQualityIssue.entity_type == "vehicle",
|
|
DataQualityIssue.entity_id == vehicle.id,
|
|
DataQualityIssue.status == "open",
|
|
)
|
|
.with_for_update()
|
|
.execution_options(populate_existing=True)
|
|
)
|
|
if existing is not None:
|
|
before_related_refs = list(existing.evidence_json.get("related_refs", []))
|
|
before_correctable_refs = list(existing.evidence_json.get("correctable_booking_refs", []))
|
|
new_signal = {
|
|
"code": "odometer.regression",
|
|
"source_type": source_type,
|
|
"params": {
|
|
"later_ref": reading_ref,
|
|
"later_km": reading_km,
|
|
"earlier_ref": earlier_ref,
|
|
"earlier_km": canonical_km,
|
|
},
|
|
}
|
|
existing_signals = list(existing.evidence_json.get("signals", []))
|
|
signal_was_new = not any(
|
|
isinstance(signal, dict)
|
|
and isinstance(signal.get("params"), dict)
|
|
and _odometer_fingerprint(
|
|
source_type=str(
|
|
signal.get("source_type", existing.evidence_json.get("source_type"))
|
|
),
|
|
later_ref=str(signal["params"].get("later_ref")),
|
|
later_km=signal["params"].get("later_km"),
|
|
)
|
|
== fingerprint
|
|
for signal in existing_signals
|
|
if isinstance(signal, dict)
|
|
and isinstance(signal.get("params"), dict)
|
|
and isinstance(signal["params"].get("later_km"), int)
|
|
)
|
|
if signal_was_new:
|
|
existing_signals.append(new_signal)
|
|
merged_related_refs = list(
|
|
dict.fromkeys([*before_related_refs, *(related_refs if signal_was_new else [])])
|
|
)
|
|
merged_correctable_refs = list(
|
|
dict.fromkeys([*before_correctable_refs, *(correctable_booking_refs or [])])
|
|
)
|
|
if (
|
|
not signal_was_new
|
|
and merged_related_refs == before_related_refs
|
|
and merged_correctable_refs == before_correctable_refs
|
|
):
|
|
return existing
|
|
source_types = list(existing.evidence_json.get("source_types", []))
|
|
previous_source_type = existing.evidence_json.get("source_type")
|
|
if (
|
|
isinstance(previous_source_type, str)
|
|
and previous_source_type != "multiple"
|
|
and previous_source_type not in source_types
|
|
):
|
|
source_types.append(previous_source_type)
|
|
if source_type not in source_types:
|
|
source_types.append(source_type)
|
|
existing.evidence_json = {
|
|
**existing.evidence_json,
|
|
"summary": (
|
|
f"{source_type.title()} {reading_ref} recorded {reading_km} km, below "
|
|
f"the canonical {canonical_km} km for {vehicle.public_ref}."
|
|
),
|
|
"related_refs": merged_related_refs,
|
|
"correctable_booking_refs": merged_correctable_refs,
|
|
"source_type": source_type if len(source_types) == 1 else "multiple",
|
|
"source_types": source_types,
|
|
"signals": existing_signals,
|
|
}
|
|
if actor_label is not None and (
|
|
merged_related_refs != before_related_refs
|
|
or merged_correctable_refs != before_correctable_refs
|
|
or signal_was_new
|
|
):
|
|
record_audit_event(
|
|
db,
|
|
actor_type=actor_type,
|
|
actor_label=actor_label,
|
|
action="data_quality_issue_evidence_updated",
|
|
entity_type="data_quality_issue",
|
|
entity_id=existing.id,
|
|
correlation_id=correlation_id,
|
|
before={
|
|
"related_refs": before_related_refs,
|
|
"correctable_booking_refs": before_correctable_refs,
|
|
},
|
|
after={
|
|
"related_refs": merged_related_refs,
|
|
"correctable_booking_refs": merged_correctable_refs,
|
|
},
|
|
metadata={"vehicle_ref": vehicle.public_ref, "reading_ref": reading_ref},
|
|
)
|
|
return existing
|
|
now = detected_at or datetime.now(UTC)
|
|
previous = db.scalar(
|
|
select(DataQualityIssue)
|
|
.where(
|
|
DataQualityIssue.rule_type == "odometer_regression",
|
|
DataQualityIssue.entity_type == "vehicle",
|
|
DataQualityIssue.entity_id == vehicle.id,
|
|
DataQualityIssue.status != "open",
|
|
)
|
|
.order_by(DataQualityIssue.detected_at.desc())
|
|
)
|
|
evidence = {
|
|
"summary": (
|
|
f"{source_type.title()} {reading_ref} recorded {reading_km} km, below "
|
|
f"the canonical {canonical_km} km for {vehicle.public_ref}."
|
|
),
|
|
"entity_ref": vehicle.public_ref,
|
|
"related_refs": related_refs,
|
|
"correctable_booking_refs": correctable_booking_refs or [],
|
|
"source_type": source_type,
|
|
"source_types": [source_type],
|
|
"signals": [
|
|
{
|
|
"code": "odometer.regression",
|
|
"source_type": source_type,
|
|
"params": {
|
|
"later_ref": reading_ref,
|
|
"later_km": reading_km,
|
|
"earlier_ref": earlier_ref,
|
|
"earlier_km": canonical_km,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
if previous is not None:
|
|
evidence["reopened_from"] = previous.public_ref
|
|
evidence["previous_decision"] = previous.status
|
|
issue = DataQualityIssue(
|
|
public_ref=public_ref or new_scan_ref("DQ-ODO"),
|
|
rule_type="odometer_regression",
|
|
entity_type="vehicle",
|
|
entity_id=vehicle.id,
|
|
severity="medium",
|
|
status="open",
|
|
evidence_json=evidence,
|
|
proposed_action_json={},
|
|
detected_at=now,
|
|
due_at=issue_due_at(now, "medium"),
|
|
)
|
|
db.add(issue)
|
|
db.flush()
|
|
if actor_label is not None:
|
|
record_audit_event(
|
|
db,
|
|
actor_type=actor_type,
|
|
actor_label=actor_label,
|
|
action="data_quality_issue_created",
|
|
entity_type="data_quality_issue",
|
|
entity_id=issue.id,
|
|
correlation_id=correlation_id,
|
|
after={"status": "open", "rule_type": "odometer_regression"},
|
|
metadata={"vehicle_ref": vehicle.public_ref, "reading_ref": reading_ref},
|
|
)
|
|
return issue
|
|
|
|
|
|
def resolve_odometer_regression(
|
|
db: Session, public_ref: str, body: ResolveOdometerRegressionRequest, actor: CurrentUser
|
|
) -> DataQualityIssue:
|
|
# Read the routing data without a row lock first. The canonical mutation order is
|
|
# booking -> vehicle -> issue -> inspection everywhere, matching checkout/return.
|
|
# Locking the issue before the booking creates a resolver-vs-return deadlock.
|
|
issue_snapshot = load_open_issue(db, public_ref, lock=False)
|
|
if issue_snapshot.rule_type != "odometer_regression":
|
|
raise AppError(
|
|
"NOT_AN_ODOMETER_ISSUE",
|
|
"This issue is not an odometer_regression issue.",
|
|
status_code=409,
|
|
)
|
|
|
|
def correctable_refs(issue: DataQualityIssue) -> list[str]:
|
|
if "correctable_booking_refs" in issue.evidence_json:
|
|
refs = issue.evidence_json.get("correctable_booking_refs", [])
|
|
else:
|
|
# Backward compatibility for issues created before source-aware evidence.
|
|
refs = [
|
|
ref
|
|
for ref in issue.evidence_json.get("related_refs", [])
|
|
if isinstance(ref, str) and ref.startswith("BK-")
|
|
]
|
|
return [ref for ref in refs if isinstance(ref, str)]
|
|
|
|
booking: Booking | None = None
|
|
if body.decision != "retain_canonical":
|
|
if body.booking_ref not in correctable_refs(issue_snapshot):
|
|
raise AppError(
|
|
"INVALID_BOOKING_REFERENCE",
|
|
"booking_ref must be one of this issue's related bookings.",
|
|
status_code=422,
|
|
)
|
|
if body.corrected_odometer_km is None:
|
|
raise AppError(
|
|
"CORRECTED_VALUE_REQUIRED",
|
|
"corrected_odometer_km is required when correcting a reading.",
|
|
status_code=422,
|
|
)
|
|
booking = db.scalar(
|
|
select(Booking).where(Booking.public_ref == body.booking_ref).with_for_update()
|
|
)
|
|
if booking is None:
|
|
raise AppError(
|
|
"BOOKING_NOT_FOUND", "The booking to correct was not found.", status_code=404
|
|
)
|
|
vehicle = db.scalar(
|
|
select(Vehicle).where(Vehicle.id == issue_snapshot.entity_id).with_for_update()
|
|
)
|
|
if vehicle is None:
|
|
raise AppError(
|
|
"VEHICLE_NOT_FOUND", "The vehicle for this issue was not found.", status_code=404
|
|
)
|
|
|
|
# A return or scan may have appended evidence while we waited for the domain locks.
|
|
# Lock and refresh the issue only now, then revalidate every decision against that
|
|
# current evidence instead of resolving a stale snapshot.
|
|
issue = load_open_issue(db, public_ref)
|
|
if issue.rule_type != "odometer_regression" or issue.entity_id != vehicle.id:
|
|
raise AppError(
|
|
"ISSUE_CHANGED",
|
|
"The issue changed while the correction was being prepared. Review it again.",
|
|
status_code=409,
|
|
)
|
|
if booking is not None:
|
|
if booking.vehicle_id != vehicle.id or booking.public_ref not in correctable_refs(issue):
|
|
raise AppError(
|
|
"INVALID_BOOKING_REFERENCE",
|
|
"booking_ref must be one of this issue's related bookings.",
|
|
status_code=422,
|
|
)
|
|
|
|
correlation_id = uuid.uuid4()
|
|
|
|
if body.decision == "retain_canonical":
|
|
retained_fingerprints: list[dict[str, str | int]] = []
|
|
fallback_source_type = str(issue.evidence_json.get("source_type", "unknown"))
|
|
for signal in issue.evidence_json.get("signals", []):
|
|
if not isinstance(signal, dict) or signal.get("code") != "odometer.regression":
|
|
continue
|
|
params = signal.get("params")
|
|
if not isinstance(params, dict):
|
|
continue
|
|
source_type = signal.get("source_type", fallback_source_type)
|
|
later_ref = params.get("later_ref")
|
|
later_km = params.get("later_km")
|
|
earlier_ref = params.get("earlier_ref")
|
|
earlier_km = params.get("earlier_km")
|
|
if (
|
|
isinstance(source_type, str)
|
|
and isinstance(later_ref, str)
|
|
and isinstance(later_km, int)
|
|
and isinstance(earlier_ref, str)
|
|
and isinstance(earlier_km, int)
|
|
):
|
|
retained_fingerprints.append(
|
|
_odometer_fingerprint(
|
|
source_type=source_type,
|
|
later_ref=later_ref,
|
|
later_km=later_km,
|
|
)
|
|
)
|
|
issue.evidence_json = {
|
|
**issue.evidence_json,
|
|
"resolution_decision": "retain_canonical",
|
|
"retained_odometer_fingerprints": retained_fingerprints,
|
|
}
|
|
record_audit_event(
|
|
db,
|
|
actor_type="user",
|
|
actor_label=actor.display_name,
|
|
action="data_quality_odometer_retained",
|
|
entity_type="vehicle",
|
|
entity_id=vehicle.id,
|
|
correlation_id=correlation_id,
|
|
metadata={
|
|
"issue_ref": issue.public_ref,
|
|
"canonical_odometer_km": vehicle.odometer_km,
|
|
"retained_fingerprint_count": len(retained_fingerprints),
|
|
},
|
|
)
|
|
else:
|
|
assert booking is not None and body.corrected_odometer_km is not None
|
|
# Never silently lower the canonical odometer: a correction must be at or above
|
|
# the current canonical value, otherwise it would just create a new regression.
|
|
if body.corrected_odometer_km < vehicle.odometer_km:
|
|
raise AppError(
|
|
"CORRECTION_BELOW_CANONICAL",
|
|
(
|
|
f"Corrected value {body.corrected_odometer_km} km is still below the "
|
|
f"canonical {vehicle.odometer_km} km; it would not resolve the regression."
|
|
),
|
|
status_code=422,
|
|
)
|
|
|
|
# A returned booking has two persisted representations of the same reading.
|
|
# Correct every return inspection for the selected booking, including historical
|
|
# seed rows whose old issue evidence did not yet cite the inspection explicitly.
|
|
related_inspections = db.scalars(
|
|
select(Inspection)
|
|
.where(Inspection.booking_id == booking.id, Inspection.type == "return")
|
|
.with_for_update()
|
|
).all()
|
|
before = {
|
|
"booking_end_odometer_km": booking.end_odometer_km,
|
|
"vehicle_odometer_km": vehicle.odometer_km,
|
|
"inspection_odometer_km": {
|
|
inspection.public_ref: inspection.odometer_km for inspection in related_inspections
|
|
},
|
|
}
|
|
booking.end_odometer_km = body.corrected_odometer_km
|
|
for inspection in related_inspections:
|
|
inspection.odometer_km = body.corrected_odometer_km
|
|
vehicle.odometer_km = body.corrected_odometer_km
|
|
vehicle.version += 1
|
|
record_audit_event(
|
|
db,
|
|
actor_type="user",
|
|
actor_label=actor.display_name,
|
|
action="data_quality_odometer_corrected",
|
|
entity_type="vehicle",
|
|
entity_id=vehicle.id,
|
|
correlation_id=correlation_id,
|
|
before=before,
|
|
after={
|
|
"booking_end_odometer_km": booking.end_odometer_km,
|
|
"vehicle_odometer_km": vehicle.odometer_km,
|
|
"inspection_odometer_km": {
|
|
inspection.public_ref: inspection.odometer_km
|
|
for inspection in related_inspections
|
|
},
|
|
},
|
|
metadata={"issue_ref": issue.public_ref, "booking_ref": booking.public_ref},
|
|
)
|
|
|
|
has_remaining_evidence = False
|
|
if booking is not None:
|
|
target_refs = {booking.public_ref, *(item.public_ref for item in related_inspections)}
|
|
before_related = [
|
|
ref for ref in issue.evidence_json.get("related_refs", []) if isinstance(ref, str)
|
|
]
|
|
before_correctable = correctable_refs(issue)
|
|
before_signals = [
|
|
signal for signal in issue.evidence_json.get("signals", []) if isinstance(signal, dict)
|
|
]
|
|
remaining_signals = [
|
|
signal
|
|
for signal in before_signals
|
|
if not (
|
|
isinstance(signal.get("params"), dict)
|
|
and signal["params"].get("later_ref") in target_refs
|
|
)
|
|
]
|
|
remaining_related = [ref for ref in before_related if ref not in target_refs]
|
|
remaining_correctable = [ref for ref in before_correctable if ref != booking.public_ref]
|
|
has_remaining_evidence = bool(remaining_signals or remaining_correctable)
|
|
issue.evidence_json = {
|
|
**issue.evidence_json,
|
|
"summary": (
|
|
"Additional odometer regression evidence remains for review."
|
|
if has_remaining_evidence
|
|
else issue.evidence_json.get("summary", "Odometer reading corrected.")
|
|
),
|
|
"related_refs": remaining_related,
|
|
"correctable_booking_refs": remaining_correctable,
|
|
"signals": remaining_signals,
|
|
}
|
|
if has_remaining_evidence:
|
|
record_audit_event(
|
|
db,
|
|
actor_type="user",
|
|
actor_label=actor.display_name,
|
|
action="data_quality_issue_partially_resolved",
|
|
entity_type="data_quality_issue",
|
|
entity_id=issue.id,
|
|
correlation_id=correlation_id,
|
|
before={
|
|
"related_refs": before_related,
|
|
"correctable_booking_refs": before_correctable,
|
|
"signal_count": len(before_signals),
|
|
},
|
|
after={
|
|
"related_refs": remaining_related,
|
|
"correctable_booking_refs": remaining_correctable,
|
|
"signal_count": len(remaining_signals),
|
|
},
|
|
metadata={
|
|
"decision": body.decision,
|
|
"booking_ref": booking.public_ref,
|
|
"note": body.note,
|
|
},
|
|
)
|
|
|
|
if not has_remaining_evidence:
|
|
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="data_quality_issue_resolved",
|
|
entity_type="data_quality_issue",
|
|
entity_id=issue.id,
|
|
correlation_id=correlation_id,
|
|
before={"status": "open"},
|
|
after={"status": "resolved"},
|
|
metadata={"decision": body.decision, "note": body.note},
|
|
)
|
|
db.commit()
|
|
return issue
|