"""Evidence-backed coverage contracts for offers and live seller comparisons. The comparison engine can calculate an answer exactly *for the price and offers it is given*. That is different from proving that every seller and every bank offer has been checked. This module makes that boundary executable: a flight may only be labelled a verified best offer when the current seller table is fully mapped and the approved coverage manifest says every in-scope OTA and issuer has been read successfully within the configured freshness window. An absent, stale, partial, or manually edited manifest is deliberately a coverage failure, never an implicit assertion of completeness. The manifest is produced by dataops/approve.py after a clean crawl and human approval. """ from __future__ import annotations import json import os from datetime import date, datetime, timedelta, timezone from typing import Dict, Iterable, List, Optional MANIFEST_SCHEMA_VERSION = 1 MANIFEST_PATH = os.getenv( "FLIGHT_COVERAGE_MANIFEST", os.path.join(os.path.dirname(__file__), "offers_data", "coverage_manifest.json"), ) # Flight prices may change minute-to-minute, but a bank/OTA offer verification # has to be from the same operational day before we make a completeness claim. MAX_COVERAGE_AGE_HOURS = int(os.getenv("FLIGHT_COVERAGE_MAX_AGE_HOURS", "24")) _IST = timezone(timedelta(hours=5, minutes=30)) def today_ist() -> str: return datetime.now(_IST).date().isoformat() def _parse_day(raw: object) -> Optional[date]: if not isinstance(raw, str) or not raw: return None try: # A manifest may contain a date or a full ISO timestamp. The date is # sufficient because dataops runs as a daily operational check. return date.fromisoformat(raw[:10]) except ValueError: return None def _fresh(entry: object, as_of: str, max_age_hours: int) -> bool: if not isinstance(entry, dict) or entry.get("status") != "verified": return False checked = _parse_day(entry.get("checked_at")) now = _parse_day(as_of) if checked is None or now is None: return False # Date-granularity dataops records cannot use a fractional hour precisely; # accept only checks made today when configured below a day. allowed_days = 0 if max_age_hours < 24 else max_age_hours // 24 return checked <= now <= checked + timedelta(days=allowed_days) def load_manifest(path: Optional[str] = None) -> Dict: """Return an untrusted manifest as data. Malformed/missing = no coverage.""" try: with open(path or MANIFEST_PATH, encoding="utf-8") as f: value = json.load(f) return value if isinstance(value, dict) else {} except (OSError, ValueError, TypeError): return {} def flight_channel_keys() -> List[str]: from channels import FLIGHT_CHANNELS return [str(ch["key"]) for ch in FLIGHT_CHANNELS] def supported_issuers() -> List[str]: """Issuer labels actually represented by individual card products. Composite/placeholder labels cannot be verified as a bank and are excluded from the contractual denominator; each actual card issuer remains in it. """ from card_catalogue import all_cards excluded = {"", "Other bank", "Any bank", "BoB/Federal/SBM"} return sorted({str(card.issuer) for card in all_cards() if str(card.issuer) not in excluded}) def channel_requirements(category: str, brand_key: Optional[str] = None) -> List[str]: """Offer-bearing merchant keys for the channels this purchase compares. The physical/direct row has no platform key. When the resolved merchant is known it becomes that row's requirement; an unknown direct merchant remains an explicit coverage gap rather than being treated as a verified zero. """ from channels import PLATFORM_BRANDS, _channels_for _kind, plan = _channels_for(category, brand_key) if not plan: return [brand_key] if brand_key else [] keys = {str(ch.get("brand_key") or "") for ch in plan if ch.get("brand_key")} if brand_key and brand_key not in PLATFORM_BRANDS: keys.add(str(brand_key)) return sorted(k for k in keys if k) def issuers_for_cards(card_ids: Iterable[str]) -> List[str]: from card_catalogue import CATALOGUE_BY_ID supported = set(supported_issuers()) return sorted({CATALOGUE_BY_ID[cid].issuer for cid in card_ids if cid in CATALOGUE_BY_ID and CATALOGUE_BY_ID[cid].issuer in supported}) def requirements() -> Dict[str, List[str]]: return {"channels": flight_channel_keys(), "issuers": supported_issuers()} def _required_failures( manifest: Dict, group: str, required: Iterable[str], as_of: str, max_age_hours: int, ) -> List[str]: records = manifest.get(group) if isinstance(manifest.get(group), dict) else {} failures: List[str] = [] for key in required: if not _fresh(records.get(key), as_of, max_age_hours): failures.append(key) return failures def assess_flight_coverage( sellers: Iterable[Dict], *, live: bool, held_card_ids: Optional[Iterable[str]] = None, manifest: Optional[Dict] = None, as_of: Optional[str] = None, max_age_hours: int = MAX_COVERAGE_AGE_HOURS, ) -> Dict: """Return the machine-readable truthfulness guard for one seller quote. This assesses the *declared India flight coverage scope* (all supported flight OTAs and, when supplied, only the customer's held-card issuers), plus whether the particular live seller table has an unmapped seller. It never treats a missing seller as a price match and never lets the recommendation claim completeness. """ checked_at = as_of or today_ist() m = manifest if isinstance(manifest, dict) else load_manifest() schema_ok = m.get("schema_version") == MANIFEST_SCHEMA_VERSION req = requirements() rows = [s for s in sellers if isinstance(s, dict)] priced_rows = [] for row in rows: try: positive_price = float(row.get("price_inr") or 0) > 0 except (TypeError, ValueError): positive_price = False if positive_price and row.get("price_comparable") is not False: priced_rows.append(row) anomalous = sorted({str(s.get("seller") or "Unknown seller") for s in rows if s.get("price_comparable") is False}) unmapped = sorted({str(s.get("seller") or "Unknown seller") for s in rows if not s.get("channel_key")}) mapped_keys = sorted({str(s.get("channel_key")) for s in rows if s.get("channel_key")}) unknown_channel_keys = sorted(set(mapped_keys) - set(req["channels"])) missing_channels = _required_failures(m, "channels", req["channels"], checked_at, max_age_hours) supplied_card_ids = (list(dict.fromkeys(str(card_id) for card_id in held_card_ids if str(card_id))) if held_card_ids is not None else None) from card_catalogue import CATALOGUE_BY_ID recognized_card_ids = ([card_id for card_id in supplied_card_ids if card_id in CATALOGUE_BY_ID] if supplied_card_ids is not None else []) unrecognized_card_ids = ([card_id for card_id in supplied_card_ids if card_id not in CATALOGUE_BY_ID] if supplied_card_ids is not None else []) required_issuers = (issuers_for_cards(recognized_card_ids) if supplied_card_ids is not None else req["issuers"]) missing_issuers = _required_failures( m, "issuers", required_issuers, checked_at, max_age_hours) channel_records = m.get("channels") if isinstance(m.get("channels"), dict) else {} fresh_channel_keys = { key for key in req["channels"] if _fresh(channel_records.get(key), checked_at, max_age_hours) } offer_checked_rows = [row for row in priced_rows if str(row.get("channel_key") or "") in fresh_channel_keys] issuer_records = m.get("issuers") if isinstance(m.get("issuers"), dict) else {} verified_card_offer_count = sum( 1 for card_id in recognized_card_ids if _fresh(issuer_records.get(CATALOGUE_BY_ID[card_id].issuer), checked_at, max_age_hours) ) reasons: List[str] = [] if not live: reasons.append("live seller quote unavailable") if not schema_ok: reasons.append("approved coverage manifest unavailable or incompatible") if not rows: reasons.append("seller table empty") if anomalous: reasons.append("one or more seller prices failed itinerary consistency checks") if unmapped: reasons.append("one or more live sellers are not mapped to a supported OTA") if unknown_channel_keys: reasons.append("seller table contains unknown channel keys") if missing_channels: reasons.append("one or more supported OTAs lack a fresh verified crawl") if missing_issuers: reasons.append("one or more represented card issuers lack a fresh verified crawl") if unrecognized_card_ids: reasons.append("one or more supplied held cards are not recognized by the card catalogue") # OTA/query evidence is separable from issuer evidence. The mobile app keeps # the wallet on-device, so it combines this OTA result with the returned # issuer freshness list locally instead of uploading card selections. ota_scope_complete = bool( live and schema_ok and rows and not anomalous and not unmapped and not unknown_channel_keys and not missing_channels ) verified_issuers = sorted( issuer for issuer in req["issuers"] if _fresh(issuer_records.get(issuer), checked_at, max_age_hours) ) complete = not reasons return { "scope": ("India flight offers: every supported OTA and supplied held-card issuer" if supplied_card_ids is not None else "India flight offers: every supported OTA and represented card issuer"), "as_of": checked_at, "max_age_hours": max_age_hours, "manifest_schema_ok": schema_ok, "ota_scope_complete": ota_scope_complete, "verified_issuers": verified_issuers, "live_seller_count": len(rows), "priced_seller_count": len(priced_rows), "mapped_seller_count": sum(1 for row in priced_rows if row.get("channel_key")), "offer_checked_seller_count": len(offer_checked_rows), "held_card_count": len(supplied_card_ids or []), "recognized_held_card_count": len(recognized_card_ids), "verified_card_offer_count": verified_card_offer_count, "unrecognized_card_ids": unrecognized_card_ids, "required_issuers": required_issuers, "mapped_channel_keys": mapped_keys, "unmapped_sellers": unmapped, "anomalous_sellers": anomalous, "unknown_channel_keys": unknown_channel_keys, "unverified_channels": missing_channels, "unverified_issuers": missing_issuers, "complete": complete, # The UI and API must key any “best/optimal” language off this one # boolean, rather than interpreting an available comparison as proof. "recommendation_allowed": complete, "reasons": reasons, } def assess_offer_coverage( category: Optional[str] = None, *, brand_key: Optional[str] = None, held_card_ids: Optional[Iterable[str]] = None, manifest: Optional[Dict] = None, as_of: Optional[str] = None, max_age_hours: int = MAX_COVERAGE_AGE_HOURS, ) -> Dict: """Audit both arithmetic accuracy and evidence-source breadth. ``accuracy_gate_passed`` means every offer allowed into a rupee total has a URL, verbatim evidence and all material terms represented. It does *not* mean the public internet contains no additional offer. ``market_complete`` therefore stays false until contractual/official feeds provide a finite, auditable denominator for targeted as well as public offers. """ from offers import accuracy_reasons, all_active, is_priceable checked_at = as_of or today_ist() active = all_active(checked_at) priceable = [o for o in active if is_priceable(o)] display_only = [o for o in active if not is_priceable(o)] invalid_priceable = [ {"id": o.get("id"), "reasons": accuracy_reasons(o)} for o in priceable if accuracy_reasons(o) ] m = manifest if isinstance(manifest, dict) else load_manifest() schema_ok = m.get("schema_version") == MANIFEST_SCHEMA_VERSION required_merchants = channel_requirements(category, brand_key) if category else [] required_issuers = (issuers_for_cards(held_card_ids or []) if held_card_ids is not None else supported_issuers()) missing_merchants = _required_failures( m, "merchants", required_merchants, checked_at, max_age_hours) missing_issuers = _required_failures( m, "issuers", required_issuers, checked_at, max_age_hours) represented_issuers = set() for off in active: issuer = str(off.get("applies_to_issuer") or "") if issuer: represented_issuers.add(issuer) supported = supported_issuers() sources = {str(o.get("source") or "") for o in active if o.get("source")} merchants = {str(o.get("merchant_key") or "") for o in active if o.get("merchant_key")} from merchants import CATEGORIES, MERCHANT_MAP brand_categories: Dict[str, set] = {} for mapped_category, mapped_brand in MERCHANT_MAP.values(): if mapped_brand: brand_categories.setdefault(str(mapped_brand), set()).add(str(mapped_category)) category_counts = {key: {"priceable": 0, "display_only": 0} for key in CATEGORIES} for off in active: cats = set(off.get("categories") or []) or brand_categories.get( str(off.get("merchant_key") or ""), set()) bucket = "priceable" if is_priceable(off) else "display_only" for cat in cats: if cat in category_counts: category_counts[cat][bucket] += 1 declared_complete = bool(schema_ok and not missing_merchants and not missing_issuers) accuracy_ok = not invalid_priceable reasons: List[str] = [] if not schema_ok: reasons.append("approved source-coverage manifest unavailable or incompatible") if missing_merchants: reasons.append("one or more compared merchants/platforms lack a fresh verified crawl") if missing_issuers: reasons.append("one or more relevant issuers lack a fresh verified crawl") if invalid_priceable: reasons.append("one or more priceable offers failed the material-term accuracy gate") # Even a clean watchlist crawl does not see private/account-targeted offers. reasons.append("private, targeted and checkout-only offers have no complete public denominator") return { "scope": (f"Verified public offers for {category}" if category else "Verified public offers across supported categories"), "as_of": checked_at, "max_age_hours": max_age_hours, "accuracy_gate_passed": accuracy_ok, "priceable_offers": len(priceable), "display_only_offers": len(display_only), "active_sources": len(sources), "merchants_with_active_offers": len(merchants), "supported_issuers": len(supported), "issuers_with_active_offers": len(set(supported) & represented_issuers), "issuers_without_active_offers": sorted(set(supported) - represented_issuers), "category_counts": category_counts, "categories_without_priceable_offers": sorted( key for key, counts in category_counts.items() if counts["priceable"] == 0), "required_merchants": required_merchants, "required_issuers": required_issuers, "unverified_merchants": missing_merchants, "unverified_issuers": missing_issuers, "invalid_priceable_offers": invalid_priceable, "manifest_schema_ok": schema_ok, "declared_sources_complete": declared_complete, "market_complete": False, "recommendation_label": "Best checked deal", "reasons": reasons, "note": ("Only fresh, source-backed offers with representable terms enter totals. " "Missing or partially scoped offers remain visible but worth ₹0; full-market " "coverage is not claimed."), } def manifest_from_dataops_report(report: Dict, *, approved_at: Optional[str] = None) -> Dict: """Build the manifest dataops may publish after human offer approval. A clean run proves only sources it actually read. Sources outside the watchlist remain absent and therefore fail closed in ``assess``. A verified zero-offer page is legitimate coverage; unread and suspicious pages are not. This function contains no network access and is deterministic so it can be unit-tested from a saved report. """ stamp = approved_at or str(report.get("generated") or today_ist())[:10] result: Dict = { "schema_version": MANIFEST_SCHEMA_VERSION, "generated_at": stamp, "scope": "Supported merchant and issuer offer sources across all categories", "channels": {}, "merchants": {}, "issuers": {}, "source_report": { "generated": report.get("generated"), "read_ok": bool(report.get("read_ok")), "pages_unread": int(report.get("pages_unread") or 0), "suspicious_zeroes": len(report.get("suspicious_zeroes") or []), }, } # A globally unclean report must not mint any fresh verification stamps. clean_run = bool(report.get("read_ok")) by_channel: Dict[str, List[Dict]] = {} by_merchant: Dict[str, List[Dict]] = {} by_issuer: Dict[str, List[Dict]] = {} for source in report.get("by_source") or []: if not isinstance(source, dict): continue merchant = str(source.get("merchant_key") or "") issuer = str(source.get("issuer") or "") if merchant: by_channel.setdefault(merchant, []).append(source) by_merchant.setdefault(merchant, []).append(source) if issuer: by_issuer.setdefault(issuer, []).append(source) def record(rows: List[Dict]) -> Dict: pages_read = sum(int(r.get("pages_read") or 0) for r in rows) unread = sum(int(r.get("unread_pages") or 0) for r in rows) suspicious = sum(int(r.get("suspicious_pages") or 0) for r in rows) ok = clean_run and bool(rows) and pages_read > 0 and unread == 0 and suspicious == 0 return { "status": "verified" if ok else "unverified", "checked_at": stamp if ok else None, "sources": [str(r.get("url") or "") for r in rows], "pages_read": pages_read, "unread_pages": unread, "suspicious_pages": suspicious, } for key in flight_channel_keys(): result["channels"][key] = record(by_channel.get(key, [])) # Record every merchant/platform the approved report actually attempted. # Category-specific audits compare their required channel keys against this # map; an absent key is an explicit gap, never a verified zero. for key, rows in sorted(by_merchant.items()): result["merchants"][key] = record(rows) for issuer in supported_issuers(): result["issuers"][issuer] = record(by_issuer.get(issuer, [])) return result