spb/rent-ka

1# -----------------------------------------------------------------------------2# Rent-Ka — Rental listings aggregator (Canada, outside Québec)3# Author: 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_RENTKA_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_RENTKA"),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 (Scrapfly render_js) : enregistrer le HTML rendu, clé = URL110    # cible (les args supplémentaires — wait_ms — n'entrent pas dans la clé)111    orig_rendered = inst.get_rendered112113    def _rec_rendered(url: str, *args, **kw) -> str:114        html = orig_rendered(url, *args, **kw)115        key = _req_key("RENDERED", url, None)116        fname = key + ".html"117        (out_dir / fname).write_text(118            _redact(html.encode("utf-8")).decode("utf-8"), encoding="utf-8")119        store[key] = {"method": "RENDERED", "url": url, "status": 200,120                      "content_type": "text/html", "file": fname}121        return html122123    inst.get_rendered = _rec_rendered  # type: ignore[method-assign]124125    listings = inst.fetch()126    (out_dir / "index.json").write_text(127        json.dumps(store, ensure_ascii=False, indent=1), encoding="utf-8")128    (out_dir / "expected.json").write_text(129        json.dumps(snapshot(listings), ensure_ascii=False, indent=1),130        encoding="utf-8")131    return {"source": source_id, "requests": len(store),132            "listings": len(listings)}133134135def snapshot(listings: list[Listing]) -> dict:136    """Instantané compact et déterministe du parsing (champs bruts, pré-finalize).137138    La date de disponibilité normalisée dépend du jour d'exécution : on ne139    snapshotte que les champs stables pour éviter les tests fragiles.140    """141    rows = []142    for l in sorted(listings, key=lambda x: x.uid):143        rows.append({144            "uid": l.uid, "url": l.url, "title": l.title,145            "address": l.address, "sector": l.sector, "city": l.city,146            "unit_type": l.unit_type, "price": l.price,147            "availability": l.availability, "area_sqft": l.area_sqft,148            "n_images": len(l.images), "n_amenities": len(l.amenities),149        })150    return {"count": len(rows), "listings": rows}151152153# ---------------------------------------------------------------------------154# Rejeu155# ---------------------------------------------------------------------------156157class _ReplayAdapter(BaseAdapter):158    def __init__(self, index: dict, fixtures_dir: Path):159        super().__init__()160        self.index = index161        self.dir = fixtures_dir162163    def send(self, request, **kwargs):  # type: ignore[override]164        key = _req_key(request.method or "GET", request.url or "", request.body)165        meta = self.index.get(key)166        if meta is None:167            raise MissingFixture(f"{request.method} {request.url}")168        resp = requests.models.Response()169        resp.status_code = meta["status"]170        resp._content = (self.dir / meta["file"]).read_bytes()171        resp.headers["Content-Type"] = meta.get("content_type", "")172        if meta.get("location"):173            resp.headers["Location"] = meta["location"]174        resp.url = request.url or ""175        resp.request = request176        resp.encoding = "utf-8"177        return resp178179    def close(self):180        pass181182183def replay_connector(source_id: str):184    """Instancie le connecteur branché sur ses fixtures (aucun réseau)."""185    from .connectors import CONNECTORS186    cls = CONNECTORS[source_id]187    fdir = FIXTURES_DIR / source_id188    index = json.loads((fdir / "index.json").read_text(encoding="utf-8"))189190    inst = cls()191    inst.request_delay = 0.0             # pas de politesse hors-ligne192    inst.use_detail_cache = False        # tests indépendants de la BD193    adapter = _ReplayAdapter(index, fdir)194    inst.session.mount("https://", adapter)195    inst.session.mount("http://", adapter)196197    def _replay_rendered(url: str, *args, **kw) -> str:198        key = _req_key("RENDERED", url, None)199        meta = index.get(key)200        if meta is None:201            raise MissingFixture(f"RENDERED {url}")202        return (fdir / meta["file"]).read_text(encoding="utf-8")203204    inst.get_rendered = _replay_rendered  # type: ignore[method-assign]205    return inst206207208def recorded_sources() -> list[str]:209    """Sources ayant des fixtures enregistrées (pour paramétrer les tests)."""210    if not FIXTURES_DIR.exists():211        return []212    return sorted(p.name for p in FIXTURES_DIR.iterdir()213                  if (p / "index.json").exists())214