room_cal/scripts/har_audit.py

"""Audit HAR files for service_api client drift.

Filters captured browser traffic down to the API calls we care about, compares
identifiers (GQL hashes, version headers) against current client constants, and
lists any operations that weren't captured — with a hint for what action
triggers each one.

Usage:
  uv run python scripts/har_audit.py path/to/*.har

One or more HAR files may be passed. Platform is detected from request domains;
a single HAR may contain both platforms.
"""

import json
import re
import sys
import urllib.parse
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from room_cal.service_api._airbnb import DRIFT_MANIFEST as _AIRBNB_MANIFEST
from room_cal.service_api._booking import DRIFT_MANIFEST as _BOOKING_MANIFEST

# Each client module owns its DRIFT_MANIFEST — imported directly from
# _airbnb.py / _booking.py, so there is no separate mirror to drift out of
# sync. A new service gets drift-checking by adding its own import above.

_AIRBNB_DOMAIN = _AIRBNB_MANIFEST['domain']
_AIRBNB_API_KEY = _AIRBNB_MANIFEST['api_key']
_AIRBNB_CLIENT_VERSION = _AIRBNB_MANIFEST['client_version']
_AIRBNB_OPS: dict[str, dict] = _AIRBNB_MANIFEST['ops']

_BOOKING_DOMAIN = _BOOKING_MANIFEST['domain']
_BOOKING_ENDPOINTS: dict[str, dict] = _BOOKING_MANIFEST['endpoints']


# ── HAR helpers ────────────────────────────────────────────────────────────

def _req_headers(entry: dict) -> dict[str, str]:
  return {h['name'].lower(): h['value'] for h in entry['request']['headers']}

def _req_body(entry: dict) -> dict | None:
  text = (entry['request'].get('postData') or {}).get('text', '')
  try:
    return json.loads(text) if text else None
  except (json.JSONDecodeError, ValueError):
    return None

def _req_variables(entry: dict) -> dict | None:
  """The GQL `variables` object for a request, whether sent as a GET query param or in a POST body."""
  if (body := _req_body(entry)) is not None:
    return body.get('variables')
  qs = urllib.parse.urlparse(entry['request']['url']).query
  params = urllib.parse.parse_qs(qs)
  raw = params.get('variables', [None])[0]
  if raw is None:
    return None
  try:
    return json.loads(raw)
  except (json.JSONDecodeError, ValueError):
    return None

def _short(h: str) -> str:
  """Truncate a hash/token for display: first 16 chars + ellipsis."""
  return h[:16] + '...' if len(h) > 20 else h


# ── Airbnb ─────────────────────────────────────────────────────────────────

def _collect_airbnb(entries: list[dict]) -> dict[str, list[dict]]:
  """Return {operation_name: [entry, ...]} for Airbnb /api/v3 calls."""
  found: dict[str, list[dict]] = {}
  for e in entries:
    url = e['request']['url']
    m = re.search(r'/api/v3/([A-Za-z]+)/([0-9a-f]{64})', url)
    if m:
      op = m.group(1)
      found.setdefault(op, []).append(e)
  return found


def _audit_airbnb(entries: list[dict], har_name: str) -> bool:
  found = _collect_airbnb(entries)
  if not found:
    return False

  print(f'\n{"═" * 70}')
  print(f'  Airbnb  |  {har_name}')
  print(f'{"═" * 70}')

  # Operations
  print('\n  Operations:\n')
  col = max(len(op) for op in _AIRBNB_OPS) + 2

  all_ok = True
  for op, spec in _AIRBNB_OPS.items():
    if op not in found:
      print(f'    —  {op:<{col}} not captured')
      print(f'       {"":>{col}}   → {spec["hint"]}')
      continue

    entry = found[op][0]
    url = entry['request']['url']
    m = re.search(r'/api/v3/[A-Za-z]+/([0-9a-f]{64})', url)
    har_hash = m.group(1) if m else '?'
    expected = spec['hash']

    if har_hash == expected:
      print(f'    ✓  {op:<{col}} hash matches')
    else:
      all_ok = False
      print(f'    ✗  {op:<{col}} STALE hash')
      print(f'       {"client:":<{col+5}} {expected}')
      print(f'       {"HAR:":<{col+5}} {har_hash}   ← update _HASHES[\'{op}\']')

    expected_keys = spec.get('var_keys')
    if expected_keys is not None:
      # Union keys across every captured call of this op, not just the first —
      # the real app sends different variable shapes for the same op depending
      # on which page/action triggered it (e.g. initial calendar load vs.
      # month-by-month paging), so a single entry isn't representative.
      har_key_sets = [
        set(v.keys()) for e in found[op]
        if isinstance(v := _req_variables(e), dict)
      ]
      if not har_key_sets:
        all_ok = False
        print(f'    ✗  {op:<{col}} variables not found in any HAR entry')
      else:
        har_keys = set().union(*har_key_sets)
        missing = expected_keys - har_keys   # client sends, no HAR call does — likely dropped
        extra = har_keys - expected_keys     # some HAR call has, client doesn't — likely added/renamed
        if missing or extra:
          all_ok = False
          print(f'    ✗  {op:<{col}} variable key drift')
          if missing:
            print(f'       {"client sends, HAR lacks:":<{col+5}} {sorted(missing)}   ← dropped/renamed; remove or fix in client')
          if extra:
            print(f'       {"HAR has, client lacks:":<{col+5}} {sorted(extra)}   ← new/renamed field; add to client')
        else:
          print(f'    ✓  {op:<{col}} variable keys match')

  # Global headers (from any /api/v3 call)
  sample_entry = next(iter(found.values()))[0]
  hdrs = _req_headers(sample_entry)

  print('\n  Global headers:\n')
  hcol = max(len('X-Airbnb-API-Key'), len('X-Client-Version')) + 2

  api_key = hdrs.get('x-airbnb-api-key', '')
  if api_key == _AIRBNB_API_KEY:
    print(f'    ✓  {"X-Airbnb-API-Key":<{hcol}} matches')
  else:
    all_ok = False
    print(f'    ✗  {"X-Airbnb-API-Key":<{hcol}} MISMATCH')
    print(f'       client: {_AIRBNB_API_KEY}')
    print(f'       HAR:    {api_key}   ← update _API_KEY')

  cv = hdrs.get('x-client-version', '')
  if cv == _AIRBNB_CLIENT_VERSION:
    print(f'    ✓  {"X-Client-Version":<{hcol}} matches')
  else:
    all_ok = False
    print(f'    ✗  {"X-Client-Version":<{hcol}} stale')
    print(f'       client: {_AIRBNB_CLIENT_VERSION}')
    print(f'       HAR:    {cv}   ← update _CLIENT_VERSION')

  missing = [op for op in _AIRBNB_OPS if op not in found]
  if not missing:
    print('\n  All operations captured.')
  else:
    print(f'\n  {len(missing)} operation(s) not captured — to get full coverage:')
    printed_hints: set[str] = set()
    for op in missing:
      hint = _AIRBNB_OPS[op]['hint']
      if hint not in printed_hints:
        print(f'    • {hint}')
        printed_hints.add(hint)

  return True


