# ----------------------------------------------------------------------------- # Lou-Ka — Location court terme # connectors/mews.py : hôtels québécois en RÉSERVATION DIRECTE via le moteur # Mews (app.mews.com/distributor/) — 121 établissements du roster # data/hotels_engines.json (engine == "mews", GUID dans engine_ids.distributor). # 1 fiche StListing par HÔTEL (pas par chambre). # # API (rétro-ingéniérie du bundle distributor-app.js 5821.0.0, JSON pur) : # base POST https://api.mews.com/api/bookingEngine/v1 + corps JSON avec # auth ANONYME embarquée dans le payload. En-têtes Origin/Referer # app.mews.com OBLIGATOIRES (sinon ~60 % des hôtels : « Operation is not # enabled on your current subscription ») : # client = "Mews Distributor 5821.0.0" # session = (token + md5(token+client)).upper() où token = chaque caractère # de (4 octets aléatoires hex + nowUtc ISO) encodé en décimal # sur 3 chiffres (reproduction exacte du widget). # - /configurations/get {ids:[guid], primaryId:guid} # → bookingEngines (serviceId, languageCode, currencyCode), services, # enterprises (nom/description localisés, adresse, tél, courriel, # imageId/introImageId, ianaTimeZoneIdentifier, pricing Net|Gross), # ageCategories (Adult/Child par service). CACHE détail « v1 ». # - /resourceCategories/getAll {serviceId, extent:{imageAssignments:true}} # → catégories de chambres (name/description localisés, normalBedCount, # extraBedCount) + imageAssignments (imageId ordonnés). CACHE « v1 ». # - /services/getAvailability {serviceId, bookingEngineId, startUtc, endUtc, # enterpriseId, fullAmounts:false, languageCode} # → dispo par catégorie. ⚠️ startUtc doit être MINUIT LOCAL de l'hôtel # converti en UTC (« StartUtc is not start of TimeUnit » sinon) — # calculé via zoneinfo(ianaTimeZoneIdentifier). FRAIS à chaque sync. # - /services/getPricing {…, occupancyData:[{ageCategoryId, personCount:2}]} # → categoryPrices[].occupancyPrices[].rateGroupPrices[].minPrice # .totalAmount {grossValue, netValue}. FRAIS à chaque sync. # # Mapping : # - external_id = GUID distributor ; url = booking_url du roster (résa directe). # - Séjour témoin : arrivée à J+21, 1 nuit, 2 adultes ; price_night = tarif # minimum toutes catégories DISPONIBLES (netValue si pricing Net — l'affichage # du widget, taxes en sus — sinon grossValue). Aucun prix → fiche quand même. # - capacity = occupation max de la PLUS GRANDE catégorie (normal+extra) ; # nb de catégories dans details["room_types"]. # - Images : photo de l'hôtel + intro + galeries de chambres (cap 40), CDN # https://cdn.mews.com/media/image/?quality=85 (vérifié 200). # - Textes localisés : fr-CA > fr-* > en-US/en-GB > premier disponible ; # description de l'enterprise souvent vide → paragraphe éditorial assemblé # à partir des faits (ville/région, catégories, capacité, résa directe). # - region / citq / repli city : join du roster par GUID. # - Circuit breaker : après 10 échecs consécutifs de sondage prix, les fiches # suivantes sortent sans prix (l'API config, cachée, continue seule). # - Robustesse : try/except PAR hôtel — un GUID mort ne fait jamais planter # fetch() (log [mews] sur stderr). # ----------------------------------------------------------------------------- from __future__ import annotations import datetime as _dt import hashlib import json import os import sys from pathlib import Path from zoneinfo import ZoneInfo from ..schema import StListing from .base import StConnector from .sithotels import _property_type, _titre ROSTER = (Path(__file__).resolve().parent.parent.parent.parent / "data" / "hotels_engines.json") API = "https://api.mews.com/api/bookingEngine/v1" CLIENT = "Mews Distributor 5821.0.0" CDN = "https://cdn.mews.com/media/image/" MAX_IMAGES = 40 LEAD_DAYS = 21 # arrivée du séjour témoin (J+21, 1 nuit, 2 adultes) BREAKER_LIMIT = 10 # échecs prix consécutifs avant arrêt des sondages # spaceType Mews qui ne sont PAS de l'hébergement (vu : STATIONNEMENT vendu # comme catégorie ParkingSpot au Griffintown) — exclus des types de chambres, # de la capacité ET du minimum de prix NON_LODGING_SPACES = {"ParkingSpot", "MeetingRoom", "Office", "Desk", "Coworking", "EventVenue"} def _session_token() -> str: """Reproduit le jeton de session anonyme du widget : (4 octets aléatoires en hex + nowUtc ISO), chaque caractère encodé en décimal sur 3 chiffres.""" raw = os.urandom(4).hex() + _dt.datetime.now(_dt.timezone.utc) \ .strftime("%Y-%m-%dT%H:%M:%SZ") return "".join(f"{ord(c):03d}" for c in raw) def _loc(d: dict | None) -> str: """Choisit la meilleure variante localisée : fr-CA > fr-* > en > première.""" if not isinstance(d, dict) or not d: return "" for k in ("fr-CA", "fr-FR"): if d.get(k): return str(d[k]).strip() for k, v in d.items(): if k.lower().startswith("fr") and v: return str(v).strip() for k in ("en-US", "en-GB"): if d.get(k): return str(d[k]).strip() for v in d.values(): if v: return str(v).strip() return "" def _img(image_id: str, width: int = 1600) -> str: return f"{CDN}{image_id}?quality=85&width={width}" class Mews(StConnector): source_id = "mews" request_delay = 0.4 timeout = 40 def __init__(self) -> None: super().__init__() token = _session_token() self._auth = { "client": CLIENT, "session": (token + hashlib.md5( (token + CLIENT).encode()).hexdigest()).upper(), } # -- API ------------------------------------------------------------------ def _call(self, path: str, payload: dict, guid: str = "") -> dict: """POST bookingEngine v1. ⚠️ Origin + Referer OBLIGATOIRES : sans eux, ~60 % des hôtels (plans Mews restreints) répondent « Operation is not enabled on your current subscription » — vérifié live.""" resp = self.post(API + path, json={**self._auth, **payload}, headers={"Content-Type": "application/json", "X-Accept-Casing": "Camel", "Origin": "https://app.mews.com", "Referer": "https://app.mews.com/distributor/" + guid}) return resp.json() def _config(self, guid: str) -> dict: """Configuration + catégories de chambres (payload cachable « v1 »).""" conf = self._call("/configurations/get", {"ids": [guid], "primaryId": guid}, guid) engines = conf.get("bookingEngines") or [] engine = next((b for b in engines if b.get("id") == guid), engines[0] if engines else {}) svc_id = engine.get("serviceId") or "" rooms = {} if svc_id: rooms = self._call("/resourceCategories/getAll", {"serviceId": svc_id, "extent": {"imageAssignments": True}}, guid) return {"conf": conf, "rooms": rooms} # -- prix (frais à chaque sync : jamais caché) ------------------------------ def _probe_price(self, guid: str, conf: dict, engine: dict, enterprise: dict, lodging_ids: set[str] | None = None) -> float | None: """Tarif minimum (1 nuit à J+21, 2 adultes) toutes catégories dispo.""" svc_id = engine.get("serviceId") or "" ent_id = enterprise.get("id") or "" if not svc_id or not ent_id: return None try: tz = ZoneInfo(enterprise.get("ianaTimeZoneIdentifier") or "America/Montreal") except Exception: # noqa: BLE001 tz = ZoneInfo("America/Montreal") # ⚠️ startUtc doit être le début d'un TimeUnit = minuit LOCAL en UTC day = _dt.date.today() + _dt.timedelta(days=LEAD_DAYS) start = _dt.datetime(day.year, day.month, day.day, tzinfo=tz) \ .astimezone(_dt.timezone.utc) end = start + _dt.timedelta(days=1) fmt = "%Y-%m-%dT%H:%M:%SZ" lang = engine.get("languageCode") or "fr-CA" currency = engine.get("currencyCode") or "CAD" adult = next((a.get("id") for a in conf.get("ageCategories") or [] if a.get("serviceId") == svc_id and a.get("classification") == "Adult"), None) \ or next((a.get("id") for a in conf.get("ageCategories") or [] if a.get("serviceId") == svc_id), None) if not adult: return None # catégories réellement disponibles pour la nuit témoin available: set[str] | None = None try: av = self._call("/services/getAvailability", { "serviceId": svc_id, "bookingEngineId": guid, "startUtc": start.strftime(fmt), "endUtc": end.strftime(fmt), "enterpriseId": ent_id, "fullAmounts": False, "languageCode": lang, }, guid) available = {c.get("categoryId") for c in av.get("categoryAvailabilities") or [] if (c.get("availabilities") or [0])[0] > 0} except Exception: # noqa: BLE001 available = None # dispo inconnue : on prend le min global pricing = self._call("/services/getPricing", { "serviceId": svc_id, "bookingEngineId": guid, "enterpriseId": ent_id, "startUtc": start.strftime(fmt), "endUtc": end.strftime(fmt), "occupancyData": [{"ageCategoryId": adult, "personCount": 2}], "currencyCode": currency, "languageCode": lang, "fullAmounts": False, "categoryIds": None, "productIds": None, "voucherCode": None, "availabilityBlockId": None, }, guid) # affichage du widget : net (taxes en sus) si pricing Net, sinon brut net_first = (enterprise.get("pricing") or "Net") == "Net" best = None for cp in pricing.get("categoryPrices") or []: if lodging_ids is not None \ and cp.get("categoryId") not in lodging_ids: continue # stationnement, salle de réunion… if available and cp.get("categoryId") not in available: continue for op in cp.get("occupancyPrices") or []: for rg in op.get("rateGroupPrices") or []: ta = (rg.get("minPrice") or {}).get("totalAmount") or {} val = (ta.get("netValue") if net_first else ta.get("grossValue")) if val is None: val = ta.get("grossValue") if net_first \ else ta.get("netValue") if isinstance(val, (int, float)) and 20 <= val <= 20000 \ and (best is None or val < best): best = float(val) return best # -- une fiche par hôtel ----------------------------------------------------- def _listing(self, hotel: dict, price_ok: bool) -> tuple[StListing, bool]: """Construit la fiche ; retourne (listing, échec_du_sondage_prix).""" guid = str(hotel["engine_ids"]["distributor"]).strip() det = self.detail(guid, "v1", lambda g=guid: self._config(g)) conf = det.get("conf") or {} rooms = det.get("rooms") or {} engines = conf.get("bookingEngines") or [] engine = next((b for b in engines if b.get("id") == guid), engines[0] if engines else {}) svc_id = engine.get("serviceId") or "" service = next((s for s in conf.get("services") or [] if s.get("id") == svc_id), {}) enterprises = conf.get("enterprises") or [] enterprise = next((e for e in enterprises if e.get("id") == service.get("enterpriseId")), enterprises[0] if enterprises else {}) if not enterprise: raise ValueError("configuration Mews sans enterprise") addr = enterprise.get("address") or {} title = _loc(enterprise.get("name")) or _titre(hotel.get("name") or "") city = str(addr.get("city") or "").strip() or hotel.get("city") or "" # catégories de chambres (hébergement seulement) : capacité, noms, galerie cats = sorted((c for c in rooms.get("resourceCategories") or [] if c.get("spaceType") not in NON_LODGING_SPACES), key=lambda c: c.get("ordering") or 0) lodging_ids = {c.get("id") for c in cats} capacity = None room_names: list[str] = [] for c in cats: occ = (c.get("normalBedCount") or 0) + (c.get("extraBedCount") or 0) if occ and (capacity is None or occ > capacity): capacity = occ nom = _loc(c.get("name")) if nom and nom not in room_names: room_names.append(nom) images: list[str] = [] for iid in (enterprise.get("imageId"), enterprise.get("introImageId")): if iid: u = _img(iid) if u not in images: images.append(u) cat_order = {c.get("id"): c.get("ordering") or 0 for c in cats} assigns = sorted((a for a in rooms.get("imageAssignments") or [] if a.get("categoryId") in lodging_ids), key=lambda a: (cat_order.get(a.get("categoryId"), 99), a.get("ordering") or 0)) for a in assigns: if len(images) >= MAX_IMAGES: break if a.get("imageId"): u = _img(a["imageId"]) if u not in images: images.append(u) # prix frais (jamais caché) — sauté si le circuit breaker est ouvert ; # un échec du sondage ne fait jamais tomber la fiche (prix absent) price, price_error = None, False if price_ok: try: price = self._probe_price(guid, conf, engine, enterprise, lodging_ids or None) except Exception as exc: # noqa: BLE001 price_error = True print(f"[mews] prix {title} ({guid}) : {exc}", file=sys.stderr) ptype = _property_type(title) # « Saguenay--Lac-Saint-Jean » (variante SIT) → tiret cadratin canon region = str(hotel.get("region") or "").replace("--", "–") # description : celle de l'hôtel si renseignée, sinon assemblage factuel description = _loc(enterprise.get("description")) if not description: feminin = ptype == "Auberge" phrases = [f"{title} est un{'e' if feminin else ''} " f"{ptype.lower()}" + (f" situé{'e' if feminin else ''} à {city}" if city else "") + (f", dans la région {region}" if region else "") + "."] if room_names: phrases.append(f"{len(room_names)} type" f"{'s' if len(room_names) > 1 else ''} " "d'unités : " + ", ".join(room_names[:8]) + ("…" if len(room_names) > 8 else "") + ".") if capacity: phrases.append("Les plus grandes unités accueillent jusqu'à " f"{capacity:g} personnes.") phrases.append("Réservation directe en ligne auprès de " "l'établissement (moteur Mews).") if hotel.get("citq"): phrases.append("Établissement d'hébergement touristique " f"enregistré (no CITQ {hotel['citq']}).") description = " ".join(phrases) details: dict = {"booking_engine": "mews"} if cats: details["room_types"] = len(cats) if room_names: details["room_type_names"] = ", ".join(room_names[:12]) if enterprise.get("telephone"): details["phone"] = enterprise["telephone"] if enterprise.get("email"): details["email"] = enterprise["email"] if addr.get("postalCode"): details["postal_code"] = addr["postalCode"] if hotel.get("website"): details["website"] = hotel["website"] # géoloc/adresse : Mews d'abord, sinon repli sur le SIT (via le roster, # enrichi lat/lng/address pour les 1 174 hôtels sondés) lat, lng = addr.get("latitude"), addr.get("longitude") if lat is None or lng is None: lat, lng = hotel.get("lat"), hotel.get("lng") address = str(addr.get("line1") or "").strip() \ or str(hotel.get("address") or "").strip() return StListing( source=self.source_id, external_id=guid, url=hotel.get("booking_url") or f"https://app.mews.com/distributor/{guid}", title=title, property_type=ptype, address=address, city=city, region=region, price_night=price, price_label=(f"À partir de {price:g} $ / nuit" if price else ""), capacity=float(capacity) if capacity else None, citq=str(hotel.get("citq") or "").strip(), description=description[:5000], details=details, images=images, lat=float(lat) if lat is not None else None, lng=float(lng) if lng is not None else None, ), price_error # -- contrat ---------------------------------------------------------------- def fetch(self) -> list[StListing]: try: roster = json.loads(ROSTER.read_text(encoding="utf-8")) except Exception as exc: # noqa: BLE001 print(f"[mews] roster illisible ({ROSTER}) : {exc}", file=sys.stderr) return [] hotels = [h for h in roster.get("hotels") or [] if h.get("engine") == "mews" and (h.get("engine_ids") or {}).get("distributor")] listings: list[StListing] = [] vus: set[str] = set() price_failures = 0 # circuit breaker sur le sondage des prix for hotel in hotels: guid = str(hotel["engine_ids"]["distributor"]).strip() if not guid or guid in vus: continue vus.add(guid) try: lst, price_error = self._listing( hotel, price_ok=price_failures < BREAKER_LIMIT) except Exception as exc: # noqa: BLE001 print(f"[mews] {hotel.get('name')} ({guid}) : {exc}", file=sys.stderr) continue if price_error: price_failures += 1 if price_failures == BREAKER_LIMIT: print(f"[mews] {BREAKER_LIMIT} échecs de sondage de prix " "consécutifs : arrêt des requêtes prix (fiches " "sans prix ensuite)", file=sys.stderr) elif price_failures < BREAKER_LIMIT: price_failures = 0 listings.append(lst) return listings