spb/lou-ka Public
Lou·Ka — tous les logements à louer du Québec, un seul endroit.
HTML 99.7%
1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# fixtures.py : enregistrement/rejeu des réponses HTTP des sources.5# `python run.py record <source>` capture toutes les réponses (HTML/JSON)6# dans tests/fixtures/<source>/ + un instantané `expected.json` du parsing.7# Les tests (tests/test_connectors.py) rejouent ces fixtures hors-ligne :8# si un connecteur régresse, le test casse — sans toucher au site réel.9# -----------------------------------------------------------------------------10from __future__ import annotations1112import hashlib13import json14import re15from pathlib import Path1617import requests18from requests.adapters import BaseAdapter, HTTPAdapter1920from .schema import Listing2122ROOT = Path(__file__).resolve().parent.parent23FIXTURES_DIR = ROOT / "tests" / "fixtures"242526class MissingFixture(Exception):27 """Le connecteur a demandé une URL absente des fixtures enregistrées."""282930# Secrets tiers présents dans le HTML public des sites sources (clés Google31# Maps, jetons) : caviardés à l'enregistrement — les fixtures sont publiées32# sur GitHub et ces clés n'ont aucun rôle dans le parsing.33_SECRET_RES = [34 (re.compile(rb"AIza[0-9A-Za-z_-]{35}"), b"AIza_CLE_CAVIARDEE_LOUKA_000000000000000"),35 # jetons Mapbox (sk. = secret, pk. = public) embarqués par certains sites36 (re.compile(rb"(sk|pk)\.eyJ[A-Za-z0-9_.=-]+"), rb"\1.JETON_CAVIARDE_LOUKA"),37]383940def _redact(content: bytes) -> bytes:41 for rx, repl in _SECRET_RES:42 content = rx.sub(repl, content)43 return content444546def _req_key(method: str, url: str, body: bytes | str | None) -> str:47 if isinstance(body, str):48 body = body.encode("utf-8")49 h = hashlib.sha1()50 h.update(f"{method.upper()} {url} ".encode())51 if body:52 h.update(body)53 return h.hexdigest()[:20]545556def _ext_for(content_type: str) -> str:57 if "json" in content_type:58 return ".json"59 if "html" in content_type:60 return ".html"61 return ".txt"626364# ---------------------------------------------------------------------------65# Enregistrement66# ---------------------------------------------------------------------------6768class _RecordAdapter(HTTPAdapter):69 def __init__(self, store: dict, out_dir: Path):70 super().__init__()71 self.store = store72 self.out_dir = out_dir7374 def send(self, request, **kwargs): # type: ignore[override]75 resp = super().send(request, **kwargs)76 key = _req_key(request.method or "GET", request.url or "", request.body)77 ctype = resp.headers.get("Content-Type", "")78 fname = key + _ext_for(ctype)79 (self.out_dir / fname).write_bytes(_redact(resp.content))80 self.store[key] = {81 "method": request.method, "url": request.url,82 "status": resp.status_code, "content_type": ctype, "file": fname,83 }84 if resp.headers.get("Location"):85 self.store[key]["location"] = resp.headers["Location"]86 return resp878889def record(source_id: str) -> dict:90 """Exécute le connecteur en enregistrant chaque réponse HTTP en fixture.9192 Écrit tests/fixtures/<source>/{index.json, expected.json, *.html|json}.93 Retourne un résumé {source, requests, listings}.94 """95 from .connectors import CONNECTORS96 cls = CONNECTORS[source_id]97 out_dir = FIXTURES_DIR / source_id98 out_dir.mkdir(parents=True, exist_ok=True)99 for old in out_dir.iterdir():100 old.unlink()101102 inst = cls()103 inst.use_detail_cache = False # capturer aussi les pages détail104 store: dict = {}105 adapter = _RecordAdapter(store, out_dir)106 inst.session.mount("https://", adapter)107 inst.session.mount("http://", adapter)108109 # get_rendered (Firecrawl) : enregistrer le HTML rendu, clé = URL cible110 orig_rendered = inst.get_rendered111112 def _rec_rendered(url: str) -> str:113 html = orig_rendered(url)114 key = _req_key("RENDERED", url, None)115 fname = key + ".html"116 (out_dir / fname).write_text(117 _redact(html.encode("utf-8")).decode("utf-8"), encoding="utf-8")118 store[key] = {"method": "RENDERED", "url": url, "status": 200,119 "content_type": "text/html", "file": fname}120 return html121122 inst.get_rendered = _rec_rendered # type: ignore[method-assign]123124 listings = inst.fetch()125 (out_dir / "index.json").write_text(126 json.dumps(store, ensure_ascii=False, indent=1), encoding="utf-8")127 (out_dir / "expected.json").write_text(128 json.dumps(snapshot(listings), ensure_ascii=False, indent=1),129 encoding="utf-8")130 return {"source": source_id, "requests": len(store),131 "listings": len(listings)}132133134def snapshot(listings: list[Listing]) -> dict:135 """Instantané compact et déterministe du parsing (champs bruts, pré-finalize).136137 La date de disponibilité normalisée dépend du jour d'exécution : on ne138 snapshotte que les champs stables pour éviter les tests fragiles.139 """140 rows = []141 for l in sorted(listings, key=lambda x: x.uid):142 rows.append({143 "uid": l.uid, "url": l.url, "title": l.title,144 "address": l.address, "sector": l.sector, "city": l.city,145 "unit_type": l.unit_type, "price": l.price,146 "availability": l.availability, "area_sqft": l.area_sqft,147 "n_images": len(l.images), "n_amenities": len(l.amenities),148 })149 return {"count": len(rows), "listings": rows}150151152# ---------------------------------------------------------------------------153# Rejeu154# ---------------------------------------------------------------------------155156class _ReplayAdapter(BaseAdapter):157 def __init__(self, index: dict, fixtures_dir: Path):158 super().__init__()159 self.index = index160 self.dir = fixtures_dir161162 def send(self, request, **kwargs): # type: ignore[override]163 key = _req_key(request.method or "GET", request.url or "", request.body)164 meta = self.index.get(key)165 if meta is None:166 raise MissingFixture(f"{request.method} {request.url}")167 resp = requests.models.Response()168 resp.status_code = meta["status"]169 resp._content = (self.dir / meta["file"]).read_bytes()170 resp.headers["Content-Type"] = meta.get("content_type", "")171 if meta.get("location"):172 resp.headers["Location"] = meta["location"]173 resp.url = request.url or ""174 resp.request = request175 resp.encoding = "utf-8"176 return resp177178 def close(self):179 pass180181182def replay_connector(source_id: str):183 """Instancie le connecteur branché sur ses fixtures (aucun réseau)."""184 from .connectors import CONNECTORS185 cls = CONNECTORS[source_id]186 fdir = FIXTURES_DIR / source_id187 index = json.loads((fdir / "index.json").read_text(encoding="utf-8"))188189 inst = cls()190 inst.request_delay = 0.0 # pas de politesse hors-ligne191 inst.use_detail_cache = False # tests indépendants de la BD192 adapter = _ReplayAdapter(index, fdir)193 inst.session.mount("https://", adapter)194 inst.session.mount("http://", adapter)195196 def _replay_rendered(url: str) -> str:197 key = _req_key("RENDERED", url, None)198 meta = index.get(key)199 if meta is None:200 raise MissingFixture(f"RENDERED {url}")201 return (fdir / meta["file"]).read_text(encoding="utf-8")202203 inst.get_rendered = _replay_rendered # type: ignore[method-assign]204 return inst205206207def recorded_sources() -> list[str]:208 """Sources ayant des fixtures enregistrées (pour paramétrer les tests)."""209 if not FIXTURES_DIR.exists():210 return []211 return sorted(p.name for p in FIXTURES_DIR.iterdir()212 if (p / "index.json").exists())213