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
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 _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}\']')

  # 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

    entry = found[key][0]
    hdrs = _req_headers(entry)
    expected_headers = spec.get('headers', {})

    mismatches: list[tuple[str, str, str]] = []
    for hname, hexpected in expected_headers.items():
      hactual = hdrs.get(hname, '')
      if hactual != hexpected:
        mismatches.append((hname, hexpected, hactual))

    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, hactual in mismatches:
        print(f'       {hname}')
        print(f'         client: {hexpected}')
        print(f'         HAR:    {hactual}   ← 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()