room_cal/room_cal/service_api/_booking.py

import asyncio
import json
import logging
import os
import random
import time
import warnings
from dataclasses import dataclass
from room_cal._utils import local_today as _local_today
from urllib.parse import parse_qs, urlparse

import httpx

from ._scripts import refresh_booking_cookies, trigger_booking_chat_auth
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, set_credentials
from .types import DayInfo, ServiceAuthError, Reservation

_log = logging.getLogger(__name__)


def _is_past(d: str) -> bool:
  return d < _local_today().isoformat()


_BASE = 'https://admin.booking.com'
_AUTH_HINT = 'refresh-creds'

_UPDATE_MUTATION = """mutation updateRoomDateMutation($input: PartnerAvailabilityUpdateInput) {
  partnerAvailabilityMutations {
    partnerAvailabilityUpdate(input: $input) {
      success
      hotel {
        rooms {
          roomId
          dates {
            date
            hidden
            roomsToSell { changed value __typename }
            status { changed value __typename }
            statusReason { changed value __typename }
            __typename
          }
          rates {
            rateId
            restrictionAdvice { date minStayThrough __typename }
            dates {
              date status statusReason
              roomsToSell { changed value __typename }
              priceSingleUse { changed value __typename }
              price { occupancy value __typename }
              minStayThrough { changed value __typename }
              minStayArrival { changed value __typename }
              minAdvanceRes { changed value __typename }
              maxStayThrough { changed value __typename }
              maxStayArrival { changed value __typename }
              maxAdvanceRes { changed value __typename }
              lengthOfStay { changed value __typename }
              exactStayArrival { changed value __typename }
              closedOnDeparture { changed value __typename }
              closedOnArrival { changed value __typename }
              closed
              __typename
            }
            __typename
          }
          __typename
        }
        __typename
      }
      result {
        warnings { message dates code __typename }
        untilDate success fromDate
        errors { dates message __typename }
        __typename
      }
      __typename
    }
    __typename
  }
}"""

_CONVERSATION_MESSAGES_QUERY = """query conversationMessages($input: ChatSingleConversationInput!) {
  chatConversation {
    singleConversation(input: $input) {
      conversation {
        flags { isReplyNeeded __typename }
        __typename
      }
      messages {
        messageId created content type
        sender { participantType __typename }
        attributes {
          ... on SelfServiceEventDetailChatEntity {
            eventDetail { body __typename }
            __typename
          }
        }
        __typename
      }
      __typename
    }
    __typename
  }
}"""

_POST_FREE_TEXT_MUTATION = """mutation postFreeTextMessage($input: ChatPostFreeTextMessageInput!) {
  chatMessage {
    postFreeTextMessage(input: $input) {
      message {
        messageId created content type
        sender { participantType __typename }
        __typename
      }
      __typename
    }
    __typename
  }
}"""


class BookingAuthError(ServiceAuthError):
  pass


@dataclass
class BookingMessage:
  id: str
  created_at: str
  body: str | None
  sender: str            # 'GUEST' | 'PARTNER'
  body_translated: str | None = None


@dataclass
class BookingThread:
  reservation_id: str
  messages: list[BookingMessage]
  is_reply_needed: bool = False


def _page_view_id() -> str:
  return ''.join(random.choices('0123456789abcdef', k=16))


