"""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))