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']+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:

Name

cn_m = re.search(r'

[^<]*]*>([^<]+)', 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[^>]*>(.*?)', 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