# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # recycled.py : détection des annonces RECYCLÉES — le même logement republié # (souvent plus cher) après un passage antérieur sur le marché. # # Méthode : les candidates sont les annonces INACTIVES du même immeuble # (building_key — même clé d'adresse que la déduplication). Chaque paire reçoit # un score de similarité 0–100 fondé sur des signaux indépendants : # numéro d'unité identique, type d'unité, chambres, superficie (±3 %), # photos identiques (hash SHA1 du contrôle d'images), texte de description # (similarité de Jaccard sur les mots). # Seules les correspondances ≥ SEUIL_AFFICHAGE sont retournées, avec les # signaux qui les fondent (inférence explicable, jamais de fusion automatique). # Cache : table recycled_cache (TTL 7 jours — recalcul léger, borné à # l'immeuble). # ----------------------------------------------------------------------------- from __future__ import annotations import json import re import time from . import db TTL = 7 * 86400 SEUIL_AFFICHAGE = 70 # score minimal pour présenter une correspondance MAX_CANDIDATES = 200 # garde-fou (tours à très fort volume) _WORD = re.compile(r"[a-zà-ÿ0-9]{3,}") def _tokens(text: str | None) -> set[str]: return set(_WORD.findall((text or "").lower())) def _jaccard(a: set[str], b: set[str]) -> float: if not a or not b: return 0.0 return len(a & b) / len(a | b) def _sha1_set(con, images_json: str | None) -> set[str]: try: urls = json.loads(images_json or "[]")[:20] except (ValueError, TypeError): return set() if not urls: return set() marks = ",".join("?" * len(urls)) return {r["sha1"] for r in con.execute( f"SELECT sha1 FROM image_checks WHERE url IN ({marks})" " AND sha1 IS NOT NULL", urls)} def _unit_of(address: str | None, city: str | None) -> str | None: from .dedup import _parse_address return _parse_address(address or "", city or "")[1] def _score(con, cur: dict, cand: dict) -> tuple[int, list[str]]: """Score 0–100 + signaux lisibles justifiant la correspondance.""" score = 0 signals: list[str] = [] if cur["unit"] and cand["unit"]: if cur["unit"] == cand["unit"]: score += 35 signals.append(f"même numéro d'unité ({cur['unit']})") else: return 0, [] # unités connues et différentes : pas le même if cur["unit_type"] and cand["unit_type"]: if cur["unit_type"] == cand["unit_type"]: score += 10 signals.append(f"même type ({cur['unit_type']})") else: return 0, [] if cur["bedrooms"] is not None and cand["bedrooms"] is not None: if cur["bedrooms"] == cand["bedrooms"]: score += 8 signals.append("même nombre de chambres") else: return 0, [] if cur["area"] and cand["area"]: if abs(cur["area"] - cand["area"]) <= 0.03 * max(cur["area"], cand["area"]): score += 15 signals.append(f"même superficie (≈{round(cand['area'])} pi²)") else: score -= 10 # photos identiques (hash du contenu, pas l'URL) sh_cur = _sha1_set(con, cur["images"]) sh_cand = _sha1_set(con, cand["images"]) if sh_cur and sh_cand: overlap = len(sh_cur & sh_cand) / min(len(sh_cur), len(sh_cand)) if overlap >= 0.5: score += 30 signals.append(f"{len(sh_cur & sh_cand)} photo(s) identique(s)") elif overlap > 0: score += 12 signals.append("photos partiellement identiques") # texte de description jac = _jaccard(_tokens(cur["description"]), _tokens(cand["description"])) if jac >= 0.7: score += 20 signals.append("description quasi identique") elif jac >= 0.45: score += 10 signals.append("description très similaire") # même source + même identifiant externe = republication certaine if cur["source"] == cand["source"]: score += 5 return min(100, score), signals def lookup(uid: str, con=None) -> dict: """Correspondances probables (annonces antérieures du même logement).""" own = con is None if own: con = db.connect() try: cached = con.execute( "SELECT matches, computed_at FROM recycled_cache WHERE uid=?", (uid,)).fetchone() if cached and time.time() - cached["computed_at"] < TTL: return json.loads(cached["matches"]) cur = con.execute( "SELECT uid, source, address, city, unit_type, bedrooms, price," " area_sqft AS area, description, images, first_seen, building_key," " dup_of FROM listings WHERE uid=?", (uid,)).fetchone() out: dict = {"matches": [], "statut": "inferred", "methode": ("similarité multi-signaux (unité, type, " "chambres, superficie, photos par hash, " "texte) parmi les annonces antérieures du " "même immeuble")} if cur is None or not cur["building_key"]: _store(con, uid, out) return out cur_d = dict(cur) cur_d["unit"] = _unit_of(cur["address"], cur["city"]) # exclure le groupe de doublons inter-sources courant (même annonce) dup_group = {uid, cur["dup_of"] or ""} for r in con.execute("SELECT uid FROM listings WHERE dup_of=?", (uid,)): dup_group.add(r["uid"]) cands = con.execute( "SELECT uid, source, address, city, unit_type, bedrooms, price," " area_sqft AS area, description, images, first_seen, last_seen" " FROM listings WHERE building_key=? AND active=0 AND uid<>?" " ORDER BY last_seen DESC LIMIT ?", (cur["building_key"], uid, MAX_CANDIDATES)).fetchall() matches = [] for c in cands: if c["uid"] in dup_group: continue # une annonce antérieure = terminée avant l'apparition de l'actuelle if (c["last_seen"] or 0) > (cur["first_seen"] or 0) + 7 * 86400: continue cd = dict(c) cd["unit"] = _unit_of(c["address"], c["city"]) score, signals = _score(con, cur_d, cd) if score >= SEUIL_AFFICHAGE: matches.append({ "uid": c["uid"], "source": c["source"], "prix": c["price"], "unit_type": c["unit_type"], "derniere_observation": c["last_seen"], "premiere_observation": c["first_seen"], "confiance": score, "signaux": signals, }) matches.sort(key=lambda m: -m["confiance"]) out["matches"] = matches[:5] _store(con, uid, out) return out finally: if own: con.close() def _store(con, uid: str, out: dict) -> None: con.execute( "INSERT INTO recycled_cache (uid, matches, computed_at) VALUES (?,?,?)" " ON CONFLICT(uid) DO UPDATE SET matches=excluded.matches," " computed_at=excluded.computed_at", (uid, json.dumps(out, ensure_ascii=False), time.time())) con.commit()