room_cal/room_cal/service_api/_scripts.py

import logging
import os
import subprocess
import sys
import threading
from pathlib import Path

from room_cal._utils import get_config

_log = logging.getLogger(__name__)


def _data_home() -> Path:
  override = os.environ.get('ROOM_CAL_DATA_HOME')
  return Path(override) if override else Path.home() / '.local' / 'room_cal'


def _ts_dir() -> Path:
  return Path(__file__).parent / 'typescript'


def _npm_env() -> dict:
  dh = _data_home()
  env = os.environ.copy()
  env.setdefault('ROOM_CAL_DATA_HOME', str(dh))
  env.setdefault('PLAYWRIGHT_BROWSERS_PATH', str(dh / '.playwright-browsers'))
  api_url = get_config().api_url
  if api_url:
    env.setdefault('ROOM_CAL_API_URL', api_url)
  return env


def _ensure_installed(ts_dir: Path) -> None:
  if not (ts_dir / 'node_modules').exists():
    print('=== npm install (service-api) ===')
    subprocess.run(['npm', 'install'], cwd=ts_dir, check=True)
    print()


# One SMS code is valid at a time and Booking.com rate-limits sends regardless of how often we ask,
# so there is no benefit to concurrent booking browser operations — serialize them all.
_booking_lock = threading.Lock()


def refresh_booking_cookies() -> None:
  ts_dir = _ts_dir()
  _ensure_installed(ts_dir)
  _log.info('booking: starting credential refresh')
  with _booking_lock:
    try:
      subprocess.run(['npm', 'run', 'refresh-booking'], cwd=ts_dir, env=_npm_env(), check=True)
    except subprocess.CalledProcessError as e:
      _log.error('booking: credential refresh failed (exit %d)', e.returncode)
      from .types import ServiceAuthError
      raise ServiceAuthError(f'Booking.com credential refresh failed (exit {e.returncode})') from None
  _log.info('booking: credential refresh complete')


def refresh_airbnb_cookies() -> None:
  ts_dir = _ts_dir()
  _ensure_installed(ts_dir)
  _log.info('airbnb: starting credential refresh')
  env = _npm_env()
  env['HEADLESS'] = '1'
  try:
    subprocess.run(['npm', 'run', 'refresh-airbnb'], cwd=ts_dir, env=env, check=True)
  except subprocess.CalledProcessError as e:
    _log.error('airbnb: credential refresh failed (exit %d)', e.returncode)
    from .types import ServiceAuthError
    raise ServiceAuthError(f'Airbnb session expired — re-run without HEADLESS to re-authenticate (exit {e.returncode})') from None
  _log.info('airbnb: credential refresh complete')


def refresh_homestay_cookies() -> None:
  ts_dir = _ts_dir()
  _ensure_installed(ts_dir)
  _log.info('homestay: starting credential refresh')
  env = _npm_env()
  try:
    subprocess.run(['npm', 'run', 'refresh-homestay'], cwd=ts_dir, env=env, check=True)
  except subprocess.CalledProcessError as e:
    _log.error('homestay: credential refresh failed (exit %d)', e.returncode)
    from .types import ServiceAuthError
    raise ServiceAuthError(f'Homestay session expired — re-run without HEADLESS to re-authenticate (exit {e.returncode})') from None
  _log.info('homestay: credential refresh complete')


def trigger_booking_chat_auth() -> None:
  """Run booking-2fa; raises ServiceAuthError on failure. Safe to call from within the server."""
  ts_dir = _ts_dir()
  _ensure_installed(ts_dir)
  _log.info('booking: starting standalone messaging 2FA prime')
  with _booking_lock:
    result = subprocess.run(['npm', 'run', 'booking-2fa'], cwd=ts_dir, env=_npm_env())
  if result.returncode != 0:
    _log.error('booking: booking-2fa failed (exit %d)', result.returncode)
    from .types import ServiceAuthError
    raise ServiceAuthError(f'booking-2fa failed (exit {result.returncode}) — SMS gateway may not be configured')
  _log.info('booking: messaging 2FA prime complete')


def _trigger_booking_chat_auth_cli() -> None:
  """CLI entry point for `uv run booking-2fa`."""
  ts_dir = _ts_dir()
  _ensure_installed(ts_dir)
  result = subprocess.run(['npm', 'run', 'booking-2fa', '--'] + sys.argv[1:], cwd=ts_dir, env=_npm_env())
  sys.exit(result.returncode)


def refresh_service(key: str) -> None:
  """Trigger the automatic credential refresh for the named service. Raises ServiceAuthError on failure."""
  handlers = {
    'booking': refresh_booking_cookies,
    'airbnb': refresh_airbnb_cookies,
    'homestay': refresh_homestay_cookies,
  }
  fn = handlers.get(key)
  if fn is None:
    from .types import ServiceAuthError
    raise ServiceAuthError(f'No automatic refresh handler for service: {key}')
  fn()


def refresh_cookies() -> None:
  ts_dir = _ts_dir()
  _ensure_installed(ts_dir)
  result = subprocess.run(['npm', 'run', 'refresh-all', '--'] + sys.argv[1:], cwd=ts_dir, env=_npm_env())
  sys.exit(result.returncode)