Fix per-carrier booking class in verify/scrape commands (closes #2)

AA uses H class for oneworld Explorer business, not D. The verify and
scrape commands hardcoded D for all carriers, producing false negatives
on AA segments. Now resolves booking class per carrier from carriers.yaml
(AA=H, others=D) with --class flag as optional override.

- New rtw/carriers.py shared utility with get_booking_class()
- DClassVerifier resolves per-segment, updated cache keys
- DClassResult.display_code uses actual class letter (H9/D9)
- CLI --class default changed to auto per-carrier lookup
- 37 new tests (822 total, 0 failures)

Co-Authored-By: Claude Opus 4.6 <[email protected]>

Author: claude

rtw/booking.py

@@ -68,28 +68,15 @@ def __init__(self) -> None:
     def _get_booking_class(self, carrier: Optional[str], cabin: CabinClass) -> Optional[str]:
         """Determine the booking class for a carrier/cabin combination.
 
-        Business: AA -> H (special case), others -> D (from carriers.yaml).
-        Economy: L for most carriers.
-        First: A for most carriers.
-        Surface segments (carrier=None): returns None.
+        Delegates to shared utility in rtw.carriers. Returns None for
+        surface segments (carrier=None).
         """
         if carrier is None:
             return None
 
-        carrier = carrier.upper()
+        from rtw.carriers import get_booking_class
 
-        if cabin == CabinClass.BUSINESS:
-            # Use rtw_booking_class from carriers.yaml (AA=H, others=D)
-            carrier_data = self._carriers.get(carrier, {})
-            return carrier_data.get("rtw_booking_class", "D")
-
-        if cabin == CabinClass.ECONOMY:
-            return "L"
-
-        if cabin == CabinClass.FIRST:
-            return "A"
-
-        return "D"
+        return get_booking_class(carrier, cabin)
 
     def _is_same_city(self, airport1: str, airport2: str) -> bool:
         """Check if two airports are in the same city group."""

rtw/carriers.py

@@ -0,0 +1,45 @@
+"""Shared carrier booking class resolution.
+
+Resolves the correct booking class for a carrier/cabin combination
+using data from carriers.yaml. AA uses H class for oneworld Explorer
+business; all other carriers use D.
+"""
+
+from pathlib import Path
+from typing import Optional
+
+import yaml
+
+from rtw.models import CabinClass
+
+_DATA_DIR = Path(__file__).parent / "data"
+with open(_DATA_DIR / "carriers.yaml") as f:
+    _CARRIERS: dict = yaml.safe_load(f)
+
+
+def get_booking_class(carrier: Optional[str], cabin: CabinClass) -> str:
+    """Return the booking class for a carrier/cabin combination.
+
+    Business: AA -> H (from carriers.yaml rtw_booking_class), others -> D.
+    Economy: L for all carriers.
+    First: A for all carriers.
+    Surface segments (carrier=None): returns D as safe default.
+
+    Always returns a concrete single-letter string, never None.
+    """
+    if carrier is None:
+        return "D"
+
+    carrier = carrier.upper()
+
+    if cabin == CabinClass.BUSINESS:
+        carrier_data = _CARRIERS.get(carrier, {})
+        return carrier_data.get("rtw_booking_class", "D")
+
+    if cabin == CabinClass.ECONOMY:
+        return "L"
+
+    if cabin == CabinClass.FIRST:
+        return "A"
+
+    return "D"

rtw/cli.py

@@ -648,7 +648,7 @@ def scrape_prices(
 @scrape_app.command(name="availability")
 def scrape_availability(
     file: str = typer.Argument(help="Path to itinerary YAML file"),
-    booking_class: str = typer.Option("D", "--class", "-c", help="Booking class to check"),
+    booking_class: Optional[str] = typer.Option(None, "--class", "-c", help="Override booking class (default: auto per carrier, AA=H, others=D)"),
     verbose: VerboseFlag = False,
     quiet: QuietFlag = False,
 ) -> None:
@@ -1059,7 +1059,7 @@ def verify(
     option_ids: Annotated[
         Optional[list[int]], typer.Argument(help="Option IDs to verify (1-based). Omit for top 3.")
     ] = None,
-    booking_class: Annotated[str, typer.Option("--class", "-c", help="Booking class")] = "D",
+    booking_class: Annotated[Optional[str], typer.Option("--class", "-c", help="Override booking class (default: auto per carrier, AA=H, others=D)")] = None,
     no_cache: Annotated[bool, typer.Option("--no-cache", help="Skip cache")] = False,
     json: JsonFlag = False,
     plain: PlainFlag = False,
@@ -1128,6 +1128,7 @@ def verify(
                 cache=ScrapeCache(),
                 booking_class=booking_class,
             )
+            # Note: booking_class=None means auto per-carrier (AA=H, others=D)
 
             # Convert and verify with Rich progress
             results = []

rtw/scraper/batch.py

@@ -130,38 +130,45 @@ def _try_playwright_price(origin, dest, seg_date, cabin):
 
 def check_itinerary_availability(
     itinerary: Itinerary,
-    booking_class: str = "D",
+    booking_class: Optional[str] = None,
 ) -> list[Optional[dict]]:
     """Check award availability for all flown segments.
 
     Args:
         itinerary: The RTW itinerary to check.
-        booking_class: Booking class to check (default "D" for business award).
+        booking_class: Override booking class for all segments. When None
+            (default), resolves per carrier from carriers.yaml (AA=H, others=D).
 
     Returns:
         List of availability dicts (or None) for each segment.
         Surface segments and segments without carriers get None. Never raises.
     """
+    from rtw.carriers import get_booking_class
+
     scraper = ExpertFlyerScraper()
 
     if not scraper.credentials_available():
         logger.info("ExpertFlyer credentials not available - returning empty results")
         return [None] * len(itinerary.segments)
 
+    cabin = itinerary.ticket.cabin
+
     results: list[Optional[dict]] = []
 
     for seg in itinerary.segments:
         if seg.is_surface or seg.carrier is None or seg.date is None:
             results.append(None)
             continue
 
+        seg_bc = booking_class if booking_class is not None else get_booking_class(seg.carrier, cabin)
+
         try:
             avail = scraper.check_availability(
                 origin=seg.from_airport,
                 dest=seg.to_airport,
                 date=seg.date,
                 carrier=seg.carrier,
-                booking_class=booking_class,
+                booking_class=seg_bc,
             )
             results.append(avail)
         except Exception as exc:

rtw/verify/models.py

@@ -41,7 +41,7 @@ class FlightAvailability(BaseModel):
 
 
 class DClassResult(BaseModel):
-    """Result of a D-class check for a single flight segment."""
+    """Result of an availability check for a single flight segment."""
 
     status: DClassStatus
     seats: int = Field(default=0, ge=0, le=9)
@@ -50,6 +50,7 @@ class DClassResult(BaseModel):
     origin: str = Field(min_length=3, max_length=3)
     destination: str = Field(min_length=3, max_length=3)
     target_date: datetime.date
+    booking_class: str = "D"
     checked_at: datetime.datetime = Field(
         default_factory=lambda: datetime.datetime.now(datetime.timezone.utc)
     )
@@ -64,7 +65,7 @@ def available(self) -> bool:
 
     @property
     def available_flights(self) -> list[FlightAvailability]:
-        """Flights with D-class seats > 0, sorted by seats desc then departure."""
+        """Flights with seats > 0, sorted by seats desc then departure."""
         avail = [f for f in self.flights if f.seats > 0]
         return sorted(avail, key=lambda f: (-f.seats, f.depart_time or ""))
 
@@ -78,14 +79,15 @@ def available_count(self) -> int:
 
     @property
     def display_code(self) -> str:
-        """Short display code: D9 (3 avl), D0, D?, D!"""
+        """Short display code: H9 (3 avl), D0, D?, H!"""
+        bc = self.booking_class
         if self.status == DClassStatus.ERROR:
-            return "D!"
+            return f"{bc}!"
         if self.status == DClassStatus.UNKNOWN:
-            return "D?"
+            return f"{bc}?"
         if self.flights:
-            return f"D{self.seats} ({self.available_count} avl)"
-        return f"D{self.seats}"
+            return f"{bc}{self.seats} ({self.available_count} avl)"
+        return f"{bc}{self.seats}"
 
     @property
     def best_alternate(self) -> Optional[AlternateDateResult]:

rtw/verify/verifier.py

@@ -1,13 +1,16 @@
-"""D-class verification orchestrator.
+"""Availability verification orchestrator.
 
 Coordinates the scraper, cache, and progress reporting to verify
-D-class availability across all flown segments of an itinerary option.
+award class availability across all flown segments of an itinerary option.
+Uses per-carrier booking class resolution (AA=H, others=D for business).
 """
 
 import logging
 import time
 from typing import Optional
 
+from rtw.carriers import get_booking_class
+from rtw.models import CabinClass
 from rtw.scraper.cache import ScrapeCache
 from rtw.scraper.expertflyer import ExpertFlyerScraper, SessionExpiredError
 from rtw.verify.models import (
@@ -26,28 +29,44 @@
 
 
 class DClassVerifier:
-    """Verify D-class availability for itinerary segments.
+    """Verify award class availability for itinerary segments.
 
     Checks each flown segment against ExpertFlyer, using the cache
     to avoid redundant queries. Surface segments are skipped.
+
+    Resolves booking class per carrier (AA=H, others=D for business)
+    unless an explicit override is provided.
     """
 
     def __init__(
         self,
         scraper: ExpertFlyerScraper,
         cache: Optional[ScrapeCache] = None,
-        booking_class: str = "D",
+        booking_class: Optional[str] = None,
+        cabin: CabinClass = CabinClass.BUSINESS,
     ) -> None:
         self.scraper = scraper
         self.cache = cache or ScrapeCache()
-        self.booking_class = booking_class
+        self._booking_class_override = booking_class
+        self.cabin = cabin
         self._session_expired = False
 
+    def _get_segment_booking_class(self, seg: SegmentVerification) -> str:
+        """Resolve the booking class for a segment.
+
+        If an override was set, use it for all segments.
+        Otherwise, look up per carrier from carriers.yaml.
+        """
+        if self._booking_class_override is not None:
+            return self._booking_class_override
+        return get_booking_class(seg.carrier, self.cabin)
+
     def _cache_key(self, seg: SegmentVerification) -> str:
         """Build cache key for a segment."""
+        bc = self._get_segment_booking_class(seg)
         return (
             f"{_CACHE_KEY_PREFIX}_{seg.carrier}_{seg.origin}_"
-            f"{seg.destination}_{seg.target_date}_{self.booking_class}"
+            f"{seg.destination}_{seg.target_date}_{bc}"
         )
 
     def _check_cache(self, seg: SegmentVerification) -> Optional[DClassResult]:
@@ -111,6 +130,7 @@ def verify_option(
                     origin=seg.origin,
                     destination=seg.destination,
                     target_date=seg.target_date,
+                    booking_class=self._get_segment_booking_class(seg),
                     error_message="Session expired during batch",
                 )
                 result.segments.append(verified)
@@ -132,13 +152,14 @@ def verify_option(
 
             # Call scraper
             try:
+                seg_bc = self._get_segment_booking_class(seg)
                 start = time.time()
                 dclass = self.scraper.check_availability(
                     origin=seg.origin,
                     dest=seg.destination,
                     date=seg.target_date,
                     carrier=seg.carrier or "",
-                    booking_class=self.booking_class,
+                    booking_class=seg_bc,
                 )
                 elapsed = time.time() - start
                 logger.debug(
@@ -150,6 +171,7 @@ def verify_option(
                 )
 
                 if dclass:
+                    dclass.booking_class = seg_bc
                     verified.dclass = dclass
                     self._store_cache(seg, dclass)
                 else:
@@ -160,6 +182,7 @@ def verify_option(
                         origin=seg.origin,
                         destination=seg.destination,
                         target_date=seg.target_date,
+                        booking_class=seg_bc,
                         error_message="Scraper returned None (no session?)",
                     )
 
@@ -172,6 +195,7 @@ def verify_option(
                     origin=seg.origin,
                     destination=seg.destination,
                     target_date=seg.target_date,
+                    booking_class=self._get_segment_booking_class(seg),
                     error_message=str(exc),
                 )
             except Exception as exc:
@@ -182,6 +206,7 @@ def verify_option(
                     origin=seg.origin,
                     destination=seg.destination,
                     target_date=seg.target_date,
+                    booking_class=self._get_segment_booking_class(seg),
                     error_message=str(exc),
                 )
 

tests/test_carriers.py

@@ -0,0 +1,58 @@
+"""Tests for shared carrier booking class resolution."""
+
+import pytest
+
+from rtw.carriers import get_booking_class
+from rtw.models import CabinClass
+
+
+class TestGetBookingClass:
+    """Test get_booking_class() for all carrier/cabin combinations."""
+
+    def test_aa_business_returns_h(self):
+        assert get_booking_class("AA", CabinClass.BUSINESS) == "H"
+
+    def test_aa_lowercase_returns_h(self):
+        assert get_booking_class("aa", CabinClass.BUSINESS) == "H"
+
+    def test_ba_business_returns_d(self):
+        assert get_booking_class("BA", CabinClass.BUSINESS) == "D"
+
+    def test_cx_business_returns_d(self):
+        assert get_booking_class("CX", CabinClass.BUSINESS) == "D"
+
+    def test_qr_business_returns_d(self):
+        assert get_booking_class("QR", CabinClass.BUSINESS) == "D"
+
+    def test_jl_business_returns_d(self):
+        assert get_booking_class("JL", CabinClass.BUSINESS) == "D"
+
+    def test_qf_business_returns_d(self):
+        assert get_booking_class("QF", CabinClass.BUSINESS) == "D"
+
+    def test_unknown_carrier_business_returns_d(self):
+        assert get_booking_class("ZZ", CabinClass.BUSINESS) == "D"
+
+    def test_economy_returns_l(self):
+        assert get_booking_class("AA", CabinClass.ECONOMY) == "L"
+
+    def test_economy_any_carrier_returns_l(self):
+        assert get_booking_class("BA", CabinClass.ECONOMY) == "L"
+
+    def test_first_returns_a(self):
+        assert get_booking_class("AA", CabinClass.FIRST) == "A"
+
+    def test_first_any_carrier_returns_a(self):
+        assert get_booking_class("QF", CabinClass.FIRST) == "A"
+
+    def test_none_carrier_returns_d(self):
+        """Surface segments (carrier=None) return safe default."""
+        assert get_booking_class(None, CabinClass.BUSINESS) == "D"
+
+    def test_never_returns_none(self):
+        """Function always returns a string, never None."""
+        for carrier in [None, "AA", "BA", "ZZ"]:
+            for cabin in CabinClass:
+                result = get_booking_class(carrier, cabin)
+                assert isinstance(result, str)
+                assert len(result) == 1