From ec8fc8113eb438670cd31f31985067c51d2ac648 Mon Sep 17 00:00:00 2001 From: peacemeals Date: Mon, 9 Feb 2026 23:35:06 +0000 Subject: [PATCH] =?UTF-8?q?Add=20=C2=B13=20days=20date=20flex=20for=20avai?= =?UTF-8?q?lability=20verification?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When --flex is passed to `rtw verify`, segments with no award availability on the target date are additionally checked on ±1, ±2, and ±3 adjacent days. Alternate dates with availability are reported in the results and the best alternate (most seats, closest date) is highlighted. - Add date_flex parameter to DClassVerifier with _check_alternate_dates() - Add --flex flag to verify CLI command - Add 5 tests covering flex behavior (triggered/not, session expiry, multi-seg) Co-Authored-By: Claude Opus 4.6 --- rtw/cli.py | 6 ++ rtw/verify/verifier.py | 79 ++++++++++++++++++ tests/test_verify/test_verifier.py | 129 +++++++++++++++++++++++++++++ 3 files changed, 214 insertions(+) diff --git a/rtw/cli.py b/rtw/cli.py index 02d96c8..e103583 100644 --- a/rtw/cli.py +++ b/rtw/cli.py @@ -1061,6 +1061,7 @@ def verify( ] = None, 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, + date_flex: Annotated[bool, typer.Option("--flex", help="Check ±3 days for alternate availability when target date is sold out")] = False, json: JsonFlag = False, plain: PlainFlag = False, verbose: VerboseFlag = False, @@ -1070,6 +1071,10 @@ def verify( Uses ExpertFlyer to check booking class availability on each flown segment. Requires a prior `rtw search` and `rtw login expertflyer`. + + With --flex, segments with no availability on the target date will + also be checked on ±1, ±2, and ±3 adjacent days. The best alternate + date is shown in the results. """ _setup_logging(verbose, quiet) @@ -1127,6 +1132,7 @@ def verify( scraper=scraper, cache=ScrapeCache(), booking_class=booking_class, + date_flex=date_flex, ) # Note: booking_class=None means auto per-carrier (AA=H, others=D) diff --git a/rtw/verify/verifier.py b/rtw/verify/verifier.py index 02df48b..1db77de 100644 --- a/rtw/verify/verifier.py +++ b/rtw/verify/verifier.py @@ -3,8 +3,12 @@ Coordinates the scraper, cache, and progress reporting to verify award class availability across all flown segments of an itinerary option. Uses per-carrier booking class resolution (AA=H, others=D for business). + +Supports date flex mode (±3 days) to find alternate travel dates when +the target date has no availability. """ +import datetime import logging import time from typing import Optional @@ -14,6 +18,7 @@ from rtw.scraper.cache import ScrapeCache from rtw.scraper.expertflyer import ExpertFlyerScraper, SessionExpiredError from rtw.verify.models import ( + AlternateDateResult, DClassResult, DClassStatus, ProgressCallback, @@ -26,6 +31,7 @@ _CACHE_TTL_HOURS = 24 _CACHE_KEY_PREFIX = "dclass" +_FLEX_DAYS = 3 # Check ±3 days when date_flex is enabled class DClassVerifier: @@ -36,6 +42,10 @@ class DClassVerifier: Resolves booking class per carrier (AA=H, others=D for business) unless an explicit override is provided. + + When date_flex=True, segments with no availability (seats=0) on the + target date will be checked on ±1-3 adjacent days. The best + alternate date is reported via DClassResult.alternate_dates. """ def __init__( @@ -44,11 +54,13 @@ def __init__( cache: Optional[ScrapeCache] = None, booking_class: Optional[str] = None, cabin: CabinClass = CabinClass.BUSINESS, + date_flex: bool = False, ) -> None: self.scraper = scraper self.cache = cache or ScrapeCache() self._booking_class_override = booking_class self.cabin = cabin + self.date_flex = date_flex self._session_expired = False def _get_segment_booking_class(self, seg: SegmentVerification) -> str: @@ -96,6 +108,61 @@ def _store_cache(self, seg: SegmentVerification, result: DClassResult) -> None: key = self._cache_key(seg) self.cache.set(key, result.model_dump(mode="json"), ttl_hours=_CACHE_TTL_HOURS) + def _check_alternate_dates( + self, + seg: SegmentVerification, + booking_class: str, + ) -> list[AlternateDateResult]: + """Check ±3 days around the target date for availability. + + Only called when the target date has no availability (seats=0). + Queries are made in order of proximity: ±1, ±2, ±3 days. + Stops early if session expires. + + Returns list of AlternateDateResult for dates with seats > 0. + """ + alternates: list[AlternateDateResult] = [] + target = seg.target_date + + # Check in order of proximity: ±1, ±2, ±3 + for offset in range(1, _FLEX_DAYS + 1): + for direction in (+1, -1): + day_offset = offset * direction + alt_date = target + datetime.timedelta(days=day_offset) + + if self._session_expired: + return alternates + + try: + alt_result = self.scraper.check_availability( + origin=seg.origin, + dest=seg.destination, + date=alt_date, + carrier=seg.carrier or "", + booking_class=booking_class, + ) + if alt_result and alt_result.seats > 0: + alternates.append(AlternateDateResult( + date=alt_date, + seats=alt_result.seats, + offset_days=day_offset, + )) + logger.info( + " Date flex %s→%s: %s has %s%d", + seg.origin, seg.destination, + alt_date, booking_class, alt_result.seats, + ) + except SessionExpiredError: + self._session_expired = True + return alternates + except Exception as exc: + logger.debug( + "Date flex check failed for %s (%+d days): %s", + alt_date, day_offset, exc, + ) + + return alternates + def verify_option( self, option: VerifyOption, @@ -108,6 +175,9 @@ def verify_option( On SessionExpiredError, remaining segments are marked UNKNOWN. On individual segment errors, that segment is marked ERROR and verification continues. + + When date_flex is enabled, sold-out segments (seats=0) are + additionally checked on ±1-3 adjacent days. """ result = VerifyResult(option_id=option.option_id, segments=[]) @@ -172,6 +242,15 @@ def verify_option( if dclass: dclass.booking_class = seg_bc + # Date flex: check alternate dates if target date has no availability + if ( + self.date_flex + and seg.target_date + and dclass.seats == 0 + and not self._session_expired + ): + alternates = self._check_alternate_dates(seg, seg_bc) + dclass.alternate_dates = alternates verified.dclass = dclass self._store_cache(seg, dclass) else: diff --git a/tests/test_verify/test_verifier.py b/tests/test_verify/test_verifier.py index 1fad4f3..c7fc75e 100644 --- a/tests/test_verify/test_verifier.py +++ b/tests/test_verify/test_verifier.py @@ -301,3 +301,132 @@ def test_dclass_result_has_booking_class_set(self): ) result = verifier.verify_option(option) assert result.segments[0].dclass.booking_class == "H" + + +class TestDateFlex: + """Tests for the ±3 days date flex feature.""" + + def _make_verifier(self, scraper_results=None, date_flex=True): + scraper = MagicMock() + if scraper_results is not None: + scraper.check_availability.side_effect = scraper_results + cache = MagicMock() + cache.get.return_value = None + return DClassVerifier(scraper=scraper, cache=cache, date_flex=date_flex), scraper, cache + + def test_date_flex_not_triggered_when_available(self): + """If target date has availability, no alternate dates are checked.""" + target_result = _make_dclass(DClassStatus.AVAILABLE, 5, origin="SYD", dest="HKG") + verifier, scraper, cache = self._make_verifier(scraper_results=[target_result]) + + option = VerifyOption( + option_id=1, + segments=[_make_segment("SYD", "HKG", date=datetime.date(2026, 4, 6))], + ) + result = verifier.verify_option(option) + # Only 1 call: the target date. No alternate date queries. + assert scraper.check_availability.call_count == 1 + assert result.segments[0].dclass.seats == 5 + assert result.segments[0].dclass.alternate_dates == [] + + def test_date_flex_checks_alternates_when_sold_out(self): + """If target date is D0, ±3 days are checked.""" + target_result = _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="SYD", dest="HKG") + # Alternates: +1=D0, -1=D3, +2=D5, -2=D0, +3=D0, -3=D0 + alt_results = [ + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="SYD", dest="HKG"), # +1 + _make_dclass(DClassStatus.AVAILABLE, 3, origin="SYD", dest="HKG"), # -1 + _make_dclass(DClassStatus.AVAILABLE, 5, origin="SYD", dest="HKG"), # +2 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="SYD", dest="HKG"), # -2 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="SYD", dest="HKG"), # +3 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="SYD", dest="HKG"), # -3 + ] + verifier, scraper, cache = self._make_verifier( + scraper_results=[target_result] + alt_results, + ) + + option = VerifyOption( + option_id=1, + segments=[_make_segment("SYD", "HKG", date=datetime.date(2026, 4, 6))], + ) + result = verifier.verify_option(option) + # 1 target + 6 alternates = 7 calls + assert scraper.check_availability.call_count == 7 + seg = result.segments[0] + assert seg.dclass.seats == 0 + assert len(seg.dclass.alternate_dates) == 2 # -1 (D3) and +2 (D5) + + # Best alternate should be +2 (D5) + best = seg.dclass.best_alternate + assert best is not None + assert best.seats == 5 + assert best.offset_days == 2 + assert best.date == datetime.date(2026, 4, 8) + + def test_date_flex_disabled_by_default(self): + """With date_flex=False, no alternate dates are checked.""" + target_result = _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="SYD", dest="HKG") + verifier, scraper, cache = self._make_verifier( + scraper_results=[target_result], + date_flex=False, + ) + + option = VerifyOption( + option_id=1, + segments=[_make_segment("SYD", "HKG", date=datetime.date(2026, 4, 6))], + ) + result = verifier.verify_option(option) + assert scraper.check_availability.call_count == 1 + assert result.segments[0].dclass.alternate_dates == [] + + def test_date_flex_stops_on_session_expired(self): + """If session expires during flex checks, remaining are skipped.""" + target_result = _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="SYD", dest="HKG") + verifier, scraper, cache = self._make_verifier( + scraper_results=[ + target_result, + _make_dclass(DClassStatus.AVAILABLE, 3, origin="SYD", dest="HKG"), # +1 + SessionExpiredError("expired"), # -1 fails + ], + ) + + option = VerifyOption( + option_id=1, + segments=[_make_segment("SYD", "HKG", date=datetime.date(2026, 4, 6))], + ) + result = verifier.verify_option(option) + seg = result.segments[0] + # Should have 1 alternate (from +1 day before session expired) + assert len(seg.dclass.alternate_dates) == 1 + assert seg.dclass.alternate_dates[0].offset_days == 1 + + def test_date_flex_with_multiple_segments(self): + """Date flex only applies to sold-out segments.""" + # Seg 1: D9 (no flex needed), Seg 2: D0 (flex), plus alternates for seg 2 + results = [ + _make_dclass(DClassStatus.AVAILABLE, 9, origin="SYD", dest="HKG"), # seg 1 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="HKG", dest="LHR"), # seg 2 + # Flex queries for seg 2: + _make_dclass(DClassStatus.AVAILABLE, 2, origin="HKG", dest="LHR"), # +1 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="HKG", dest="LHR"), # -1 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="HKG", dest="LHR"), # +2 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="HKG", dest="LHR"), # -2 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="HKG", dest="LHR"), # +3 + _make_dclass(DClassStatus.NOT_AVAILABLE, 0, origin="HKG", dest="LHR"), # -3 + ] + verifier, scraper, cache = self._make_verifier(scraper_results=results) + + option = VerifyOption( + option_id=1, + segments=[ + _make_segment("SYD", "HKG", date=datetime.date(2026, 4, 6)), + _make_segment("HKG", "LHR", date=datetime.date(2026, 4, 10)), + ], + ) + result = verifier.verify_option(option) + # Seg 1: 1 call (available, no flex) + # Seg 2: 1 target + 6 alternates = 7 calls + assert scraper.check_availability.call_count == 8 + assert result.segments[0].dclass.alternate_dates == [] + assert len(result.segments[1].dclass.alternate_dates) == 1 + assert result.segments[1].dclass.alternate_dates[0].offset_days == 1