room_cal/room_cal/service_api/_airbnb.py

import asyncio
import base64
import json
import os
import random
import string
import uuid
from datetime import datetime
from zoneinfo import ZoneInfo
import httpx

from dataclasses import dataclass

import logging

from ._scripts import refresh_airbnb_cookies
from room_cal._utils import data_home, get_config, read_env
from ._utils import month_bounds, months_in_range
from .credentials import is_stale, mark_stale
from .types import CalendarEvent, DayInfo, ServiceAuthError, Reservation

_log = logging.getLogger(__name__)


@dataclass
class AirbnbMessage:
  id: str               # global ID (base64)
  created_at_ms: str
  body: str | None      # None for non-text messages (photos, system events, etc.)
  body_translated: str | None  # Airbnb AI translation; None when not provided
  sender_account_id: str


@dataclass
class AirbnbThread:
  thread_id: str              # numeric string, matches Reservation.thread_id
  messages: list[AirbnbMessage]


@dataclass
class AirbnbData:
  suggested_prices: dict[str, int]    # date → Airbnb smart-pricing suggestion (USD)
  calendar_events: list[CalendarEvent]  # holidays and local demand events

_BASE = 'https://www.airbnb.com/api/v3'
_API_KEY = 'd306zoyjsyarp7ifhu67rjxn52tv0t20'
_CLIENT_VERSION = 'ef690fbd6464c4bd9f632ec18087662c4e4f05e1'
_HASHES = {
  'getDLSHostCalendar':               '048e22675ffa53121628d2d07b519b9b71f5dab342a1ceb75aa72d8814a00ea7',
  'EditPanelPricingSettingsMutation': 'e27fed4f08dd4ab4c0267c5d6c0a7e4920730deadf48f92feba9b0d08befafa4',
  'EditPanelAvailabilityMutation':    '2e056e7f98f8dc410a294c649adf719452980eb4d8ea95540fb1297af2637f67',
  'StayHostingDetailsQuery':          '1e81282ae4de44e05bc8b776456644efd4ebcd676451ce7f6c3c5757a1057e02',
  'getNightlyPriceTips':              '07538bf1a5afc8171f1e7443f0007d7f020142fc7ab60df065e7e1edd1a0b29f',
  'HostCalendarListingEventsQuery':   '8bfe5f5572a25fdaff5b5790604e1de3aec5f37731fc505684022d590d06e0a8',
  'ViaductGetThreadAndDataQuery':     'c6aff32757966df45d73811219d4f3a57ae7cc9e047b39bb9ba396cd0c09defa',
  'CreateBulkMessagesMutation':       'b0298188a93a73218fecdd512afb7b47c0df3112a4819c236ecc4c0c10c04bd2',
}

_AUTH_HINT = 'refresh-creds'


class AirbnbAuthError(ServiceAuthError):
  pass


def _encode_listing_id(listing_id: str) -> str:
  return base64.b64encode(f'StayListing:{listing_id}'.encode()).decode()


def _trace_id() -> str:
  return ''.join(random.choices(string.ascii_lowercase + string.digits, k=28))


