Add Rate Monitor app — Booking.com + direct booking engine competitor rates

Combines Booking.com Playwright scraper (from forecasting), direct booking
engine scraper (ported from laptop-archive/guestline-monitor), and Newbook
own-hotel rates into one focused tool. Four views: Bookability, Market View
(with price index badges + direct rate sub-rows), Direct Rates (per-competitor
room breakdown, min-stay flags, hotel config/discovery), Rate Analysis
(advance purchase curve, DOW chart, rate timeline, comparison table).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jtricerolph 2026-07-05 12:06:30 +00:00
commit e05054172f
50 changed files with 11860 additions and 0 deletions

View file

@ -0,0 +1,28 @@
from .guestline import GuestlineProfile
from .newbook_scrape import NewbookScrapeProfile
from .travelclick import TravelClickProfile
from .directbook import DirectBookProfile
from .base import BaseProfile
PROFILES = {
"guestline": GuestlineProfile,
"newbook_scrape": NewbookScrapeProfile,
"travelclick": TravelClickProfile,
"directbook": DirectBookProfile,
}
def get_profile(name: str) -> BaseProfile:
cls = PROFILES.get(name)
if not cls:
raise ValueError(f"Unknown engine profile: {name}")
return cls()
def detect_profile(url: str) -> dict | None:
"""Given a booking URL, return suggested profile name and extracted params."""
for name, cls in PROFILES.items():
result = cls.detect(url)
if result is not None:
return {"profile": name, **result}
return None

View file

@ -0,0 +1,25 @@
from abc import ABC, abstractmethod
class BaseProfile(ABC):
name: str = ""
label: str = ""
# Fields required to configure this engine, shown in the add-hotel form
# Each entry: {"key": "hotel_id", "label": "Hotel ID", "help": "e.g. THREEWAYS"}
required_params: list[dict] = []
@classmethod
@abstractmethod
def detect(cls, url: str) -> dict | None:
"""Return extracted params dict if URL matches this engine, else None."""
@abstractmethod
async def fetch_arrival_dates(self, client, params: dict) -> list[str]:
"""Return list of bookable arrival date strings YYYY-MM-DD."""
@abstractmethod
async def fetch_night_rates(self, client, params: dict, arrival: str, nights: int = 1) -> list[dict]:
"""Return list of room/rate dicts for the stay.
Each dict must have: roomId, rateId, availability, prices[{amountBeforeTax, amountAfterTax}], currencyCode
prices list has one entry per night when nights > 1.
"""

View file

