bsncubed
408fde440d
Photograph/screenshot a boarding pass, decode its BCBP barcode (PDF417/ Aztec/QR/DataMatrix), fetch the historical flight track from FlightAware the day after the flight, stage it for review, and write it into AirTrail via its REST API on approval. ntfy notifications carry signed Approve/Deny/Investigate actions.
142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
"""Parser for the mandatory fixed-width fields of an IATA Resolution 792
|
|
(BCBP) barcode string - the format encoded in PDF417 boarding pass barcodes.
|
|
|
|
Only the mandatory "unique" + first "repeated" (leg) field block is parsed
|
|
(offsets 0-59). Conditional/optional items beyond that are not needed for
|
|
this pipeline and are intentionally skipped.
|
|
"""
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, datetime, timedelta
|
|
|
|
FIELDS = [
|
|
("format_code", 1),
|
|
("legs", 1),
|
|
("name", 20),
|
|
("eticket_indicator", 1),
|
|
("pnr", 7),
|
|
("from_iata", 3),
|
|
("to_iata", 3),
|
|
("operating_carrier", 3),
|
|
("flight_number", 5),
|
|
("julian_day", 3),
|
|
("compartment", 1),
|
|
("seat", 4),
|
|
("checkin_sequence", 5),
|
|
("passenger_status", 1),
|
|
("field_size_hex", 2),
|
|
]
|
|
MANDATORY_LENGTH = sum(length for _, length in FIELDS) # 60
|
|
|
|
_AIRPORT_RE = re.compile(r"^[A-Z]{3}$")
|
|
_SEAT_RE = re.compile(r"^\d{1,3}[A-Z]$")
|
|
|
|
|
|
class BCBPParseError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class ParsedBCBP:
|
|
raw: str
|
|
format_code: str = ""
|
|
passenger_name: str = ""
|
|
pnr: str = ""
|
|
from_iata: str = ""
|
|
to_iata: str = ""
|
|
operating_carrier: str = ""
|
|
flight_number: str = ""
|
|
julian_day: int | None = None
|
|
compartment: str = ""
|
|
seat: str = ""
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
|
|
def _clean_leading_zeros(raw: str) -> str:
|
|
"""'0001 ' -> '1', '0025B' -> '25B'. Leaves non-numeric-prefixed values alone."""
|
|
s = raw.strip()
|
|
if not s:
|
|
return s
|
|
m = re.match(r"^(\d+)(.*)$", s)
|
|
if not m:
|
|
return s
|
|
digits, rest = m.groups()
|
|
digits = digits.lstrip("0") or "0"
|
|
return digits + rest.strip()
|
|
|
|
|
|
def _clean_name(raw: str) -> str:
|
|
s = raw.strip()
|
|
if "/" in s:
|
|
last, _, first = s.partition("/")
|
|
return f"{last.strip()} {first.strip()}".strip()
|
|
return s
|
|
|
|
|
|
def parse_bcbp(raw: str, upload_dt: datetime | None = None) -> ParsedBCBP:
|
|
if not raw or len(raw) < MANDATORY_LENGTH:
|
|
raise BCBPParseError(
|
|
f"barcode data too short ({len(raw) if raw else 0} chars, "
|
|
f"need at least {MANDATORY_LENGTH})"
|
|
)
|
|
|
|
mandatory = raw[:MANDATORY_LENGTH].upper()
|
|
offsets = {}
|
|
pos = 0
|
|
for name, length in FIELDS:
|
|
offsets[name] = mandatory[pos : pos + length]
|
|
pos += length
|
|
|
|
result = ParsedBCBP(raw=raw)
|
|
warnings = result.warnings
|
|
|
|
result.format_code = offsets["format_code"]
|
|
if result.format_code != "M":
|
|
warnings.append(f"unexpected format code {result.format_code!r} (expected 'M')")
|
|
|
|
result.passenger_name = _clean_name(offsets["name"])
|
|
result.pnr = offsets["pnr"].strip()
|
|
result.compartment = offsets["compartment"].strip()
|
|
|
|
from_iata = offsets["from_iata"].strip()
|
|
to_iata = offsets["to_iata"].strip()
|
|
if not _AIRPORT_RE.match(from_iata):
|
|
warnings.append(f"from_iata {from_iata!r} doesn't look like a valid IATA code")
|
|
if not _AIRPORT_RE.match(to_iata):
|
|
warnings.append(f"to_iata {to_iata!r} doesn't look like a valid IATA code")
|
|
result.from_iata = from_iata
|
|
result.to_iata = to_iata
|
|
|
|
result.operating_carrier = offsets["operating_carrier"].strip()
|
|
result.flight_number = _clean_leading_zeros(offsets["flight_number"])
|
|
|
|
seat = _clean_leading_zeros(offsets["seat"])
|
|
if seat and not _SEAT_RE.match(seat):
|
|
warnings.append(f"seat {seat!r} doesn't match the expected pattern")
|
|
result.seat = seat
|
|
|
|
julian_raw = offsets["julian_day"].strip()
|
|
try:
|
|
julian_day = int(julian_raw)
|
|
if not (1 <= julian_day <= 366):
|
|
raise ValueError
|
|
result.julian_day = julian_day
|
|
except ValueError:
|
|
warnings.append(f"julian_day {julian_raw!r} is not a valid day-of-year (1-366)")
|
|
result.julian_day = None
|
|
|
|
return result
|
|
|
|
|
|
def resolve_flight_date(julian_day: int, upload_dt: datetime) -> date:
|
|
"""BCBP encodes only a day-of-year, no year. Pick whichever of last/this/
|
|
next year's occurrence of that day-of-year is closest to the upload time -
|
|
handles both "photographed just before departure" and "photographed after
|
|
landing" cases."""
|
|
candidates = [
|
|
date(upload_dt.year + delta, 1, 1) + timedelta(days=julian_day - 1)
|
|
for delta in (-1, 0, 1)
|
|
]
|
|
return min(candidates, key=lambda d: abs((d - upload_dt.date()).days))
|