| 1 | """Watchlist matching. Pure matcher (testable) + DB-bound evaluation used during ingest.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | from dataclasses import dataclass |
| 5 | |
| 6 | |
| 7 | @dataclass |
| 8 | class WatchEntry: |
| 9 | id: str |
| 10 | plate_norm: str | None = None |
| 11 | partial_plate: str | None = None |
| 12 | vehicle_desc: dict | None = None |
| 13 | |
| 14 | |
| 15 | def match_entry(entry: WatchEntry, plate_norm: str, vehicle: dict | None = None) -> str | None: |
| 16 | """Return the match reason string, or None. Order: exact > partial > vehicle description.""" |
| 17 | if entry.plate_norm and entry.plate_norm == plate_norm: |
| 18 | return "exact_plate" |
| 19 | if entry.partial_plate: |
| 20 | p = entry.partial_plate.upper() |
| 21 | if plate_norm.startswith(p) or p in plate_norm: |
| 22 | return "partial_plate" |
| 23 | if entry.vehicle_desc and vehicle: |
| 24 | wanted = {k: v for k, v in entry.vehicle_desc.items() if v} |
| 25 | if wanted and all(str(vehicle.get(k, "")).lower() == str(v).lower() for k, v in wanted.items()): |
| 26 | return "vehicle_description" |
| 27 | return None |
| 28 | |
| 29 | |
| 30 | def match_all(entries: list[WatchEntry], plate_norm: str, vehicle: dict | None = None) -> list[dict]: |
| 31 | hits = [] |
| 32 | for e in entries: |
| 33 | reason = match_entry(e, plate_norm, vehicle) |
| 34 | if reason: |
| 35 | hits.append({"entry_id": e.id, "matched_on": reason}) |
| 36 | return hits |