import hashlib import os import re from datetime import date, datetime, time, timedelta, timezone from zoneinfo import ZoneInfo from .base import BaseProfile CLIENT = "Mews Distributor 5745.0.0" API_BASE = "https://api.mews.com/api/bookingEngine/v1" DAYS_AHEAD = 90 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "application/json", "Content-Type": "application/json", "X-Accept-Casing": "Camel", } def _session() -> str: rand = os.urandom(4).hex() now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") raw = rand + now ek = "".join(f"{ord(c):03d}" for c in raw) return (ek + hashlib.md5((ek + CLIENT).encode()).hexdigest()).upper() def _midnight_utc(d: date, tz: ZoneInfo) -> str: return datetime.combine(d, time.min, tzinfo=tz).astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _utc_to_local_date(utc_str: str, tz: ZoneInfo) -> str: return datetime.fromisoformat(utc_str.replace("Z", "+00:00")).astimezone(tz).date().isoformat() async def _ensure_config(client, params: dict) -> bool: if "enterprise_id" in params: return True r = await client.post( f"{API_BASE}/configurations/get", json={"ids": [params["booking_engine_id"]], "client": CLIENT, "session": _session()}, headers=HEADERS, timeout=20, ) if r.status_code != 200: return False data = r.json() be = (data.get("bookingEngines") or [{}])[0] service_id = be.get("serviceId", "") # enterpriseId lives on services[], not bookingEngines[] enterprise_id = "" for svc in data.get("services") or []: if svc.get("id") == service_id: enterprise_id = svc.get("enterpriseId", "") break if not enterprise_id: enterprise_id = (data.get("enterprises") or [{}])[0].get("id", "") tz_name = (data.get("enterprises") or [{}])[0].get("ianaTimeZoneIdentifier", "Europe/London") # Adult age category: classification=="Adult" or no maximalAge, filtered to this service age_cat_id = "" for cat in data.get("ageCategories") or []: if cat.get("serviceId") != service_id: continue if cat.get("classification") == "Adult" or cat.get("maximalAge") is None: age_cat_id = cat["id"] break params["enterprise_id"] = enterprise_id params["service_id"] = service_id params["age_category_id"] = age_cat_id params["_tz"] = ZoneInfo(tz_name) return bool(enterprise_id and service_id) class MewsProfile(BaseProfile): name = "mews" label = "Mews" required_params = [ {"key": "booking_engine_id", "label": "Booking Engine ID", "help": "UUID from the /distributor/ URL, e.g. d0b39796-92ee-4664-bb50-aa4d00eb0c0c"}, ] @classmethod def detect(cls, url: str) -> dict | None: m = re.search(r'app\.mews\.com/distributor/([0-9a-f-]{36})', url) if m: return {"booking_engine_id": m.group(1)} return None async def fetch_category_names(self, client, params: dict) -> tuple[dict, dict]: if not await _ensure_config(client, params): return {}, {} tz = params.get("_tz", ZoneInfo("Europe/London")) today = date.today() r = await client.post( f"{API_BASE}/services/getCalendarData", json={ "bookingEngineId": params["booking_engine_id"], "serviceId": params["service_id"], "startUtc": _midnight_utc(today, tz), "endUtc": _midnight_utc(today + timedelta(days=7), tz), "client": CLIENT, "session": _session(), }, headers=HEADERS, timeout=20, ) if r.status_code != 200: return {}, {} data = r.json() lang = "en-US" room_names = { cat["id"]: cat["name"].get(lang, cat["id"]) for cat in data.get("resourceCategories", []) if "name" in cat } rate_names = { rate["id"]: rate["name"].get(lang, rate["id"]) for rate in data.get("rates", []) if "name" in rate } return room_names, rate_names async def fetch_arrival_dates(self, client, params: dict) -> list[str]: if not await _ensure_config(client, params): return [] tz = params["_tz"] today = date.today() r = await client.post( f"{API_BASE}/services/getAvailability", json={ "bookingEngineId": params["booking_engine_id"], "startUtc": _midnight_utc(today, tz), "endUtc": _midnight_utc(today + timedelta(days=DAYS_AHEAD), tz), "client": CLIENT, "session": _session(), }, headers=HEADERS, timeout=30, ) if r.status_code != 200: return [] data = r.json() time_units = data.get("timeUnitStartsUtc", []) cat_avails = data.get("categoryAvailabilities", []) avail_by_date: dict[str, dict[str, int]] = {} for cat in cat_avails: cat_id = cat["categoryId"] for i, utc_str in enumerate(time_units): local_date = _utc_to_local_date(utc_str, tz) avail_by_date.setdefault(local_date, {})[cat_id] = cat["availabilities"][i] params["_avail"] = avail_by_date return sorted(d for d, cats in avail_by_date.items() if any(v > 0 for v in cats.values())) async def fetch_night_rates(self, client, params: dict, arrival: str, nights: int = 1) -> list[dict]: if not await _ensure_config(client, params): return [] tz = params.get("_tz", ZoneInfo("Europe/London")) arrival_date = date.fromisoformat(arrival) age_cat_id = params.get("age_category_id") or None r = await client.post( f"{API_BASE}/services/getPricing", json={ "configurationId": params["booking_engine_id"], "startUtc": _midnight_utc(arrival_date, tz), "endUtc": _midnight_utc(arrival_date + timedelta(days=nights), tz), "occupancyData": [{"ageCategoryId": age_cat_id, "personCount": 2}], "client": CLIENT, "session": _session(), }, headers=HEADERS, timeout=20, ) if r.status_code != 200: return [] avail_cache = params.get("_avail", {}).get(arrival, {}) rooms = [] for cat_price in r.json().get("categoryPrices", []): cat_id = cat_price["categoryId"] avail = avail_cache.get(cat_id, 1) for occ in cat_price.get("occupancyPrices", []): for rg in occ.get("rateGroupPrices", []): rate_id = rg.get("minRateId", "") amount = rg.get("minPrice", {}).get("totalAmount", {}) gross = amount.get("grossValue") net = amount.get("netValue") if gross is None: continue rooms.append({ "roomId": cat_id, "rateId": rate_id, "availability": avail, "prices": [ {"amountBeforeTax": net or gross, "amountAfterTax": gross} for _ in range(nights) ], "currencyCode": amount.get("currency", "GBP"), }) return rooms