# -----------------------------------------------------------------------------
# Lou-Ka — Location court terme
# connectors/airbnb.py : Airbnb (airbnb.ca) — source vedette, toute la province.
#
# Méthode : les pages de recherche Airbnb embarquent un JSON complet dans
# ',
html, re.S):
try:
data = json.loads(blob)
except ValueError:
continue
for entry in data.get("niobeClientData") or []:
if not (isinstance(entry, list) and len(entry) > 1
and isinstance(entry[1], dict)):
continue
node = ((entry[1].get("data") or {}).get("node") or {})
if isinstance(node.get("pdpPresentation"), dict):
pp = node["pdpPresentation"]
break
if pp:
break
if not pp:
return {}
out: dict = {}
# description longue : texte ORIGINAL de l'hôte (souvent français au
# Québec), repli sur la version traduite
desc = (pp.get("descriptions") or {}).get("longDescriptionHtml") or {}
txt = (desc.get("localizedString")
or desc.get("localizedStringWithTranslationPreference") or "")
if txt:
out["description"] = _html_to_text(txt)[:6000]
cap = pp.get("personCapacity")
if isinstance(cap, (int, float)) and 0 < cap <= 200:
out["capacity"] = float(cap)
# overview.items : "6 guests", "2 bedrooms", "3 beds", "2 baths"
ov = pp.get("overview") or {}
for item in ov.get("items") or []:
low = (item or "").lower()
m = _NUM_RE.search(low)
if not m:
continue
val = float(m.group(1))
if "guest" in low:
out.setdefault("capacity", val)
elif "bedroom" in low:
out["bedrooms"] = val
elif "bed" in low:
out["beds"] = val
elif "bath" in low:
out["bathrooms"] = val
if ov.get("title"):
out["overview_title"] = ov["title"]
# commodités disponibles (les groupes "Not included" ont available=False)
amen: list[str] = []
for grp in (pp.get("amenities") or {}).get("seeAllAmenitiesGroups") or []:
for a in grp.get("amenities") or []:
t = (a.get("title") or "").strip()
if a.get("available") and t and t not in amen:
amen.append(t)
if amen:
out["amenities"] = amen[:120]
# règles de la maison → animaux ("No pets", "Pets allowed", "2 pets…")
for grp in (pp.get("rules") or {}).get("groupItems") or []:
for it in grp.get("items") or []:
if it.get("type") != "HOUSE_RULES_PETS":
continue
t = (it.get("title") or "").lower()
if "no pets" in t or "pas d" in t or "aucun animal" in t:
out["pets"] = "non"
elif "pets allowed" in t or "animaux accept" in t:
out["pets"] = "oui"
elif t:
out["pets"] = "conditions"
loc = (pp.get("localizedLocation") or "").split(",")[0].strip()
if loc:
out["city"] = loc
return out
@staticmethod
def _apply_pdp(lst: StListing, p: dict) -> None:
"""Applique un payload détail sans écraser ce que la carte a fourni."""
if p.get("description") and not lst.description:
lst.description = p["description"]
if p.get("amenities") and not lst.amenities:
lst.amenities = list(p["amenities"])
for attr in ("capacity", "bedrooms", "beds", "bathrooms"):
if getattr(lst, attr) is None and p.get(attr) is not None:
setattr(lst, attr, p[attr])
if p.get("pets") and lst.pets is None:
lst.pets = p["pets"]
if p.get("city") and not lst.city:
lst.city = p["city"]
if not lst.property_type and p.get("overview_title"):
ptype, _ = _card_type_and_city(p["overview_title"])
lst.property_type = ptype
lst.finalize() # drapeaux/citq/pets dérivés du nouveau texte
def _enrich_details(self, listings: list[StListing]) -> None:
"""Visite les fiches détail via le cache self.detail() sous budget :
les hits de cache sont gratuits, seuls les fetchs réseau comptent."""
limit = max(0, int(os.environ.get("LOUKA_AIRBNB_DETAIL_LIMIT", "800")
or 800))
used = enriched = 0
streak = 0 # échecs réseau consécutifs
for lst in listings:
def fetch_fn(rid=lst.external_id):
nonlocal used, streak
if used >= limit or streak >= 8:
raise _DetailSkip
used += 1
html = self._pdp_html(rid)
if "data-deferred-state" not in html:
streak += 1
raise _DetailSkip # blocage/vide : pas de mise en cache
streak = 0
return self._parse_pdp(html)
try:
payload = self.detail(lst.external_id, _PDP_KEY, fetch_fn)
except _DetailSkip:
continue
except Exception: # noqa: BLE001 — une fiche ne bloque pas le run
continue
if payload:
self._apply_pdp(lst, payload)
enriched += 1
print(f"[airbnb] détail : {enriched} annonces enrichies"
f" ({used}/{limit} fetchs réseau, série d'échecs {streak})",
file=sys.stderr)
# -- parse JSON embarqué ----------------------------------------------------
@staticmethod
def _deferred_results(html: str) -> tuple[list[dict], list[str]]:
"""(searchResults, pageCursors) depuis les ',
html, re.S):
try:
data = json.loads(blob)
except ValueError:
continue
found: list = []
def walk(o):
if found:
return
if isinstance(o, dict):
res = ((o.get("staysSearch") or {}).get("results")
if isinstance(o.get("staysSearch"), dict) else None)
if isinstance(res, dict) and isinstance(
res.get("searchResults"), list):
found.append(res)
return
for v in o.values():
walk(v)
elif isinstance(o, list):
for v in o:
walk(v)
walk(data)
if found:
res = found[0]
cursors = (res.get("paginationInfo") or {}).get("pageCursors") or []
return res["searchResults"], list(cursors)
return [], []
# -- une carte → StListing --------------------------------------------------
def _to_listing(self, r: dict, region: str) -> StListing | None:
demand = r.get("demandStayListing") or {}
room_id = _b64_room_id(demand.get("id") or "")
if not room_id or not room_id.isdigit():
return None
name = (((demand.get("description") or {}).get("name") or {})
.get("localizedStringWithTranslationPreference")
or ((r.get("nameLocalized") or {})
.get("localizedStringWithTranslationPreference"))
or r.get("subtitle") or r.get("title") or "").strip()
if not name:
return None
card_title = (r.get("title") or "").strip()
ptype, city = _card_type_and_city(card_title)
# note / avis : "4.92 (128)"
rating = reviews = None
m = _RATING_RE.match(r.get("avgRatingLocalized") or "")
if m:
try:
rating = float(m.group(1).replace(",", "."))
reviews = int(re.sub(r"[\s,]", "", m.group(2)))
except ValueError:
rating = reviews = None
if rating is not None and not (0 < rating <= 5):
rating = reviews = None
# chambres / lits / sdb : structuredContent.primaryLine
bedrooms = beds = baths = None
sc = r.get("structuredContent") or {}
for line in (sc.get("primaryLine") or []):
body = (line.get("body") or "").lower()
mm = _NUM_RE.search(body)
if not mm:
continue
val = float(mm.group(1))
if "bedroom" in body or "chambre" in body:
bedrooms = val
elif re.search(r"\bbeds?\b|\blits?\b", body):
beds = val
elif "bath" in body or "salle" in body and "bain" in body:
baths = val
# prix : "5 nights x $277.60 CAD" (détail), sinon ligne « per night »
price_night, price_label = None, ""
sdp = r.get("structuredDisplayPrice") or {}
primary = sdp.get("primaryLine") or {}
blob = json.dumps(sdp, ensure_ascii=False)
mm = _NIGHTLY_RE.search(blob)
if mm:
try:
price_night = float(mm.group(2).replace(",", ""))
price_label = f"{mm.group(1)} nights x ${mm.group(2)} CAD"
except ValueError:
price_night = None
if price_night is None:
label = (primary.get("accessibilityLabel")
or primary.get("price") or "")
qual = (primary.get("qualifier") or "").lower()
mv = _MONEY_RE.search(label)
if mv and ("night" in qual or "night" in label.lower()
or "nuit" in label.lower()):
try:
price_night = float(mv.group(1).replace(",", ""))
price_label = label.strip()
except ValueError:
price_night = None
if not price_label:
price_label = (primary.get("accessibilityLabel") or "").strip()
coord = ((demand.get("location") or {}).get("coordinate") or {})
lat = coord.get("latitude")
lng = coord.get("longitude")
images = [p.get("picture") for p in (r.get("contextualPictures") or [])
if isinstance(p.get("picture"), str)
and p["picture"].startswith("http")]
details: dict = {}
badges = [b.get("text") for b in (r.get("badges") or []) if b.get("text")]
if badges:
details["badges"] = badges
if card_title:
details["card_title"] = card_title
return StListing(
source=self.source_id,
external_id=room_id,
url=f"https://www.airbnb.ca/rooms/{room_id}",
title=name,
property_type=ptype,
city=city,
region=region,
price_night=price_night,
price_label=price_label,
bedrooms=bedrooms,
beds=beds,
bathrooms=baths,
rating=rating,
reviews=reviews,
details=details,
images=images,
lat=lat if isinstance(lat, (int, float)) else None,
lng=lng if isinstance(lng, (int, float)) else None,
).finalize()
# -- quadtree ------------------------------------------------------------
@staticmethod
def _map_url(s: float, w: float, n: float, e: float) -> str:
zoom = min(18, max(4, round(math.log2(360.0 / max(e - w, 1e-6))) + 1))
return ("https://www.airbnb.ca/s/Qu%C3%A9bec--Canada/homes"
"?locale=en¤cy=CAD&search_by_map=true"
f"&ne_lat={n:.5f}&ne_lng={e:.5f}"
f"&sw_lat={s:.5f}&sw_lng={w:.5f}&zoom_level={zoom}")
def _ingest(self, cards: list[dict], out: list[StListing],
seen: set[str]) -> int:
added = 0
for card in cards:
coord = (((card.get("demandStayListing") or {})
.get("location") or {}).get("coordinate") or {})
lat, lng = coord.get("latitude"), coord.get("longitude")
if lat is not None and lng is not None and not _in_quebec(lat, lng):
continue # Ontario / États-Unis / NB dans le bbox
lst = self._to_listing(card, _region_from_latlng(lat, lng))
if lst is None or lst.external_id in seen:
continue
seen.add(lst.external_id)
out.append(lst)
added += 1
return added
# -- contrat ------------------------------------------------------------
def fetch(self) -> list[StListing]:
pages = max(1, int(os.environ.get("LOUKA_AIRBNB_PAGES", "15") or 15))
budget = max(1, int(os.environ.get("LOUKA_AIRBNB_BUDGET", "1200")
or 1200))
max_depth = max(0, int(os.environ.get("LOUKA_AIRBNB_DEPTH", "7") or 7))
s0, w0, n0, e0 = QC_BBOX
rows, cols = GRID0
dlat, dlng = (n0 - s0) / rows, (e0 - w0) / cols
stack: list[tuple[float, float, float, float, int]] = [
(s0 + i * dlat, w0 + j * dlng,
s0 + (i + 1) * dlat, w0 + (j + 1) * dlng, 0)
for i in range(rows) for j in range(cols)
]
random.shuffle(stack)
out: list[StListing] = []
seen: set[str] = set()
used = 0
while stack and used < budget:
cs, cw, cn, ce, depth = stack.pop()
if not _cell_touches_qc(cs, cw, cn, ce):
continue # cellule entièrement hors Québec
base = self._map_url(cs, cw, cn, ce)
used += 1
try:
html = self._search_html(base)
results, cursors = self._deferred_results(html)
except Exception as exc: # noqa: BLE001 — une cellule ne bloque pas les autres
print(f"[airbnb] cellule ({cs:.2f},{cw:.2f}) en échec : {exc}",
file=sys.stderr)
continue
if not results:
continue
if len(cursors) >= 15 and depth < max_depth:
# Cellule saturée (~270 résultats) → on garde la page 1 (déjà
# payée, dédoublonnée) et on subdivise en 4.
self._ingest(results, out, seen)
mlat, mlng = (cs + cn) / 2, (cw + ce) / 2
stack.extend([
(cs, cw, mlat, mlng, depth + 1),
(cs, mlng, mlat, ce, depth + 1),
(mlat, cw, cn, mlng, depth + 1),
(mlat, mlng, cn, ce, depth + 1),
])
print(f"[airbnb] cellule ({cs:.2f},{cw:.2f})→({cn:.2f},{ce:.2f})"
f" saturée → subdivision (prof. {depth + 1},"
f" req {used}/{budget})", file=sys.stderr)
continue
n_before = len(out)
self._ingest(results, out, seen)
for cur in cursors[1:pages]:
if used >= budget:
break
used += 1
try:
more, _ = self._deferred_results(self._search_html(
base + "&cursor=" + urllib.parse.quote(cur, safe="")))
except Exception: # noqa: BLE001
break
if not more:
break
self._ingest(more, out, seen)
print(f"[airbnb] cellule ({cs:.2f},{cw:.2f})→({cn:.2f},{ce:.2f}) :"
f" {len(out) - n_before} nouvelles (total {len(out)},"
f" req {used}/{budget})", file=sys.stderr)
if stack:
print(f"[airbnb] budget épuisé ({budget} req),"
f" {len(stack)} cellules non visitées", file=sys.stderr)
self._enrich_details(out)
return out