@ -0,0 +1,175 @@
import json
import re
from datetime import date, timedelta
from urllib.parse import quote
from .base import BaseProfile
API_BASE = "https://direct-book.com"
SETTINGS_HASH = "52d786f5c45232c8c16022bc3af6dab2e1994f4919b953afafe188095125e9b6"
QUOTESETS_HASH = "1012a6203854357e44786380240eefad6b2ad863aee6ba79748c81a851f29217"
ROOMTYPES_HASH = "8021345a2e1f993717b1960097489b456a6b2dc136990b6f919adba1b1fe2c1f"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "application/json",
# Required to bypass Apollo CSRF protection on the /api/graphql endpoint
"Content-Type": "application/json",
}
def _graphql_url(operation: str, variables: dict, sha256: str) -> str:
return (
f"{API_BASE}/api/graphql"
f"?operationName={operation}"
f"&variables={_json_compact(variables)}"
f"&extensions={_json_compact({'persistedQuery': {'version': 1, 'sha256Hash': sha256}})}"
)
def _json_compact(obj) -> str:
return quote(json.dumps(obj, separators=(',', ':')), safe='')
async def _get_property_id(client, channel_code: str) -> str:
"""Fetch numeric propertyId from settings query."""
url = _graphql_url("settings", {"channelCode": channel_code}, SETTINGS_HASH)
r = await client.get(url, headers=HEADERS, timeout=20)
r.raise_for_status()
return str(r.json()["data"]["settings"]["uuid"])
class DirectBookProfile(BaseProfile):
name = "directbook"
label = "SiteMinder Direct Book"
required_params = [
{"key": "channel_code", "label": "Channel Code",
"help": "The property slug in the booking URL, e.g. 'grapevinestowdirect'"},
]
@classmethod
def detect(cls, url: str) -> dict | None:
m = re.search(r'direct-book\.com/properties/([^/?#\s]+)', url)
if m:
return {"channel_code": m.group(1)}
return None
async def fetch_category_names(self, client, params: dict) -> tuple[dict[str, str], dict[str, str]]:
channel_code = params["channel_code"]
url = _graphql_url(
"roomTypes",
{"channelCode": channel_code, "checkInDate": date.today().isoformat(),
"checkOutDate": (date.today() + timedelta(days=1)).isoformat(), "locale": "en"},
ROOMTYPES_HASH,
)
r = await client.get(url, headers=HEADERS, timeout=30)
r.raise_for_status()
room_types = r.json()["data"]["roomTypes"]
room_labels = {rt["uuid"]: rt["name"] for rt in room_types}
rate_labels = {}
for rt in room_types:
for rate in rt.get("rates", []):
rate_labels[rate["uuid"]] = rate["name"]
return room_labels, rate_labels
async def fetch_arrival_dates(self, client, params: dict) -> list[str]:
channel_code = params["channel_code"]
today = date.today()
# Fetch 12 months of availability in monthly chunks (API seems to accept wide ranges too)
end = today.replace(day=1) + timedelta(days=365)
# Build monthly windows to stay within API limits
available_dates: list[str] = []
current = today.replace(day=1)
while current <= end:
# Last day of the month
if current.month == 12:
month_end = current.replace(year=current.year + 1, month=1, day=1) - timedelta(days=1)
else:
month_end = current.replace(month=current.month + 1, day=1) - timedelta(days=1)
check_from = max(today, current).isoformat()
check_to = month_end.isoformat()
url = (
f"{API_BASE}/api/properties/{channel_code}/availability"
f"?checkInsFrom={check_from}&checkInsTo={check_to}"
)
try:
r = await client.get(url, headers=HEADERS, timeout=20)
r.raise_for_status()
for entry in r.json().get("result", []):
if entry.get("canCheckIn"):
d = entry["date"][:10]
if d >= today.isoformat():
available_dates.append(d)
except Exception:
pass
# Advance to next month
if current.month == 12:
current = current.replace(year=current.year + 1, month=1)
else:
current = current.replace(month=current.month + 1)
return sorted(set(available_dates))
async def fetch_night_rates(self, client, params: dict, arrival: str, nights: int = 1) -> list[dict]:
channel_code = params["channel_code"]
# Get propertyId (numeric) — cache it on the params dict to avoid repeat fetches
if "property_id" not in params:
params["property_id"] = await _get_property_id(client, channel_code)
property_id = int(params["property_id"])
departure = (date.fromisoformat(arrival) + timedelta(days=nights)).isoformat()
url = _graphql_url(
"quoteSets",
{
"propertyId": property_id,
"promocode": "",
"checkInDate": arrival,
"checkOutDate": departure,
"adults": 2,
"children": 0,
"infants": 0,
"currencyFrom": "GBP",
"currencyTo": "GBP",
},
QUOTESETS_HASH,
)
r = await client.get(url, headers=HEADERS, timeout=20)
if r.status_code == 404:
return []
r.raise_for_status()
rooms = []
for qs in r.json()["data"].get("quoteSets", []):
room_id = str(qs["roomTypeId"])
for quote in qs.get("quotes", []):
rate_id = str(quote["roomRateId"])
price = quote["price"]["amount"]
available = quote.get("available", 0)
# Build per-night prices from breakdown if multi-night
breakdown = quote.get("breakdown", [])
if breakdown:
prices = [
{"amountBeforeTax": b["price"]["amount"], "amountAfterTax": b["price"]["amount"]}
for b in breakdown
]
else:
prices = [{"amountBeforeTax": price, "amountAfterTax": price}]
rooms.append({
"roomId": room_id,
"rateId": rate_id,
"availability": available,
"prices": prices,
"min_stay_nights": None,
"currencyCode": "GBP",
})
return rooms

View file

