Spaces:
Paused
Paused
| """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 | |