# ----------------------------------------------------------------------------- # Lou-Ka — Location court terme # connectors/chaletsdanslenord.py : Les Chalets dans le Nord # (leschaletsdanslenord.com — Laurentides) # # Petite agence familiale (Sainte-Lucie-des-Laurentides / lac Sarrazin, # ~6 chalets) — vitrine WordPress + moteur de réservation HOSTAWAY # (reservation.leschaletsdanslenord.com, compte 96792). # # Méthode : # 1. IDS : la racine du moteur Hostaway (Next.js rendu serveur) référence # tous les chalets via des liens "/listings/" ; la homepage WP donne # en plus prix (« dès N $ / nuit ») et lien de la fiche vitrine (l'id # Hostaway est dans l'URL des photos S3 `96792--…`). # 2. DÉTAIL (cache self.detail) : /listings/ du moteur embarque le JSON # complet dans le payload React Flight (`self.__next_f`) : prix de base # par nuit, lat/lng, ville, capacité, chambres, sdb, lits, type, note # (sur 10 → /2 par finalize), nb d'avis, ~50 photos, ~70 commodités et # description (référence Flight « $xx » résolue via les segments T). # External_id = id de listing Hostaway (stable, dans l'URL du moteur). # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from ..schema import StListing from .base import StConnector SITE = "https://leschaletsdanslenord.com" ENGINE = "https://reservation.leschaletsdanslenord.com" def _num(v) -> float | None: try: return float(v) if v not in (None, "") else None except (TypeError, ValueError): return None def _flight_blob(html: str) -> str: parts = [] for c in re.findall(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)', html): try: parts.append(json.loads(f'"{c}"')) except ValueError: continue return "".join(parts) def _flight_text(blob: str, ref: str) -> str: """Résout une référence texte Flight « $xx » (segment `xx:T,`, longueur en OCTETS utf-8).""" rid = ref.lstrip("$") m = re.search(rf"(?:^|\n){re.escape(rid)}:T([0-9a-f]+),", blob) if not m: return "" n = int(m.group(1), 16) raw = blob[m.end():].encode("utf-8")[:n] return raw.decode("utf-8", errors="ignore") class ChaletsDansLeNord(StConnector): source_id = "chaletsdanslenord" # -- ids + carte prix/urls vitrine -------------------------------------- def _engine_ids(self) -> list[str]: h = self.get(f"{ENGINE}/").text return sorted(set(re.findall(r'"/listings/(\d+)"', h))) def _wp_cards(self) -> dict[str, dict]: """id Hostaway → {url fiche vitrine, prix « dès N $ »} (homepage WP).""" try: h = self.get(f"{SITE}/").text except Exception: return {} cards: dict[str, dict] = {} for block in re.split(r'
  • dict: """Titre + description EN FRANÇAIS depuis la fiche vitrine WP.""" h = self.get(url).text d: dict = {} m = re.search(r"(?s)]*>(.*?)", h) if m: d["title"] = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", m.group(1))).strip() m = re.search(r"(?s)]*>(.*?)", h) if m: import html as _h paras, seen = [], set() for p in re.findall(r"(?s)]*>(.*?)

    ", m.group(1)): t = _h.unescape(re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", p))).strip() if len(t) < 60 or t in seen or "Voir les" in t[:30]: continue seen.add(t) paras.append(t) if len(paras) >= 10: break if paras: d["description"] = " ".join(paras)[:4000] return d # -- détail (moteur Hostaway + fiche vitrine FR) -------------------------- def _detail(self, hid: str, wp_url: str = "") -> dict: h = self.get(f"{ENGINE}/listings/{hid}").text blob = _flight_blob(h) i = blob.find(f'"listing":{{"id":{hid}') if i < 0: return {} obj, _ = json.JSONDecoder().raw_decode(blob[i + len('"listing":'):]) inner = obj.get("listing") or {} desc = str(inner.get("description") or "") if desc.startswith("$"): desc = _flight_text(blob, desc) desc = re.sub(r"\s+", " ", desc).strip() images = [] for ph in obj.get("listingImage") or []: u = (ph or {}).get("url") if u and u not in images: images.append(u) if len(images) >= 20: break # fiche vitrine WP : titre + description en français (prioritaires) wp: dict = {} if wp_url: try: wp = self._wp_fiche(wp_url) except Exception: wp = {} if wp.get("description"): desc = wp["description"] pt = ((inner.get("propertyType") or {}).get("name") or "").strip() return { "title": wp.get("title") or (inner.get("name") or "").strip(), "price": _num(inner.get("price")), "lat": _num(inner.get("lat")), "lng": _num(inner.get("lng")), "city": (inner.get("city") or "").strip(), "capacity": _num(inner.get("personCapacity")), "bedrooms": _num(inner.get("bedroomsNumber")), "beds": _num(inner.get("bedsNumber")), "bathrooms": _num(inner.get("bathroomsNumber")), "property_type": pt, "rating": _num(obj.get("averageReviewRating")), "reviews": obj.get("reviewsCount"), "description": desc[:4000], "amenities": [n for n in (((a.get("amenity") or {}).get("name") or a.get("name") or "").strip() for a in obj.get("listingAmenity") or [] if isinstance(a, dict)) if n][:80], "images": images, } # -- contrat ---------------------------------------------------------- def fetch(self) -> list[StListing]: cards = self._wp_cards() listings: list[StListing] = [] for hid in self._engine_ids(): card = cards.get(hid) or {} key = json.dumps([hid, card.get("price"), card.get("url")]) try: det = self.detail( hid, key, lambda i=hid, u=card.get("url") or "": self._detail(i, u)) except Exception: det = {} if not det.get("title"): continue price = det.get("price") or card.get("price") reviews = det.get("reviews") listings.append(StListing( source=self.source_id, external_id=hid, url=card.get("url") or f"{ENGINE}/listings/{hid}", title=det["title"], property_type=det.get("property_type") or "Chalet", city=det.get("city") or "", region="Laurentides", price_night=price, price_label=(f"à partir de {price:.0f} $ / nuit" if price else ""), capacity=det.get("capacity"), bedrooms=det.get("bedrooms"), beds=det.get("beds"), bathrooms=det.get("bathrooms"), rating=det.get("rating"), reviews=int(reviews) if reviews else None, description=det.get("description") or "", amenities=det.get("amenities") or [], details={"booking_url": f"{ENGINE}/listings/{hid}"}, images=det.get("images") or [], lat=det.get("lat"), lng=det.get("lng"), )) return listings