@ -0,0 +1,49 @@
import re
from datetime import date, timedelta
from .base import BaseProfile
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "application/json",
}
class GuestlineProfile(BaseProfile):
name = "guestline"
label = "Guestline"
required_params = [
{"key": "hotel_id", "label": "Hotel ID", "help": "e.g. THREEWAYS — from the booking URL ?hotel= parameter"},
{"key": "collection_id", "label": "Collection ID", "help": "e.g. MT — the path segment before /availability in the booking URL"},
]
@classmethod
def detect(cls, url: str) -> dict | None:
# Matches: https://booking.eu.guestline.app/MT/availability?hotel=THREEWAYS
m = re.search(r'booking\.(?:eu\.)?guestline\.app/([^/?\s]+)/availability\?hotel=([^&\s]+)', url)
if m:
return {"collection_id": m.group(1), "hotel_id": m.group(2)}
return None
async def fetch_arrival_dates(self, client, params: dict) -> list[str]:
hotel_id = params["hotel_id"]
today = date.today()
url = f"https://booking.eu.guestline.app/api/availabilities/{hotel_id}/arrivals"
r = await client.get(url, params={
"month": today.month, "year": today.year,
"adults": 2, "children": 0, "count": 12,
}, headers=HEADERS, timeout=20)
r.raise_for_status()
return [a["date"] for a in r.json().get("arrivals", [])]
async def fetch_night_rates(self, client, params: dict, arrival: str, nights: int = 1) -> list[dict]:
hotel_id = params["hotel_id"]
collection_id = params.get("collection_id", "MT")
dep = (date.fromisoformat(arrival) + timedelta(days=nights)).isoformat()
url = f"https://booking.eu.guestline.app/api/availabilities/{collection_id}/{hotel_id}/enhanced"
r = await client.get(url, params={
"arrival": arrival, "departure": dep, "adults": 2, "children": 0,
}, headers=HEADERS, timeout=20)
if r.status_code == 404:
return []
r.raise_for_status()
return r.json().get("availabilities", {}).get("rooms", [])

View file