class _AirbnbClient:
  def __init__(self, cookies: str, listing_id: str, tz: str = 'America/Los_Angeles'):
    self._cookies = cookies
    self._listing_id = listing_id
    self._global_id = _encode_listing_id(listing_id)
    self._tz = tz
    self._cache: dict[str, DayInfo] = {}
    self._fetched_months: set[tuple[int, int]] = set()
    self._raw_months: dict[tuple[int, int], list[dict]] = {}
    self._months_inflight: dict[tuple[int, int], asyncio.Task] = {}
    self._refresh_lock = asyncio.Lock()
    self._refresh_task: asyncio.Task | None = None

  @classmethod
  def from_env(cls) -> '_AirbnbClient':
    env_file = data_home() / 'airbnb.env'
    env = {**read_env(env_file), **os.environ}
    cookies = env.get('AIRBNB_COOKIES', '')
    if not cookies:
      raise ValueError(f'AIRBNB_COOKIES must be set in {env_file}')
    listing_id = get_config().services['airbnb'].room_id
    if not listing_id:
      raise ValueError('airbnb.room_id must be set in config.json')
    return cls(cookies=cookies, listing_id=listing_id)

  def clear_cache(self) -> None:
    self._cache.clear()
    self._fetched_months.clear()
    self._raw_months.clear()

  def _reload_from_env(self) -> None:
    env_file = data_home() / 'airbnb.env'
    env = {**read_env(env_file), **os.environ}
    self._cookies = env.get('AIRBNB_COOKIES', self._cookies)
    self.clear_cache()

  async def _auto_refresh(self) -> None:
    _log.warning('Airbnb session may be expired — launching headless Playwright')
    try:
      refresh_airbnb_cookies()
      self._reload_from_env()
    except ServiceAuthError:
      mark_stale('airbnb')
      raise

  async def _ensure_refreshed(self) -> None:
    async with self._refresh_lock:
      if self._refresh_task is None or self._refresh_task.done():
        self._refresh_task = asyncio.create_task(self._auto_refresh())
      task = self._refresh_task
    await task

  def _headers(self, referer: str, extra: dict | None = None) -> dict[str, str]:
    h = {
      'Cookie': self._cookies,
      'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:152.0) Gecko/20100101 Firefox/152.0',
      'Accept': '*/*',
      'Accept-Language': 'en-US,en;q=0.9',
      'Content-Type': 'application/json',
      'Referer': referer,
      'Sec-Fetch-Mode': 'cors',
      'Sec-Fetch-Dest': 'empty',
      'Sec-Fetch-Site': 'same-origin',
      'X-Airbnb-API-Key': _API_KEY,
      'X-Airbnb-GraphQL-Platform': 'web',
      'X-Airbnb-GraphQL-Platform-Client': 'minimalist-niobe',
      'X-Airbnb-Supports-Airlock-V2': 'true',
      'X-Niobe-Short-Circuited': 'true',
      'X-Client-Version': _CLIENT_VERSION,
      'X-CSRF-Token': '',
      'X-CSRF-Without-Token': '1',
      'x-airbnb-client-trace-id': _trace_id(),
      'x-client-request-id': _trace_id(),
      'x-airbnb-network-log-link': _trace_id(),
    }
    if extra:
      h.update(extra)
    return h

  def _check_auth(self, body: dict) -> None:
    if any(
      e.get('extensions', {}).get('response', {}).get('status') in (400, 401, 403)
      for e in body.get('errors', [])
    ):
      mark_stale('airbnb')
      raise AirbnbAuthError(f'Airbnb session expired — run: {_AUTH_HINT}')

  async def _gql(self, op: str, h: str, variables: dict, referer: str) -> dict:
    url = f'{_BASE}/{op}/{h}'
    params = {
      'operationName': op,
      'locale': 'en',
      'currency': 'USD',
      'variables': json.dumps(variables),
      'extensions': json.dumps({'persistedQuery': {'version': 1, 'sha256Hash': h}}),
    }
    async def _attempt() -> httpx.Response:
      async with httpx.AsyncClient() as client:
        return await client.get(url, params=params, headers=self._headers(referer))
    try:
      r = await _attempt()
    except (httpx.TransportError, OSError, ExceptionGroup) as e:
      raise AirbnbAuthError(f'Airbnb connection failed — session may be expired, run: {_AUTH_HINT}') from e
    if r.status_code in (400, 401, 403):
      await self._ensure_refreshed()
      r = await _attempt()
    if r.status_code in (400, 401, 403):
      mark_stale('airbnb')
      raise AirbnbAuthError(f'Airbnb session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')
    r.raise_for_status()
    body = r.json()
    self._check_auth(body)
    return body

  async def _mutate(self, op: str, h: str, variables: dict, referer: str) -> dict:
    url = f'{_BASE}/{op}/{h}?operationName={op}&locale=en&currency=USD'
    payload = {'operationName': op, 'variables': variables,
               'extensions': {'persistedQuery': {'version': 1, 'sha256Hash': h}}}
    async def _attempt() -> httpx.Response:
      async with httpx.AsyncClient() as client:
        return await client.post(
          url, json=payload,
          headers=self._headers(referer, {'Origin': 'https://www.airbnb.com'}),
        )
    try:
      r = await _attempt()
    except (httpx.TransportError, OSError, ExceptionGroup) as e:
      raise AirbnbAuthError(f'Airbnb connection failed — session may be expired, run: {_AUTH_HINT}') from e
    if r.status_code in (400, 401, 403):
      await self._ensure_refreshed()
      r = await _attempt()
    if r.status_code in (400, 401, 403):
      mark_stale('airbnb')
      raise AirbnbAuthError(f'Airbnb session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')
    r.raise_for_status()
    body = r.json()
    self._check_auth(body)
    return body

  async def _fetch_month_days(self, month: int, year: int) -> list[dict]:
    start, end = month_bounds(month, year)
    variables = {
      'listingId': self._global_id,
      'startDate': start,
      'endDate': end,
      'hostCalendarViewType': 'MONTH_VIEW',
      'lensTypes': ['NOTE', 'PRICE', 'PROMOTION'],
      'timeZone': self._tz,
      'localizationContext': {'timeZone': self._tz, 'firstDayOfWeek': 0},
      'hostCalendarSectionTypes': ['MAIN_VIEW'],
      'readConsistency': 'EVENTUAL',
    }
    referer = f'https://www.airbnb.com/multicalendar/{self._listing_id}'

    def _extract_days(resp: dict) -> list[dict]:
      section = (
        resp.get('data', {}).get('presentation', {})
            .get('hostCalendar', {}).get('sections', {})
            .get('calendarGridViewSection')
      ) or {}
      return section.get('days', [])

    def _has_prices(days: list[dict]) -> bool:
      return any(d.get('priceData', {}).get('price') is not None for d in days)

    resp = await self._gql('getDLSHostCalendar', _HASHES['getDLSHostCalendar'], variables, referer)
    days = _extract_days(resp)
    if not days or not _has_prices(days):
      _log.warning('Airbnb: empty or priceless calendar for %d-%02d; retrying', year, month)
      await asyncio.sleep(1)
      resp = await self._gql('getDLSHostCalendar', _HASHES['getDLSHostCalendar'], variables, referer)
      days = _extract_days(resp)
      if not days or not _has_prices(days):
        _log.warning('Airbnb: still empty/priceless calendar for %d-%02d after retry', year, month)
    return days

  async def _get_month_days(self, month: int, year: int) -> list[dict]:
    key = (month, year)
    if key in self._raw_months:
      return self._raw_months[key]
    if key in self._months_inflight:
      return await self._months_inflight[key]
    task = asyncio.create_task(self._fetch_month_days(month, year))
    self._months_inflight[key] = task
    try:
      result = await task
    finally:
      self._months_inflight.pop(key, None)
    self._raw_months[key] = result
    return result

  async def get_calendar(self, date_from: str, date_to: str) -> dict[str, DayInfo]:
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    month_list = months_in_range(date_from, date_to)
    all_days = await asyncio.gather(*[self._get_month_days(m, y) for m, y in month_list])
    for (month, year), days in zip(month_list, all_days):
      if (month, year) not in self._fetched_months:
        for day in days:
          d = day.get('day', '')
          if d:
            self._cache[d] = DayInfo(
              date=d,
              price=day.get('priceData', {}).get('price'),
              available=day.get('available', False),
              booked=day.get('reservationData') is not None,
              rooms_to_sell=None,
            )
        self._fetched_months.add((month, year))
    return {d: info for d, info in self._cache.items() if date_from <= d <= date_to}

  async def get_reservations(self, date_from: str, date_to: str) -> list[Reservation]:
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    month_list = months_in_range(date_from, date_to)
    all_days = await asyncio.gather(*[self._get_month_days(m, y) for m, y in month_list])
    booked: dict[str, str] = {}
    for days in all_days:
      for day in days:
        d = day.get('day', '')
        res_data = day.get('reservationData')
        if res_data and date_from <= d <= date_to:
          booked[d] = res_data.get('confirmationCode', '')

    if not booked:
      return []

    unique_codes = set(booked.values())

    def _utc_to_local_date(ts: str) -> str:
      dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
      return dt.astimezone(ZoneInfo(self._tz)).date().isoformat()

    async def fetch_details(c: str) -> tuple[str, str | None, str | None, str | None, str | None, str | None]:
      resp = await self._gql(
        'StayHostingDetailsQuery', _HASHES['StayHostingDetailsQuery'],
        {'viewerTimeZoneOffset': 0, 'requestSource': 'RESERVATION_TAB', 'confirmationCode': c},
        f'https://www.airbnb.com/hosting/reservations/details/{c}',
      )
      sections = (
        resp.get('data', {}).get('presentation', {})
            .get('hostingDetails', {}).get('stayHostingDetails', {})
            .get('rootPlacement', [])
      )
      facepile = next((s for s in sections if s.get('sectionId') == 'FACEPILE_SECTION'), None)
      guests = (facepile or {}).get('sectionData', {}).get('guests', [])
      name = guests[0].get('name') if guests else None

      checkinout = next((s for s in sections if s.get('sectionId') == 'CHECK_IN_CHECK_OUT_SECTION'), None)
      ci_out_data = (checkinout or {}).get('sectionData') or {}
      action = ci_out_data.get('checkInAction', {})
      ci_ts = action.get('checkInTime')
      co_ts = action.get('checkOutTime')
      ci = _utc_to_local_date(ci_ts) if ci_ts else None
      co = _utc_to_local_date(co_ts) if co_ts else None
      door_code = (ci_out_data.get('airkeyAction') or {}).get('guestPhoneNumberLastFourDigits')

      # thread_id is in footerPlacement FLOATINGFOOTER_SECTION, not rootPlacement
      details = resp.get('data', {}).get('presentation', {}).get('hostingDetails', {}).get('stayHostingDetails', {})
      footer_sections = details.get('footerPlacement', [])
      footer = next((s for s in footer_sections if s.get('sectionId') == 'FLOATINGFOOTER_SECTION'), None)
      buttons = ((footer or {}).get('sectionData') or {}).get('buttons', [])
      thread_id = None
      for btn in buttons:
        action = btn.get('buttonAction') or {}
        if action.get('__typename') == 'NavigateToMessageGuest':
          url = action.get('url', '')
          m = __import__('re').search(r'/thread/(\d+)', url)
          if m:
            thread_id = m.group(1)
            break

      return c, name, ci, co, door_code, thread_id

    details_list = await asyncio.gather(*[fetch_details(c) for c in unique_codes])
    details = {c: (name, ci, co, door_code, thread_id) for c, name, ci, co, door_code, thread_id in details_list}

    return sorted(
      [
        Reservation(
          service='airbnb',
          reservation_id=c,
          check_in=ci,
          check_out=co,
          guest_name=name,
          door_code=door_code,
          thread_id=thread_id,
          service_data={k: v for k, v in {'door_code': door_code, 'thread_id': thread_id}.items() if v is not None},
          chat_id=thread_id,
        )
        for c, (name, ci, co, door_code, thread_id) in details.items()
        if ci and co
      ],
      key=lambda r: r.check_in,
    )

  async def set_price(self, d: str, price: int) -> None:
    price = int(round(price))
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    if (cached := self._cache.get(d)) and cached.price == price:
      return
    await self._mutate(
      'EditPanelPricingSettingsMutation', _HASHES['EditPanelPricingSettingsMutation'],
      {'input': {
        'listingId': self._listing_id,
        'selectedDateRanges': [{'startDate': d, 'endDate': d}],
        'nightlyPriceAmount': price,
        'turnOnSmartPricing': False,
        'isCalendarV2': True,
      }},
      f'https://www.airbnb.com/multicalendar/{self._listing_id}/edit-selected-dates/{d}/{d}/nightly-price',
    )
    self.clear_cache()
    _log.info('airbnb set_price date=%s price=$%d', d, price)

  async def set_availability(self, d: str, available: bool) -> None:
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    if (cached := self._cache.get(d)) and cached.available == available:
      return
    body = await self._mutate(
      'EditPanelAvailabilityMutation', _HASHES['EditPanelAvailabilityMutation'],
      {'input': {
        'listingId': self._listing_id,
        'selectedDateRanges': [{'startDate': d, 'endDate': d}],
        'actionType': 'OPEN' if available else 'BLOCK',
      }, 'blockNotesEnabled': True},
      f'https://www.airbnb.com/multicalendar/{self._listing_id}/edit-selected-dates/{d}/{d}',
    )
    if errors := body.get('errors'):
      _log.error('airbnb set_availability date=%s errors=%s', d, errors)
      raise AirbnbAuthError(f'Airbnb set_availability failed for {d}: {errors[0].get("message", errors)}')
    self.clear_cache()
    _log.info('airbnb set_availability date=%s available=%s', d, available)

  def _global_thread_id(self, thread_id: str) -> str:
    return base64.b64encode(f'MessageThread:{thread_id}'.encode()).decode()

  def send_ready(self) -> bool:
    return True

  async def get_thread(self, thread_id: str) -> AirbnbThread:
    """Fetch a conversation thread and its messages by numeric thread ID."""
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    resp = await self._gql(
      'ViaductGetThreadAndDataQuery', _HASHES['ViaductGetThreadAndDataQuery'],
      {
        'globalThreadId': self._global_thread_id(thread_id),
        'numRequestedMessages': 50,
        'getThreadState': True,
        'getParticipants': True,
        'getLastReads': True,
        'getInboxFields': True,
        'getInboxOnlyFields': False,
        'getMessageFields': True,
        'getThreadOnlyFields': True,
        'forceUgcTranslation': False,
        'isNovaLite': False,
        'mockThreadIdentifier': None,
        'mockMessageTestIdentifier': None,
        'mockListFooterSlot': None,
        'forceReturnAllReadReceipts': False,
        'originType': 'USER_INBOX',
        'skipOldMessagePreviewFields': False,
      },
      f'https://www.airbnb.com/messages/{thread_id}',
    )
    raw_messages = (
      resp.get('data', {}).get('threadData', {})
          .get('messageData', {}).get('messages', [])
    )
    messages = [
      AirbnbMessage(
        id=m['id'],
        created_at_ms=m['createdAtMs'],
        body=(((m.get('hydratedContent') or {}).get('content') or {}).get('body')),
        body_translated=(((m.get('hydratedContent') or {}).get('content') or {}).get('bodyTranslated')),
        sender_account_id=(m.get('account') or {}).get('accountId', ''),
      )
      for m in raw_messages
    ]
    return AirbnbThread(thread_id=thread_id, messages=messages)

  async def send_message(self, thread_id: str, body: str) -> None:
    """Send a text message to an existing thread."""
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    await self._mutate(
      'CreateBulkMessagesMutation', _HASHES['CreateBulkMessagesMutation'],
      {
        'messageThreadId': thread_id,
        'messages': [{
          'content': {'textContent': {'body': body}},
          'uniqueIdentifier': str(uuid.uuid4()),
          'loggingSource': {'sourceType': '', 'sourceId': ''},
          'loggingExtras': {'entries': []},
        }],
        'actAs': 'PARTICIPANT',
        'businessJustification': {'feature': 'USER_INBOX'},
        'originType': 'USER_INBOX',
        'senderPlatform': 'WEB',
        'skipOldMessagePreviewFields': False,
      },
      f'https://www.airbnb.com/messages/{thread_id}',
    )

  async def get_suggested_prices(self, date_from: str, date_to: str) -> dict[str, int]:
    """Airbnb's per-day smart-pricing suggestions (USD). Only includes days where shouldShowTip=True."""
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    resp = await self._gql(
      'getNightlyPriceTips', _HASHES['getNightlyPriceTips'],
      {
        'listingId': self._global_id,
        'dateRanges': [{'startDate': date_from, 'endDate': date_to}],
      },
      f'https://www.airbnb.com/multicalendar/{self._listing_id}',
    )
    tips = (
      resp.get('data', {}).get('presentation', {})
          .get('pricingGuidance', {}).get('nightlyPriceTip', {})
          .get('nightlyPriceTip', {}).get('discretePriceTips', [])
    )
    return {
      t['date']: t['priceTip']['intValue']
      for t in tips
      if t.get('shouldShowTip') and date_from <= t['date'] <= date_to
    }

  async def get_service_data(self, date_from: str, date_to: str) -> AirbnbData:
    suggested_prices, calendar_events = await asyncio.gather(
      self.get_suggested_prices(date_from, date_to),
      self.get_calendar_events(),
    )
    return AirbnbData(suggested_prices=suggested_prices, calendar_events=calendar_events)

  async def probe(self) -> None:
    from datetime import date
    today = date.today().isoformat()
    await self.get_calendar(today, today)

  async def get_calendar_events(self) -> list[CalendarEvent]:
    """Local events and holidays Airbnb shows on the host calendar (holidays, major events, etc.)."""
    if is_stale('airbnb'):
      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
    resp = await self._gql(
      'HostCalendarListingEventsQuery', _HASHES['HostCalendarListingEventsQuery'],
      {'listingId': self._global_id},
      f'https://www.airbnb.com/multicalendar/{self._listing_id}',
    )
    events = (
      resp.get('data', {}).get('presentation', {})
          .get('hostCalendarListingEvents', {}).get('events', [])
    )
    return [
      CalendarEvent(
        start_date=e['startDate'],
        end_date=e['endDate'],
        name=e['displayName'],
        type=e['type'],
        event_id=e['eventId'],
      )
      for e in events
    ]