# ============================================================================== # Author: Simon-Pierre Boucher # File: src/main.py (ka-apartments-com) # Desc: Apartments.com — logements à louer, province de Québec. # Phase 1 : pagination des pages de recherche (/qc/ = toute la # province, HTML server-rendered ; arrêt quand le ne porte # plus « - Page N », le site retombant silencieusement sur la page 1). # Phase 2 : fiches détail en français (/fr/<slug>/<id>/) : GPS, # description, plans d'unités (pricingGridItem), commodités, galerie. # Items poussés : kind="listing" (placard) et kind="detail". # ============================================================================== from __future__ import annotations import asyncio import html as htmllib import re from apify import Actor from .net import BrightData # --- recherche (placards) ----------------------------------------------------- _ARTICLE_RX = re.compile( r"<article\b(?P<attrs>[^>]*\bdata-listingid=\"[^\"]+\"[^>]*)>" r"(?P<body>.*?)</article>", re.S) _ATTR_RX = re.compile(r"data-(listingid|url|streetaddress)=\"([^\"]*)\"") _TITLE_TAG_RX = re.compile(r"<title>([^<]*)") _PLACARD_TITLE_RX = re.compile(r"js-placardTitle[^\"]*\"[^>]*>([^<]+)<") _ADDRESS_RX = re.compile(r"property-address[^\"]*\"[^>]*>([^<]+)<") _BED_RENT_RX = re.compile( r"bedTextBox\">([^<]*)<.{0,400}?priceTextBox\">\s*]*>([^<]*)<", re.S) _IMG_COUNT_RX = re.compile(r"js-spnImgCount\">(\d+)") _PHONE_RX = re.compile(r"phone-data=\"(\d{7,15})\"") _AMENITIES_BLOCK_RX = re.compile( r"property-amenities\"[^>]*>(.*?)