@ -0,0 +1,235 @@
import json
import re
from datetime import date, timedelta
from .base import BaseProfile
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "*/*",
"X-Requested-With": "XMLHttpRequest",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
}
BASE_URL = "https://bookingseu.newbook.cloud"
def _base_params(slug: str, arrival: str, departure: str, nights: int) -> dict:
return {
"REMOTE_ADDR": "1.1.1.1",
"force_booking_channel_id": "",
"HTTP_REFERER": f"bookingseu.newbook.cloud/{slug}/index.php",
"discount_total_display": "0",
"force_category_id[]": "uK0ip@c7ty%8bQ#2i",
"force_category_type_id[]": "uK0ip@c7ty%8bQ#2i",
"no_billing_booking": "0",
"force_tariff_type_id[]": "uK0ip@c7ty%8bQ#2i",
"discount_id": "null",
"facebook_user_id": "",
"category_type_id": "",
"owner_occupied_booking_id": "",
"discount_code": "",
"booking_action": "",
"available_from": arrival,
"available_to": departure,
"nights": str(nights),
"adults": "2",
"children": "0",
"infants": "0",
"promo_code": "",
"language": "EN",
}
def _fmt_date(d: date) -> str:
"""Format date as NewBook expects: 'Mon 4 Jul 2026'"""
return d.strftime("%a %-d %b %Y")
def _parse_chart_html(html: str) -> tuple[dict[str, float], dict[str, str], dict[str, str], list[dict]]:
"""
Parse an availability_chart_responsive HTML response.
Returns:
counts: {cat_id: float} room counts from category_sites_available JS var
cat_names: {cat_id: str} friendly category names e.g. "Executive Double"
rate_names: {rate_id: str} friendly tariff names e.g. "DIRECT B&B FLEX"
rooms: list of room/rate dicts compatible with base scraper format
"""
# Room counts from embedded JS
counts: dict[str, float] = {}
m = re.search(r'category_sites_available\s*=\s*(\{[^;]+\})', html)
if m:
try:
counts = {k: float(v) for k, v in json.loads(m.group(1)).items()}
except Exception:
pass
cat_names: dict[str, str] = {}
rate_names: dict[str, str] = {}
rooms = []
# Split by category box: offset="{cat_id}"
cat_blocks = re.split(r'<div[^>]+class="[^"]*newbook_online_category_box[^"]*"[^>]+offset="(\d+)"', html)
# cat_blocks: [pre, cat_id, block, cat_id, block, ...]
i = 1
while i < len(cat_blocks) - 1:
cat_id = cat_blocks[i]
block = cat_blocks[i + 1]
i += 2
avail = counts.get(cat_id, 0.0)
# Category friendly name — from category_name attr on any book button in this block
# e.g. category_name='Standard' or category_name='Executive Double'
cn_m = re.search(r"category_name='([^']+)'", block)
if not cn_m:
# Fallback: <h3><a ...>Name</a></h3>
cn_m = re.search(r'<h3>[^<]*<a[^>]*>([^<]+)</a>', block)
if cn_m:
cat_names[cat_id] = cn_m.group(1).strip()
# Split on tariff row boundaries
tariff_rows = re.split(r'class="[^"]*newbook_online_categories_tariff_type_rows[^"]*"', block)
for row in tariff_rows[1:]:
# Rate label
name_m = re.search(r'newbook_online_categories_tariff_type_label[^>]*>(.*?)</div>', row, re.DOTALL)
rate_label = re.sub(r'<[^>]+>', '', name_m.group(1)).strip() if name_m else ""
# Price
price_m = re.search(r'newbook_online_from_price_text[^>]*>£([\d.]+)<', row)
price = float(price_m.group(1)) if price_m else None
# Internal tariff type ID (stable across dates)
tid_m = re.search(r'tariff_type_id="(\d+)"', row)
rate_id = tid_m.group(1) if tid_m else rate_label
# Store rate label for this rate_id
if rate_id and rate_label:
rate_names[rate_id] = rate_label
# Min-stay: requires_date_change class + optional extend_nights attr
# If extend_nights present: min_stay = 1 + N; if absent: default to 2
min_stay = None
if 'requires_date_change' in row:
en_m = re.search(r'extend_nights="(\d+)"', row)
min_stay = 1 + int(en_m.group(1)) if en_m else 2
if rate_label and price is not None:
rooms.append({
"roomId": cat_id,
"rateId": rate_id,
"rateLabel": rate_label,
"availability": int(avail),
"prices": [{"amountBeforeTax": price, "amountAfterTax": price}],
"min_stay_nights": min_stay,
"currencyCode": "GBP",
})
return counts, cat_names, rate_names, rooms
class NewbookScrapeProfile(BaseProfile):
name = "newbook_scrape"
label = "NewBook (HTML scrape)"
required_params = [
{"key": "slug", "label": "Property Slug",
"help": "The path segment in the booking URL, e.g. 'numberfour' from bookingseu.newbook.cloud/numberfour/"},
]
@classmethod
def detect(cls, url: str) -> dict | None:
m = re.search(r'bookingseu\.newbook\.cloud/([^/?#\s]+)', url)
if m and m.group(1) not in ('index.php',):
return {"slug": m.group(1)}
return None
async def fetch_category_names(self, client, params: dict) -> tuple[dict[str, str], dict[str, str]]:
"""
Return ({cat_id: friendly_name}, {rate_id: friendly_name}) from a single chart call.
Used by discovery to pre-populate room_labels and rate_labels.
"""
slug = params["slug"]
today = date.today()
base = _base_params(slug, _fmt_date(today), _fmt_date(today + timedelta(days=1)), 1)
r = await client.post(
f"{BASE_URL}/{slug}/api.php?newbook_api_action=availability_chart_responsive",
data=base, headers=HEADERS, timeout=30,
)
r.raise_for_status()
_, cat_names, rate_names, _ = _parse_chart_html(r.text)
return cat_names, rate_names
async def fetch_arrival_dates(self, client, params: dict) -> list[str]:
"""
Use the calendar endpoint to collect all available arrival dates
across all room types. One call per room type, union of available dates.
We first do a chart call to discover category IDs, then calendar per category.
"""
slug = params["slug"]
today = date.today()
arrival_str = _fmt_date(today)
departure_str = _fmt_date(today + timedelta(days=1))
# Step 1: chart call to discover category IDs and names
base = _base_params(slug, arrival_str, departure_str, 1)
r = await client.post(
f"{BASE_URL}/{slug}/api.php?newbook_api_action=availability_chart_responsive",
data=base, headers=HEADERS, timeout=30,
)
r.raise_for_status()
html = r.text
cat_ids = re.findall(r'class="[^"]*newbook_online_category_box[^"]*"[^>]+offset="(\d+)"', html)
if not cat_ids:
return []
# Step 2: calendar call per category, collect available dates
available_dates: set[str] = set()
more_tariffs = {f"more_tariffs_{cid}": "1" for cid in cat_ids}
for cat_id in cat_ids:
cal_params = {
**base,
**more_tariffs,
"query": "newbook_calendar_initialise",
"calendar_category_id": cat_id,
}
try:
cr = await client.post(
f"{BASE_URL}/{slug}/api.php?newbook_api_action=data",
data=cal_params, headers=HEADERS, timeout=30,
)
cr.raise_for_status()
cal_data = cr.json()
cal_html = cal_data.get("calendar_display", "")
for dm in re.finditer(r'class="day available[^"]*"\s+data-date="(\d{4}-\d{2}-\d{2})"', cal_html):
available_dates.add(dm.group(1))
except Exception:
pass
return sorted(d for d in available_dates if d >= today.isoformat())
async def fetch_night_rates(self, client, params: dict, arrival: str, nights: int = 1) -> list[dict]:
"""
POST to availability_chart_responsive for a specific date.
Returns list of room/rate dicts compatible with the base scraper format.
Also injects min_stay_nights onto each row.
"""
slug = params["slug"]
arr = date.fromisoformat(arrival)
dep = arr + timedelta(days=nights)
arr_str = _fmt_date(arr)
dep_str = _fmt_date(dep)
body = _base_params(slug, arr_str, dep_str, nights)
r = await client.post(
f"{BASE_URL}/{slug}/api.php?newbook_api_action=availability_chart_responsive",
data=body, headers=HEADERS, timeout=30,
)
if r.status_code == 404:
return []
r.raise_for_status()
_, _, _, rooms = _parse_chart_html(r.text)
return rooms