class _BookingClient:

  def __init__(
    self,
    cookies: str,
    session: str,
    hotel_id: int,
    account_id: int,
    room_id: int,
    rate_id: str,
    csrf_token: str | None = None,
    lang: str = 'xu',
  ):
    self._cookies = cookies
    self._session = session
    self._hotel_id = hotel_id
    self._account_id = account_id
    self._room_id = room_id
    self._rate_id = rate_id
    self._csrf_token = csrf_token
    self._lang = lang
    self._cache: dict[str, DayInfo] = {}
    self._fetched_months: set[tuple[int, int]] = set()
    self._month_data: dict[tuple[int, int], dict[str, DayInfo]] = {}
    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) -> '_BookingClient':
    env_file = data_home() / 'booking.env'
    env = {**read_env(env_file), **os.environ}
    required_env = ['BOOKING_COOKIES', 'BOOKING_SESSION']
    missing_env = [k for k in required_env if not env.get(k)]
    if missing_env:
      raise ValueError(f'Missing in {env_file}: {", ".join(missing_env)}')
    svc = get_config().services['booking']
    missing_ids = [k for k in ('hotel_id', 'account_id', 'room_id', 'rate_id') if not getattr(svc, k)]
    if missing_ids:
      raise ValueError(f'Missing from config.json booking: {", ".join(missing_ids)}')
    return cls(
      cookies=env['BOOKING_COOKIES'],
      session=env['BOOKING_SESSION'],
      hotel_id=int(svc.hotel_id),
      account_id=int(svc.account_id),
      room_id=int(svc.room_id),
      rate_id=svc.rate_id,
      csrf_token=env.get('BOOKING_CSRF_TOKEN'),
    )

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

  def _reload_from_env(self) -> None:
    env_file = data_home() / 'booking.env'
    env = {**read_env(env_file), **os.environ}
    self._cookies = env.get('BOOKING_COOKIES', self._cookies)
    self._session = env.get('BOOKING_SESSION', self._session)
    self._csrf_token = env.get('BOOKING_CSRF_TOKEN', self._csrf_token)
    self.clear_cache()

  def _update_csrf(self, headers: httpx.Headers) -> None:
    """Capture CSRF from a response header if present and not yet known."""
    csrf = headers.get('x-booking-csrf')
    if csrf and csrf != 'undefined' and csrf != self._csrf_token:
      self._csrf_token = csrf
      set_credentials('booking', {'BOOKING_CSRF_TOKEN': csrf})

  async def _try_http_session_refresh(self) -> bool:
    """GET the monthly calendar page; extract a fresh ses= from the final URL.

    Also captures CSRF from response headers if present.
    Returns True if the session is still alive, False if re-authentication is needed.
    """
    url = f'{_BASE}/hotel/hoteladmin/extranet_ng/manage/calendar/monthly.html'
    params = {
      'hotel_id': str(self._hotel_id),
      'lang': self._lang,
      'ses': self._session,
      'view_mode': 'MONTHLY',
    }
    headers = {
      'Accept': 'text/html,application/xhtml+xml,*/*',
      'Accept-Language': 'en-US,en;q=0.9',
      'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/26.5 Safari/605.1.15',
      'Cookie': self._cookies,
    }
    try:
      async with httpx.AsyncClient(follow_redirects=True, timeout=20) as client:
        r = await client.get(url, params=params, headers=headers)
      final_url = str(r.url)
      if 'message=ERR100' in final_url or 'account.booking.com' in final_url:
        return False
      new_ses = parse_qs(urlparse(final_url).query).get('ses', [None])[0]
      if new_ses:
        self._update_csrf(r.headers)
        if new_ses != self._session:
          self._session = new_ses
          set_credentials('booking', {'BOOKING_SESSION': new_ses})
        self.clear_cache()
        return True
      return False
    except Exception:
      return False

  async def _auto_refresh(self, *, trigger: str = 'unknown') -> None:
    _log.warning('Booking.com session may be expired — trying HTTP session refresh')
    if await self._try_http_session_refresh():
      _log.info('Booking.com session refreshed via HTTP')
      return
    _log.warning('HTTP refresh failed — launching headless Playwright')
    try:
      refresh_booking_cookies()
      self._reload_from_env()
    except ServiceAuthError:
      mark_stale('booking')
      raise

  async def _ensure_refreshed(self, *, trigger: str) -> 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(trigger=trigger))
      task = self._refresh_task
    await task

  def _actor(self) -> dict:
    return {'actorId': str(self._hotel_id), 'actorType': 'HOTEL'}

  def _common_params(self, extra: dict | None = None) -> dict[str, str]:
    p = {
      'ses': self._session,
      'hotel_id': str(self._hotel_id),
      'lang': self._lang,
      'hotel_account_id': str(self._account_id),
    }
    if extra:
      p.update(extra)
    return p

  def _messaging_referer(self) -> str:
    return (f'{_BASE}/hotel/hoteladmin/extranet_ng/manage/messaging/inbox.html'
            f'?hotel_id={self._hotel_id}&lang={self._lang}&ses={self._session}&source=nav&tlc=1')

  def _monthly_referer(self) -> str:
    return (f'{_BASE}/hotel/hoteladmin/extranet_ng/manage/calendar/monthly.html'
            f'?hotel_id={self._hotel_id}&lang={self._lang}&ses={self._session}&view_mode=MONTHLY')

  def _matrix_referer(self) -> str:
    return (f'{_BASE}/hotel/hoteladmin/extranet_ng/manage/calendar/index.html'
            f'?view_mode=MATRIX&hotel_id={self._hotel_id}&lang={self._lang}&ses={self._session}')

  def _rest_headers(self, referer: str, extra: dict | None = None) -> dict[str, str]:
    h = {
      'Accept': 'application/json, text/plain, */*',
      'Accept-Language': 'en-US,en;q=0.9',
      'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/26.5 Safari/605.1.15',
      'Origin': _BASE,
      'Referer': referer,
      'Sec-Fetch-Mode': 'cors',
      'Sec-Fetch-Dest': 'empty',
      'Sec-Fetch-Site': 'same-origin',
      'X-Booking-SiteType-Id': '31',
      'X-Booking-Language-Code': 'en-us',
      'X-Booking-Pageview-Id': _page_view_id(),
      'Cookie': self._cookies,
    }
    if extra:
      h.update(extra)
    return h

  def _gql_headers(self, referer: str) -> dict[str, str]:
    return {
      'Accept': '*/*',
      'Accept-Language': 'en-US,en;q=0.9',
      'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/26.5 Safari/605.1.15',
      'Origin': _BASE,
      'Referer': referer,
      'Sec-Fetch-Mode': 'cors',
      'Sec-Fetch-Dest': 'empty',
      'Sec-Fetch-Site': 'same-origin',
      'x-booking-site-type-id': '31',
      'x-booking-context-action': 'hhemc_index',
      'x-booking-context-action-name': 'hhemc_index',
      'X-Booking-Pageview-ID': _page_view_id(),
      'x-booking-csrf-token': 'undefined',
      'apollographql-client-name': 'b-partner-availability-extranet-mfe',
      'apollographql-client-version': 'EQSXMHMY',
      'x-envoy-expected-rq-timeout-ms': '15000',
      'x-booking-timeout-budget-ms': '15000',
      'X-Booking-Topic': 'capla_browser_b-partner-availability-extranet-mfe',
      'Cookie': self._cookies,
    }

  def _msg_headers(self) -> dict[str, str]:
    return {
      'Accept': '*/*',
      'Accept-Language': 'en-US,en;q=0.9',
      'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/26.5 Safari/605.1.15',
      'Origin': _BASE,
      'Referer': self._messaging_referer(),
      'Sec-Fetch-Mode': 'cors',
      'Sec-Fetch-Dest': 'empty',
      'Sec-Fetch-Site': 'same-origin',
      'x-booking-site-type-id': '31',
      'x-booking-context-action': 'hhemm_inbox',
      'x-booking-context-action-name': 'hhemm_inbox',
      'X-Booking-Pageview-ID': _page_view_id(),
      'x-booking-csrf-token': 'undefined',
      'X-Booking-Topic': 'capla_browser_b-partner-messaging-extranet-mfe',
      'apollographql-client-name': 'b-partner-messaging-extranet-mfe',
      'apollographql-client-version': 'ABJaNSUc',
      'Cookie': self._cookies,
    }

  async def _msg_gql(self, operation_name: str, query: str, variables: dict) -> dict:
    params = {
      'hotel_id': str(self._hotel_id),
      'lang': self._lang,
      'ses': self._session,
      'source': 'nav',
      'tlc': '1',
    }
    async def _attempt() -> httpx.Response:
      async with httpx.AsyncClient() as client:
        return await client.post(
          f'{_BASE}/dml/graphql.json',
          params=params,
          json={'operationName': operation_name, 'variables': variables, 'query': query},
          headers=self._msg_headers(),
        )
    try:
      r = await _attempt()
    except (httpx.TransportError, OSError, ExceptionGroup) as e:
      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
    if r.status_code in (400, 401, 403):
      await self._ensure_refreshed(trigger='messaging')
      r = await _attempt()
    await self._check_auth(r)
    r.raise_for_status()
    raw = r.json()
    if raw.get('errors'):
      msgs = ', '.join(e.get('message', '') for e in raw['errors'])
      raise RuntimeError(f'GraphQL error: {msgs}')
    return raw

  async def _check_auth(self, r: httpx.Response) -> None:
    if r.status_code in (400, 401, 403):
      mark_stale('booking')
      raise BookingAuthError(f'Booking.com session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')

  async def _multipart_post(self, url: str, referer: str, body: dict, extra_headers: dict | None = None) -> dict:
    async def _attempt() -> httpx.Response:
      headers = self._rest_headers(referer, extra_headers)
      async with httpx.AsyncClient() as client:
        return await client.post(url, params=self._common_params(),
                                 files={'request': (None, json.dumps(body))}, headers=headers)
    try:
      r = await _attempt()
    except (httpx.TransportError, OSError, ExceptionGroup) as e:
      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
    if r.status_code in (400, 401, 403):
      await self._ensure_refreshed(trigger='inventory/fetch')
      r = await _attempt()
    await self._check_auth(r)
    r.raise_for_status()
    self._update_csrf(r.headers)
    return r.json()

  async def _gql_mutate(self, variables: dict) -> dict:
    async def _attempt() -> httpx.Response:
      referer = self._matrix_referer()
      params = {'view_mode': 'MATRIX', 'hotel_id': str(self._hotel_id),
                'lang': self._lang, 'ses': self._session}
      async with httpx.AsyncClient() as client:
        return await client.post(
          f'{_BASE}/dml/graphql.json',
          params=params,
          json={'operationName': 'updateRoomDateMutation', 'variables': variables,
                'extensions': {}, 'query': _UPDATE_MUTATION},
          headers=self._gql_headers(referer),
        )
    try:
      r = await _attempt()
    except (httpx.TransportError, OSError, ExceptionGroup) as e:
      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
    if r.status_code in (400, 401, 403):
      await self._ensure_refreshed(trigger='graphql')
      r = await _attempt()
    await self._check_auth(r)
    r.raise_for_status()
    raw = r.json()
    if raw.get('errors'):
      msgs = ', '.join(e.get('message', '') for e in raw['errors'])
      raise RuntimeError(f'GraphQL error: {msgs}')
    result = (raw.get('data', {}).get('partnerAvailabilityMutations', {})
                  .get('partnerAvailabilityUpdate', {}))
    if not result.get('success'):
      errs = ', '.join(e.get('message', '') for e in (result.get('result', {}).get('errors') or []))
      raise RuntimeError(f'updateRoomDateMutation failed{": " + errs if errs else ""}')
    return raw

  def _parse_calendar_days(self, raw: dict) -> dict[str, DayInfo]:
    hotel_data = raw.get('data', {}).get('hotel', {}).get(str(self._hotel_id), {})
    room_data = hotel_data.get('rooms', {}).get(str(self._room_id), {})
    rate_dates = room_data.get('rates', {}).get(str(self._rate_id), {}).get('dates', {})
    result: dict[str, DayInfo] = {}
    room_level_dates = room_data.get('dates', {})
    if room_level_dates:
      # Legacy backend: room-level dates include hidden/net_booked/rooms_to_sell
      for d, info in room_level_dates.items():
        price_str = rate_dates.get(d, {}).get('price', {}).get('2')
        result[d] = DayInfo(
          date=d,
          price=round(float(price_str)) if price_str else None,
          available=(info.get('hidden', 0) == 0),
          booked=(info.get('net_booked', 0) > 0),
          rooms_to_sell=info.get('rooms_to_sell', 0),
        )
    else:
      # Partner availability API: only rate-level dates; status encodes availability
      # "restricted" = open, "closed" = unavailable, "soldout" = booked
      for d, rate_info in rate_dates.items():
        price_str = rate_info.get('price', {}).get('2')
        status = rate_info.get('status', '')
        result[d] = DayInfo(
          date=d,
          price=round(float(price_str)) if price_str else None,
          available=(status == 'restricted'),
          booked=(status == 'soldout'),
          rooms_to_sell=None,
        )
    return result

  async def _fetch_month_calendar(self, month: int, year: int) -> dict[str, DayInfo]:
    start, end = month_bounds(month, year)
    if _is_past(end):
      warnings.warn(f'Booking.com: skipping calendar fetch for past month {year}-{month:02d}', stacklevel=3)
      return {}
    body = {
      'dates': {'range': True, 'dates': [start, end]},
      'hotel': {
        'fields': ['rooms', 'policygroups'],
        'rooms': {
          'id': [self._room_id],
          'fields': ['status', 'rooms_to_sell', 'permissions', 'rates', 'nr_rooms', 'net_booked', 'num_guests'],
          'rates': {'fields': ['permissions', 'price', 'restrictions', 'status', 'rooms_to_sell',
                               'policygroup_id', 'policy_overrides', 'default_policygroup_id', 'name']},
        },
        'policy': {'fields': ['info']},
      },
      'post_process_data_plugins': ['format_price', 'multi_units_check', 'last_update', 'check_dynamic_restrictions'],
    }
    raw = await self._multipart_post(
      f'{_BASE}/fresa/extranet/inventory/fetch', self._monthly_referer(), body
    )
    result = self._parse_calendar_days(raw)
    if not result or not any(info.price is not None for info in result.values()):
      _log.warning('Booking.com: empty or priceless calendar for %d-%02d; retrying', year, month)
      await asyncio.sleep(1)
      raw = await self._multipart_post(
        f'{_BASE}/fresa/extranet/inventory/fetch', self._monthly_referer(), body
      )
      result = self._parse_calendar_days(raw)
      if not result or not any(info.price is not None for info in result.values()):
        _log.warning('Booking.com: still empty/priceless calendar for %d-%02d after retry', year, month)
    return result

  async def _get_month_calendar(self, month: int, year: int) -> dict[str, DayInfo]:
    key = (month, year)
    if key in self._month_data:
      return self._month_data[key]
    if key in self._months_inflight:
      return await self._months_inflight[key]
    task = asyncio.create_task(self._fetch_month_calendar(month, year))
    self._months_inflight[key] = task
    try:
      result = await task
    finally:
      self._months_inflight.pop(key, None)
    self._month_data[key] = result
    return result

  async def get_calendar(self, date_from: str, date_to: str) -> dict[str, DayInfo]:
    if is_stale('booking'):
      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
    month_list = months_in_range(date_from, date_to)
    all_results = await asyncio.gather(*[self._get_month_calendar(m, y) for m, y in month_list])
    for month_result in all_results:
      self._cache.update(month_result)
    return {d: info for d, info in self._cache.items() if date_from <= d <= date_to}

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

  async def get_reservations(self, date_from: str, date_to: str) -> list[Reservation]:
    if is_stale('booking'):
      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
    async def _attempt() -> httpx.Response:
      referer = self._monthly_referer()
      params = self._common_params({
        'from': date_from,
        'to': date_to,
        'room_id': str(self._room_id),
        'only_basic_info': '1',
      })
      async with httpx.AsyncClient() as client:
        return await client.post(
          f'{_BASE}/fresa/extranet/calendar/get_reservations',
          params=params,
          headers=self._rest_headers(referer),
        )
    try:
      r = await _attempt()
    except (httpx.TransportError, OSError, ExceptionGroup) as e:
      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
    if r.status_code in (400, 401, 403):
      await self._ensure_refreshed(trigger='get_reservations')
      r = await _attempt()
    await self._check_auth(r)
    r.raise_for_status()
    raw = r.json()

    reservations = raw.get('data', {}).get('reservations') or {}
    return [
      Reservation(
        service='booking',
        reservation_id=str(rid),
        check_in=r_data.get('checkin', ''),
        check_out=r_data.get('checkout', ''),
        guest_name=(r_data.get('show_name')
                    or (f"{r_data['firstname']} {r_data['lastname']}"
                        if r_data.get('firstname') and r_data.get('lastname')
                        else None)),
        chat_id=str(rid),
      )
      for rid, r_data in reservations.items()
    ]

  def send_ready(self) -> bool:
    return True

  async def get_thread(self, reservation_id: str) -> BookingThread:
    """Fetch a conversation thread and its messages by reservation ID."""
    if is_stale('booking'):
      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
    raw = await self._msg_gql(
      'conversationMessages', _CONVERSATION_MESSAGES_QUERY,
      {'input': {
        'actor': self._actor(),
        'conversationReference': {
          'productId': reservation_id,
          'productType': 'POST_BOOKING',
          'verticalType': 'ACCOMMODATION',
        },
        'pagination': {'count': 50},
      }},
    )
    data = raw.get('data', {}).get('chatConversation', {}).get('singleConversation', {})
    is_reply_needed = ((data.get('conversation') or {}).get('flags') or {}).get('isReplyNeeded', False)
    messages = []
    for m in data.get('messages', []):
      msg_type = m.get('type')
      content = m.get('content')
      attrs = m.get('attributes') or {}
      event_body = (attrs.get('eventDetail') or {}).get('body')
      if msg_type == 'FREE_TEXT':
        body, body_translated = content, None
      elif event_body is not None:
        body, body_translated = content, event_body
      else:
        continue  # system message with no displayable text
      messages.append(BookingMessage(
        id=m['messageId'],
        created_at=m['created'],
        body=body,
        sender=(m.get('sender') or {}).get('participantType', ''),
        body_translated=body_translated,
      ))
    return BookingThread(reservation_id=reservation_id, messages=messages, is_reply_needed=is_reply_needed)

  async def send_message(self, reservation_id: str, body: str) -> None:
    """Send a text message to the guest conversation for a reservation."""
    if is_stale('booking'):
      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
    idempotency = f'{reservation_id}-{int(time.time() * 1000)}'
    payload = {'input': {
      'actor': self._actor(),
      'conversationReference': {
        'productId': reservation_id,
        'productType': 'POST_BOOKING',
        'verticalType': 'ACCOMMODATION',
      },
      'idempotency': idempotency,
      'content': body,
      'attachmentIds': [],
    }}
    try:
      await self._msg_gql('postFreeTextMessage', _POST_FREE_TEXT_MUTATION, payload)
    except RuntimeError as e:
      if 'Internal Server Error' not in str(e):
        raise
      # GraphQL ISE on send typically means the messaging 2FA session has expired.
      # Re-trigger the SMS challenge and retry once.
      _log.warning('Booking.com send_message failed with ISE — re-triggering 2FA')
      trigger_booking_chat_auth()
      self._reload_from_env()
      await self._msg_gql('postFreeTextMessage', _POST_FREE_TEXT_MUTATION, payload)

  async def set_price(self, d: str, price: int, occupancy: int = 2) -> None:
    price = int(round(price))
    if is_stale('booking'):
      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
    if _is_past(d):
      warnings.warn(f'Booking.com: skipping set_price for past date {d}', stacklevel=2)
      return
    if (cached := self._cache.get(d)) and cached.price == price:
      return
    if not self._csrf_token:
      raise ValueError('BOOKING_CSRF_TOKEN not available — call get_calendar() first, or run refresh-creds')
    price_str = str(price)
    body = {
      'update': [{
        'room_id': self._room_id,
        'from_date': d,
        'until_date': d,
        'limit_to_weekdays': [0, 1, 2, 3, 4, 5, 6],
        'rate_id': self._rate_id,
        'field_name': 'price',
        'field_value': price_str,
        'occupancy': occupancy,
      }],
      'modified_fields': {'price_changed': 1, 'price_changed_value': price_str,
                          'date_selection_mode': 'CELLS'},
      'post_process_data_plugins': ['format_price', 'last_update'],
      'update_source': {'product': 'calendar', 'extra_data': {'new_calendar': 1, 'location': 'input'}},
      'status_reason': 1,
      'prevent_acav': '1',
      'check_adding_av_to_closed_property': '0',
    }
    await self._multipart_post(
      f'{_BASE}/fresa/extranet/inventory/update', self._monthly_referer(), body,
      extra_headers={'X-Booking-CSRF': self._csrf_token},
    )
    self.clear_cache()
    _log.info('booking set_price date=%s price=$%d', d, price)

  async def set_availability(self, d: str, available: bool) -> None:
    if is_stale('booking'):
      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
    if _is_past(d):
      warnings.warn(f'Booking.com: skipping set_availability for past date {d}', stacklevel=2)
      return
    expected_rts = 1 if available else 0
    if (cached := self._cache.get(d)) and cached.available == available and cached.rooms_to_sell == expected_rts:
      return
    await self._gql_mutate({'input': {
      'checkAddingAVToClosedProperty': False,
      'hotelId': self._hotel_id,
      'postProcessDataPlugins': ['mlos_status_reason'],
      'preventAcav': True,
      'statusReason': True,
      'updateAvailability': [{
        'fromDate': d,
        'roomId': self._room_id,
        'untilDate': d,
        'hidden': not available,
        'additionalChecks': {'alertMode': 'WARN'},
      }],
    }})
    if not available:
      await self.set_rooms_to_sell(d, 0)
    else:
      cal = await self.get_calendar(d, d)
      if not cal.get(d, DayInfo(date=d, price=None, available=True, booked=False, rooms_to_sell=None)).booked:
        await self.set_rooms_to_sell(d, 1)
    self.clear_cache()
    _log.info('booking set_availability date=%s available=%s', d, available)

  async def set_rooms_to_sell(self, d: str, count: int) -> None:
    if is_stale('booking'):
      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
    if _is_past(d):
      warnings.warn(f'Booking.com: skipping set_rooms_to_sell for past date {d}', stacklevel=2)
      return
    entry: dict = {
      'fromDate': d,
      'roomId': self._room_id,
      'untilDate': d,
      'roomsToSell': count,
    }
    if count > 0:
      entry['additionalChecks'] = {'alertMode': 'WARN'}
    await self._gql_mutate({'input': {
      'checkAddingAVToClosedProperty': False,
      'hotelId': self._hotel_id,
      'postProcessDataPlugins': ['mlos_status_reason'],
      'preventAcav': True,
      'statusReason': True,
      'updateAvailability': [entry],
    }})
    self.clear_cache()