# ── Booking.com ────────────────────────────────────────────────────────────

def _collect_booking(entries: list[dict]) -> dict[str, list[dict]]:
  """Return {endpoint_key: [entry, ...]} for Booking.com API calls."""
  found: dict[str, list[dict]] = {}
  for e in entries:
    url = e['request']['url']
    if _BOOKING_DOMAIN not in url:
      continue
    body = _req_body(e)
    body_op = (body or {}).get('operationName')
    for key, spec in _BOOKING_ENDPOINTS.items():
      if spec['url_fragment'] not in url:
        continue
      op_name = spec.get('operation_name')
      if op_name and body_op != op_name:
        continue
      found.setdefault(key, []).append(e)
  return found


def _audit_booking(entries: list[dict], har_name: str) -> bool:
  found = _collect_booking(entries)
  if not found:
    return False

  print(f'\n{"═" * 70}')
  print(f'  Booking.com  |  {har_name}')
  print(f'{"═" * 70}')

  print('\n  Endpoints:\n')
  col = max(len(k) for k in _BOOKING_ENDPOINTS) + 2

  all_ok = True
  for key, spec in _BOOKING_ENDPOINTS.items():
    if key not in found:
      print(f'    —  {key:<{col}} not captured')
      print(f'       {"":>{col}}   → {spec["hint"]}')
      continue

    # Check every captured call of this endpoint, not just the first — the
    # same endpoint can show more than one header value within one HAR (e.g.
    # a stale value from a frontend bundle reload mid-capture), so a mismatch
    # only means something if the client's value never shows up at all.
    expected_headers = spec.get('headers', {})
    all_hdrs = [_req_headers(e) for e in found[key]]

    mismatches: list[tuple[str, str, list[str]]] = []
    for hname, hexpected in expected_headers.items():
      observed = [h.get(hname, '') for h in all_hdrs]
      if hexpected not in observed:
        mismatches.append((hname, hexpected, sorted(set(observed))))

    if not mismatches:
      print(f'    ✓  {key:<{col}} found')
    else:
      all_ok = False
      print(f'    ✗  {key:<{col}} found with stale headers')
      src = spec.get('source', 'the client headers')
      for hname, hexpected, observed in mismatches:
        print(f'       {hname}')
        print(f'         client:   {hexpected}')
        print(f'         HAR saw:  {observed}   ← update {src}')

  missing = [k for k in _BOOKING_ENDPOINTS if k not in found]
  if not missing:
    print('\n  All endpoints captured.')
  else:
    print(f'\n  {len(missing)} endpoint(s) not captured — to get full coverage:')
    printed_hints: set[str] = set()
    for key in missing:
      hint = _BOOKING_ENDPOINTS[key]['hint']
      if hint not in printed_hints:
        print(f'    • {hint}')
        printed_hints.add(hint)

  return True


# ── Entry point ────────────────────────────────────────────────────────────

def audit_har(path: Path) -> None:
  try:
    har = json.loads(path.read_text(encoding='utf-8'))
  except Exception as e:
    print(f'Error reading {path}: {e}', file=sys.stderr)
    return

  entries = har.get('log', {}).get('entries', [])
  if not entries:
    print(f'No entries in {path.name}', file=sys.stderr)
    return

  # Detect which platforms appear in this HAR
  domains = {
    e['request']['url'].split('/')[2]
    for e in entries
    if e['request']['url'].startswith('http')
  }

  name = path.name
  found_any = False
  if _AIRBNB_DOMAIN in domains:
    found_any |= _audit_airbnb(entries, name)
  if _BOOKING_DOMAIN in domains:
    found_any |= _audit_booking(entries, name)

  if not found_any:
    known = f'{_AIRBNB_DOMAIN}, {_BOOKING_DOMAIN}'
    print(f'\n{path.name}: no recognised API calls (expected domains: {known})')


def main() -> None:
  paths = sys.argv[1:]
  if not paths:
    print(__doc__)
    sys.exit(1)
  for p in paths:
    audit_har(Path(p))
  print()


if __name__ == '__main__':
  main()