View file

@ -0,0 +1,166 @@
import re
import httpx
from datetime import date, timedelta
from .base import BaseProfile
API_BASE = "https://api.travelclick.com"
TOKEN_URL = f"{API_BASE}/oauth/token-referer?grant_type=client_credentials"
HEADERS_BASE = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "application/json",
"Content-Type": "application/json",
}
async def _get_token(client, referer: str) -> str:
r = await client.post(TOKEN_URL, headers={**HEADERS_BASE, "Referer": referer}, timeout=20)
r.raise_for_status()
return r.json()["access_token"]
def _auth_headers(token: str, referer: str) -> dict:
return {**HEADERS_BASE, "Authorization": f"Bearer {token}", "Referer": referer}
class TravelClickProfile(BaseProfile):
name = "travelclick"
label = "TravelClick / Amadeus"
required_params = [
{"key": "hotel_code", "label": "Hotel Code",
"help": "Numeric hotel ID, e.g. 77346 — visible in the booking engine network requests"},
{"key": "booking_url", "label": "Booking URL Base",
"help": "e.g. https://reservations.bespokehotels.com/noelarmshotel/book/dates-of-stay"},
]
@classmethod
def detect(cls, url: str) -> dict | None:
m = re.search(r'(https://reservations\.bespokehotels\.com/[^/]+/book/[^?#\s]+)', url)
if not m:
return None
booking_url = m.group(1)
# hotel_code is in inline JS as bookingEngineHotelId: '77346' on the booking page
hotel_code = ""
try:
r = httpx.get(booking_url, timeout=10, follow_redirects=True,
headers={"User-Agent": "Mozilla/5.0"})
hm = re.search(r'bookingEngineHotelId\s*:\s*[\'"](\d+)[\'"]', r.text)
if hm:
hotel_code = hm.group(1)
except Exception:
pass
return {"booking_url": booking_url, "hotel_code": hotel_code}
async def fetch_category_names(self, client, params: dict) -> tuple[dict[str, str], dict[str, str]]:
hotel_code = params["hotel_code"]
referer = params["booking_url"]
token = await _get_token(client, referer)
r = await client.get(
f"{API_BASE}/be5-entity/v2/hotels/{hotel_code}/content",
params={"include": "roomtypes,rateplans", "lang": "EN_US"},
headers=_auth_headers(token, referer),
timeout=30,
)
r.raise_for_status()
data = r.json()
# roomtypes[].roomTypeId (numeric, used as roomId in avail) -> roomTypeName
room_labels = {
str(rt["roomTypeId"]): rt.get("roomTypeName", str(rt["roomTypeId"]))
for rt in data.get("roomtypes", [])
}
# ratePlans[].rateplanId (numeric, used as rateId in avail) -> rateplanName
rate_labels = {
str(rp["rateplanId"]): rp.get("rateplanName", str(rp["rateplanId"]))
for rp in data.get("ratePlans", [])
}
return room_labels, rate_labels
async def fetch_arrival_dates(self, client, params: dict) -> list[str]:
hotel_code = params["hotel_code"]
referer = params["booking_url"]
token = await _get_token(client, referer)
today = date.today()
end = today + timedelta(days=365)
body = {
"hotelCode": int(hotel_code),
"currency": "GBP",
"lang": "EN_US",
"dateIn": today.isoformat(),
"dateOut": end.isoformat(),
"multiRoomOccupancy": [{"adults": 2, "infant": 0, "children": 0}],
"bookerIdentifier": "",
"partnerIdentifier": "",
}
r = await client.post(
f"{API_BASE}/be5-shop/v1/hotel/{hotel_code}/basicavail/multi-room",
json=body,
headers=_auth_headers(token, referer),
timeout=30,
)
r.raise_for_status()
data = r.json()
return [
d["date"]
for d in data.get("dates", [])
if d.get("isAvailable") and d["date"] >= today.isoformat()
]
async def fetch_night_rates(self, client, params: dict, arrival: str, nights: int = 1) -> list[dict]:
hotel_code = params["hotel_code"]
referer = params["booking_url"]
token = await _get_token(client, referer)
departure = (date.fromisoformat(arrival) + timedelta(days=nights)).isoformat()
body = {
"roomStay": {
"startDate": arrival,
"endDate": departure,
"guestCount": {"adults": 2, "infants": 0, "children": {"ages": None, "count": 0}},
"roomQuantity": 1,
"productSearchCriteria": {
"sortingPreference": "SORT_BY_ORDER",
"includeUnavailable": False,
},
},
"languageCode": "EN_US",
"disableLocaitonSharing": False,
"currencyCode": "GBP",
"tpaExtension": [],
"includeMemberRate": True,
"includeNightlyRates": True,
}
r = await client.post(
f"{API_BASE}/be5-shop/v2/hotels/{hotel_code}/avail",
json=body,
headers=_auth_headers(token, referer),
timeout=30,
)
if r.status_code == 404:
return []
r.raise_for_status()
data = r.json()
# Response: roomStays[0].roomtypes[].products[] (both Regular rates and Packages)
room_stay = (data.get("roomStays") or [{}])[0]
rooms = []
for rt in room_stay.get("roomtypes", []):
room_id = str(rt["roomtypeId"])
for product in rt.get("products", []):
rate_id = str(product["productId"])
nightly = product.get("nightlyRates", [])
if not nightly:
continue
avail = nightly[0].get("inventoryCount", 0)
prices = [
{"amountBeforeTax": n["amountBeforeTax"], "amountAfterTax": n.get("amountTotal", n["amountBeforeTax"])}
for n in nightly
]
rooms.append({
"roomId": room_id,
"rateId": rate_id,
"availability": avail,
"prices": prices,
"min_stay_nights": None, # min-stay comes from basicavail per date
"currencyCode": "GBP",
})
return rooms