", re.S) _SPAN_RX = re.compile(r"]*>([^<]+)") _IMG_RX = re.compile(r"https://images1\.apartments\.com/i2/[^\"'\s?]+") _QC_SLUG_RX = re.compile(r"apartments\.com/[^\"]*-qc[-/]") def _clean(text: str) -> str: return re.sub(r"\s+", " ", htmllib.unescape(text or "")).strip() def parse_search(page_html: str) -> list[dict]: """Placards d'une page de recherche → dicts (propriétés QC seulement).""" out: list[dict] = [] for m in _ARTICLE_RX.finditer(page_html): attrs = dict(_ATTR_RX.findall(m.group("attrs"))) pid, url = attrs.get("listingid", ""), attrs.get("url", "") if not pid or not url: continue body = m.group("body") title_m = _PLACARD_TITLE_RX.search(body) addr_m = _ADDRESS_RX.search(body) title = _clean(title_m.group(1)) if title_m else "" address = _clean(addr_m.group(1)) if addr_m else "" if not address and ", QC" in title: address = title # gabarit silverpropres : adresse dans le titre # nearby hors-Québec glissé en bas de page → on filtre if ", QC" not in address and not _QC_SLUG_RX.search(url): continue rents = [{"beds": _clean(b), "price": _clean(p)} for b, p in _BED_RENT_RX.findall(body)] img_count_m = _IMG_COUNT_RX.search(body) phone_m = _PHONE_RX.search(body) amen_m = _AMENITIES_BLOCK_RX.search(body) amenities = [_clean(s) for s in _SPAN_RX.findall(amen_m.group(1))] \ if amen_m else [] img_m = _IMG_RX.search(body) out.append({ "id": pid, "url": url, "title": title, "address": address, "street": _clean(attrs.get("streetaddress", "")), "rents": rents, "img_count": int(img_count_m.group(1)) if img_count_m else 0, "phone": phone_m.group(1) if phone_m else "", "amenities": [a for a in amenities if a], "image": img_m.group(0) if img_m else "", }) return out def page_title(page_html: str) -> str: m = _TITLE_TAG_RX.search(page_html) return _clean(m.group(1)) if m else "" # --- fiche détail (version française) ------------------------------------------ _META_LAT_RX = re.compile( r"]*>([^<]+)") _ADDR_BLOCK_RX = re.compile( r"delivery-address\">([^<]+).*?([^<]+),\s*" r"\s*([^<]+)\s*" r"([^<]*)", re.S) _DESC_SECTION_RX = re.compile( r"]*id=\"descriptionSection\"[^>]*>(.*?)", re.S) _MODEL_WRAPPER_RX = re.compile( r"priceGridModelWrapper[^\"]*\"\s*data-rentalkey=\"([^\"]*)\"(.*?)" r"(?=priceGridModelWrapper|$)", re.S) _MODEL_NAME_RX = re.compile(r"modelName\">([^<]+)<") _RENT_LABEL_RX = re.compile(r"rentLabel\">\s*([^<]+)") _DETAILS_WRAPPER_RX = re.compile( r"detailsTextWrapper\">(.*?)\s*", re.S) _AVAIL_RX = re.compile(r"availabilityInfo\">([^<]+)<") _BEDS_BATHS_RX = re.compile(r"data-beds=\"([^\"]*)\"\s*data-baths=\"([^\"]*)\"") _SPEC_INFO_RX = re.compile(r"specInfo\">\s*([^<]+)") _TAG_RX = re.compile(r"<[^>]+>") def _to_float(text: str) -> float | None: try: return float(text.strip()) except (TypeError, ValueError): return None def parse_detail(page_html: str) -> dict: out: dict = {} name_m = _PROP_NAME_RX.search(page_html) if name_m: out["name"] = _clean(name_m.group(1)) addr_m = _ADDR_BLOCK_RX.search(page_html) if addr_m: out["street"] = _clean(addr_m.group(1)) out["city"] = _clean(addr_m.group(2)) out["state"] = _clean(addr_m.group(3)) out["zip"] = _clean(addr_m.group(4)) lat_m, lng_m = _META_LAT_RX.search(page_html), _META_LNG_RX.search(page_html) if lat_m and lng_m: out["lat"] = _to_float(lat_m.group(1)) out["lng"] = _to_float(lng_m.group(1)) desc_m = _DESC_SECTION_RX.search(page_html) if desc_m: lines = [_clean(t) for t in _TAG_RX.sub("\n", desc_m.group(1)).split("\n")] lines = [t for t in lines if t and not t.lower().startswith(("à propos", "about"))] if lines: out["description"] = "\n\n".join(dict.fromkeys(lines)) plans: list[dict] = [] for key, body in _MODEL_WRAPPER_RX.findall(page_html): plan: dict = {"key": key} nm = _MODEL_NAME_RX.search(body) if nm: plan["name"] = _clean(nm.group(1)) rl = _RENT_LABEL_RX.search(body) if rl: plan["rent"] = _clean(rl.group(1)) dw = _DETAILS_WRAPPER_RX.search(body) if dw: plan["details"] = [_clean(s) for s in _SPAN_RX.findall(dw.group(1) + "")] bb = _BEDS_BATHS_RX.search(body) if bb: plan["beds"] = _to_float(bb.group(1)) plan["baths"] = _to_float(bb.group(2)) av = _AVAIL_RX.search(body) if av: plan["availability"] = _clean(av.group(1)) if plan.get("name") or plan.get("rent"): # la grille est présente deux fois dans la page (desktop/mobile) if not any(p.get("key") == plan.get("key") and p.get("name") == plan.get("name") and p.get("rent") == plan.get("rent") for p in plans): plans.append(plan) if plans: out["plans"] = plans amenities = [_clean(s) for s in _SPEC_INFO_RX.findall(page_html)] if amenities: out["amenities"] = list(dict.fromkeys(a for a in amenities if a)) images = list(dict.fromkeys( u for u in _IMG_RX.findall(page_html) if "-logo" not in u)) if images: out["images"] = images return out def fr_url(url: str) -> str: if "apartments.com/fr/" in url: return url return url.replace("www.apartments.com/", "www.apartments.com/fr/", 1) # --- orchestration -------------------------------------------------------------- async def _crawl_search(bd: BrightData, base: str, max_pages: int, props: dict[str, dict]) -> None: base = base if base.endswith("/") else base + "/" for page in range(1, max_pages + 1): url = base if page == 1 else f"{base}{page}/" page_html = await bd.get(url) if page_html is None: Actor.log.warning(f"recherche : échec fetch {url}, arrêt") break if page > 1 and f"- Page {page}" not in page_title(page_html): Actor.log.info(f"recherche : fin de pagination à la page {page} " f"({base})") break placards = parse_search(page_html) if not placards: Actor.log.info(f"recherche : page {page} vide ({base})") break fresh = [p for p in placards if p["id"] not in props] for p in fresh: props[p["id"]] = p await Actor.push_data({"kind": "listing", **p}) Actor.log.info(f"recherche p.{page} : {len(placards)} placards, " f"{len(fresh)} nouveaux ({base})") async def _fetch_detail(bd: BrightData, pid: str, url: str) -> None: page_html = await bd.get(fr_url(url)) if page_html is None: Actor.log.warning(f"détail {pid} : échec fetch") return detail = parse_detail(page_html) if not detail: Actor.log.warning(f"détail {pid} : page sans données") return await Actor.push_data({"kind": "detail", "id": pid, "url": url, **detail}) async def main() -> None: async with Actor: inp = await Actor.get_input() or {} token = (inp.get("brightdataToken") or "").strip() if not token: raise ValueError("brightdataToken requis (Web Unlocker)") bd = BrightData( token=token, zone=(inp.get("brightdataZone") or "web_unlocker1").strip(), concurrency=int(inp.get("concurrency") or 4), delay=float(inp.get("requestDelay") or 0)) try: search_urls = [u.strip() for u in (inp.get("searchUrls") or ["https://www.apartments.com/qc/"]) if u and u.strip()] max_pages = int(inp.get("maxPages") or 25) props: dict[str, dict] = {} for base in search_urls: await _crawl_search(bd, base, max_pages, props) Actor.log.info(f"recherche terminée : {len(props)} propriétés") if not inp.get("getDetails", True): return skip = set(inp.get("skipDetailIds") or []) targets: list[tuple[str, str]] = [ (pid, p["url"]) for pid, p in props.items() if pid not in skip] for enc in inp.get("extraDetailIds") or []: pid, _, url = enc.partition("|") if pid and url and pid not in props and pid not in skip: targets.append((pid, url)) max_details = int(inp.get("maxDetails") or 0) if max_details >= 0: targets = targets[:max_details] Actor.log.info(f"détails : {len(targets)} fiches à visiter") await asyncio.gather( *(_fetch_detail(bd, pid, url) for pid, url in targets)) finally: await bd.close()