# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # managers.py : « Qui gère ce logement? » — fiche du gestionnaire/propriétaire. # # 1) Amorçage : les gestionnaires sont les sources DIRECTES de Lou-Ka # (data/sources.json, hors portails et petites annonces) qui ont au moins # une annonce collectée. Aucun gestionnaire n'est inventé. # 2) Résolution Google Maps (SerpApi, serveur seulement) : la fiche Google # n'est associée QUE si la confiance est suffisante (similarité de nom, # domaine du site web, ville) — jamais « le premier résultat » aveuglément. # Signaux et confiance stockés (match_confidence, match_signals). # 3) Avis Google : synchronisation paginée (next_page_token, hl=fr, plus # récents d'abord), dédupliqués par (manager_id, review_id Google) avec # repli sur un hash de contenu. Les avis vivent dans NOTRE base : les pages # ne déclenchent jamais d'appel SerpApi. # 4) Analyse : statistiques réelles (distribution, récence, tendance — pas # seulement la moyenne Google) + thèmes par lexique français, sentiment # dérivé de la note, sévérité et confiance. Statut « inferred » assumé. # ----------------------------------------------------------------------------- from __future__ import annotations import difflib import hashlib import json import re import time import unicodedata from datetime import datetime, timezone from . import db, serpapi_client from .dedup import CLASSIFIEDS, PORTALS VERSION = "gestionnaire-1.0" SEUIL_CONFIANCE = 0.60 # sous ce seuil : fiche Google NON associée MAX_PAGES = 10 # garde-fou pagination avis (par synchronisation) SYNC_TTL = 7 * 86400 # re-synchroniser les avis au plus aux 7 jours RESOLVE_RETRY = 30 * 86400 # re-tenter une résolution échouée après 30 jours # suffixes/génériques retirés pour comparer les noms d'entreprises _LEGAL = re.compile( r"\b(inc|ltee|ltée|ltd|senc|s\.e\.n\.c|enr|corp|corporation|groupe|group|" r"gestion|gestions|immobilier|immobiliere|immobilière|immeubles?|" r"appartements?|apartments?|properties|proprietes|propriétés|residences?|" r"résidences?|habitations?|siege|social|bureau|head|office|" r"les|le|la|de|du|des|et|and)\b") # types Google d'une fiche d'ENTREPRISE (vs fiche d'un immeuble précis) _CORP_TYPES = re.compile( r"agence immobili|gestion immobili|property management|real estate|" r"societe de gestion|immobilier", re.I) def _norm_name(s: str | None) -> str: s = unicodedata.normalize("NFKD", (s or "").lower()) s = "".join(c for c in s if not unicodedata.combining(c)) s = re.sub(r"[^a-z0-9 ]+", " ", s) s = _LEGAL.sub(" ", s) return " ".join(s.split()) def _domain(url: str | None) -> str | None: if not url: return None m = re.search(r"^(?:https?://)?(?:www\.)?([^/:?#]+)", url.strip().lower()) return m.group(1) if m else None # --- 1) amorçage depuis sources.json ---------------------------------------- def seed(con=None) -> dict: """Crée/actualise un gestionnaire par source directe ayant des annonces.""" own = con is None if own: con = db.connect() try: sources = json.loads((db.DB_PATH.parent / "sources.json") .read_text("utf-8"))["sources"] with_ads = {r["source"] for r in con.execute( "SELECT DISTINCT source FROM listings")} n = 0 for s in sources: sid = s.get("id") if not sid or sid in PORTALS or sid in CLASSIFIEDS: continue if sid not in with_ads and s.get("connector") not in with_ads: continue con.execute( "INSERT INTO managers (source_id, name, website, aliases)" " VALUES (?,?,?,?)" " ON CONFLICT(source_id) DO UPDATE SET name=excluded.name," " website=excluded.website", (sid, s.get("name"), s.get("url"), json.dumps({"region": s.get("region"), "connector": s.get("connector")}, ensure_ascii=False))) n += 1 con.commit() return {"managers": n} finally: if own: con.close() # --- 2) résolution Google Maps (jamais « le premier résultat ») -------------- def _score_candidate(mgr: dict, cand: dict) -> tuple[float, dict]: signals: dict = {} sim = difflib.SequenceMatcher( None, _norm_name(mgr["name"]), _norm_name(cand.get("title"))).ratio() signals["similarite_nom"] = round(sim, 3) conf = 0.6 * sim d_mgr, d_cand = _domain(mgr.get("website")), _domain(cand.get("website")) if d_mgr and d_cand and d_mgr == d_cand: conf += 0.3 signals["meme_domaine_web"] = d_mgr region = None try: region = (json.loads(mgr.get("aliases") or "{}") or {}).get("region") except (ValueError, TypeError): pass addr = (cand.get("address") or "").lower() if region and _norm_name(region) and _norm_name(region) in _norm_name(addr): conf += 0.1 signals["ville_correspondante"] = region if _CORP_TYPES.search(cand.get("type") or ""): signals["fiche_entreprise"] = cand.get("type") signals["type_fiche"] = cand.get("type") signals["nombre_avis"] = cand.get("reviews") return min(1.0, conf), signals def resolve(manager_id: int, con=None) -> dict: """Associe (ou refuse d'associer) la fiche Google Maps du gestionnaire.""" own = con is None if own: con = db.connect() try: mgr = con.execute("SELECT * FROM managers WHERE id=?", (manager_id,)).fetchone() if mgr is None: return {"ok": False, "raison": "gestionnaire inconnu"} mgr = dict(mgr) region = "" try: region = (json.loads(mgr.get("aliases") or "{}") or {}).get( "region") or "" except (ValueError, TypeError): pass q = f"{mgr['name']} {region} Québec".strip() data = serpapi_client.google_maps(q) cands = data.get("local_results") or [] if not cands and data.get("place_results"): cands = [data["place_results"]] for c in cands: # SerpApi renvoie parfois `type` en liste (multi-catégories) if isinstance(c.get("type"), list): c["type"] = ", ".join(str(t) for t in c["type"]) scored = [( *_score_candidate(mgr, c), c) for c in cands[:8]] best, best_conf, best_sig = None, 0.0, {} if scored: top = max(s[0] for s in scored) # une chaîne a souvent une fiche Google PAR IMMEUBLE en plus du # siège : parmi les candidats plausibles (confiance proche du # meilleur ET au-dessus du seuil), préférer la fiche d'entreprise # puis la plus représentative (le plus d'avis) near = [s for s in scored if s[0] >= max(SEUIL_CONFIANCE, top - 0.15)] if near: best_conf, best_sig, best = max( near, key=lambda s: (bool(_CORP_TYPES.search( s[2].get("type") or "")), s[2].get("reviews") or 0, s[0])) else: best_conf, best_sig, best = max(scored, key=lambda s: s[0]) now = time.time() if best is None or best_conf < SEUIL_CONFIANCE: con.execute( "UPDATE managers SET resolve_failed=1, resolved_at=?," " match_confidence=?, match_signals=? WHERE id=?", (now, round(best_conf, 3) if best else None, json.dumps({"refus": "confiance insuffisante", "candidats": len(cands), **best_sig}, ensure_ascii=False), manager_id)) con.commit() return {"ok": False, "raison": "confiance insuffisante", "confiance": round(best_conf, 3)} con.execute( "UPDATE managers SET gmaps_place_id=?, gmaps_data_id=?," " gmaps_name=?, gmaps_address=?, gmaps_rating=?, gmaps_reviews=?," " match_confidence=?, match_signals=?, resolved_at=?," " resolve_failed=0, phone=COALESCE(phone, ?) WHERE id=?", (best.get("place_id"), best.get("data_id"), best.get("title"), best.get("address"), best.get("rating"), best.get("reviews"), round(best_conf, 3), json.dumps(best_sig, ensure_ascii=False), now, best.get("phone"), manager_id)) con.commit() return {"ok": True, "gmaps_name": best.get("title"), "confiance": round(best_conf, 3)} finally: if own: con.close() # --- 3) synchronisation des avis (paginée, dédupliquée, en base) ------------- def _content_hash(author: str | None, rating, text: str | None) -> str: raw = f"{author or ''}|{rating}|{(text or '')[:200]}" return hashlib.sha1(raw.encode("utf-8")).hexdigest() def sync_reviews(manager_id: int, con=None, max_pages: int = MAX_PAGES) -> dict: """Rapatrie les avis Google (les plus récents d'abord) dans NOTRE base.""" own = con is None if own: con = db.connect() try: mgr = con.execute("SELECT * FROM managers WHERE id=?", (manager_id,)).fetchone() if mgr is None or not mgr["gmaps_data_id"]: return {"ok": False, "raison": "fiche Google non associée"} known = {r["external_review_id"] for r in con.execute( "SELECT external_review_id FROM manager_reviews WHERE manager_id=?", (manager_id,))} known_hash = {r["content_hash"] for r in con.execute( "SELECT content_hash FROM manager_reviews WHERE manager_id=?" " AND external_review_id IS NULL", (manager_id,))} token, new, pages = None, 0, 0 while pages < max_pages: data = serpapi_client.google_maps_reviews( mgr["gmaps_data_id"], next_page_token=token) reviews = data.get("reviews") or [] pages += 1 page_new = 0 now = time.time() for rv in reviews: ext = rv.get("review_id") text = rv.get("snippet") or rv.get("extracted_snippet", {}).get("original") user = rv.get("user") or {} chash = _content_hash(user.get("name"), rv.get("rating"), text) if (ext and ext in known) or (not ext and chash in known_hash): continue resp = rv.get("response") or {} analysis = _analyze_review(rv.get("rating"), text) con.execute( "INSERT INTO manager_reviews (manager_id," " external_review_id, source, rating, text, published_at," " relative_date_raw, author_name, author_review_count," " owner_response, owner_response_date, source_url," " fetched_at, content_hash, analysis) VALUES" " (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)" " ON CONFLICT(manager_id, external_review_id) DO NOTHING", (manager_id, ext, "google_maps", rv.get("rating"), text, rv.get("iso_date"), rv.get("date"), user.get("name"), user.get("reviews"), resp.get("snippet"), resp.get("iso_date") or resp.get("date"), rv.get("link"), now, chash, json.dumps(analysis, ensure_ascii=False))) if ext: known.add(ext) else: known_hash.add(chash) page_new += 1 new += page_new token = (data.get("serpapi_pagination") or {}).get( "next_page_token") # tri « plus récents d'abord » : une page entièrement connue # signifie qu'on a rejoint l'historique déjà synchronisé if not token or (reviews and page_new == 0): break stats = _aggregate(con, manager_id) con.execute("UPDATE managers SET review_stats=?, last_synced_at=?" " WHERE id=?", (json.dumps(stats, ensure_ascii=False), time.time(), manager_id)) con.commit() return {"ok": True, "nouveaux": new, "pages": pages, "total": stats.get("n")} finally: if own: con.close() # --- 4) analyse : thèmes (lexique fr), sentiment, sévérité, agrégats --------- TOPICS = { "entretien": ["entretien", "entreten", "maintenance", "neglige", "négligé", "delabre", "délabré", "moisissure"], "réparations": ["reparation", "réparation", "reparer", "réparer", "bris", "brise", "brisé", "fuite", "degat", "dégât", "plomberie"], "communication": ["communication", "repond", "répond", "reponse", "réponse", "rappel", "courriel", "joindre", "injoignable", "appel", "message"], "personnel": ["personnel", "employe", "employé", "concierge", "equipe", "équipe", "gerant", "gérant", "agent", "proprietaire", "propriétaire"], "propreté": ["propre", "proprete", "propreté", "sale", "salete", "saleté", "malpropre"], "chauffage": ["chauffage", "chauffe", "chauffé", "froid", "temperature", "température"], "eau chaude": ["eau chaude"], "vermine": ["punaise", "coquerelle", "souris", "rat ", "rats", "insecte", "vermine", "fourmi", "extermina"], "bruit": ["bruit", "bruyant", "tapage", "insonoris"], "sécurité": ["securite", "sécurité", "securitaire", "sécuritaire", "vol", "intrusion", "serrure", "camera", "caméra"], "ascenseur": ["ascenseur"], "stationnement": ["stationnement", "parking", "deneigement", "déneigement"], "dépôt et remboursement": ["depot", "dépôt", "rembours", "caution"], "augmentation de loyer": ["augmentation", "hausse"], "bail et location": ["bail", "signature", "visite", "location"], "service client": ["service", "clientele", "clientèle", "professionnel", "courtois", "arrogant", "impoli", "respect"], } _GRAVES = {"vermine", "sécurité", "chauffage", "eau chaude"} def _analyze_review(rating, text: str | None) -> dict: t = (text or "").lower() topics = sorted({name for name, kws in TOPICS.items() if any(k in t for k in kws)}) if rating is None: sentiment = "inconnu" elif rating <= 2: sentiment = "négatif" elif rating >= 4: sentiment = "positif" else: sentiment = "neutre" severity = "faible" if sentiment == "négatif": severity = "élevée" if any(x in _GRAVES for x in topics) else "moyenne" if len(t) >= 80 and topics: confidence = 0.7 elif t: confidence = 0.4 else: confidence = 0.2 return {"topics": topics, "sentiment": sentiment, "severite": severity, "confiance": confidence, "statut": "inferred", "methode": "lexique de thèmes fr + sentiment dérivé de la note"} def _aggregate(con, manager_id: int) -> dict: rows = con.execute( "SELECT rating, published_at, owner_response, analysis" " FROM manager_reviews WHERE manager_id=?", (manager_id,)).fetchall() rated = [r for r in rows if r["rating"] is not None] out: dict = {"n": len(rows), "version": VERSION, "statut": "calculated", "methode": ("statistiques calculées sur les avis Google " "synchronisés en base (pas seulement la moyenne " "affichée par Google) — l'échantillon peut être " "plus petit que le total annoncé par Google, il " "s'accumule à chaque synchronisation ; thèmes " "par lexique — inférence, pas une lecture " "humaine")} if not rated: return out out["moyenne"] = round(sum(r["rating"] for r in rated) / len(rated), 2) dist = {str(i): 0 for i in range(1, 6)} for r in rated: k = str(int(round(r["rating"]))) if k in dist: dist[k] += 1 out["distribution"] = dist out["pct_negatif"] = round(100 * (dist["1"] + dist["2"]) / len(rated)) out["pct_positif"] = round(100 * (dist["4"] + dist["5"]) / len(rated)) out["avec_reponse_proprietaire"] = sum( 1 for r in rows if r["owner_response"]) cutoff = datetime.now(timezone.utc).timestamp() - 365 * 86400 rec, old = [], [] for r in rated: ts = _iso_ts(r["published_at"]) (rec if ts and ts >= cutoff else old).append(r["rating"]) if rec: out["moyenne_12m"] = round(sum(rec) / len(rec), 2) out["n_12m"] = len(rec) if len(rec) >= 5 and len(old) >= 5: delta = out["moyenne_12m"] - sum(old) / len(old) out["tendance"] = ("en amélioration" if delta >= 0.3 else "en dégradation" if delta <= -0.3 else "stable") out["tendance_delta"] = round(delta, 2) themes: dict[str, dict] = {} for r in rows: try: a = json.loads(r["analysis"] or "{}") except (ValueError, TypeError): continue for tp in a.get("topics") or []: d = themes.setdefault(tp, {"mentions": 0, "negatif": 0, "positif": 0}) d["mentions"] += 1 if a.get("sentiment") == "négatif": d["negatif"] += 1 elif a.get("sentiment") == "positif": d["positif"] += 1 out["themes"] = dict(sorted(themes.items(), key=lambda kv: -kv[1]["mentions"])) out["plaintes_frequentes"] = [ k for k, v in sorted(themes.items(), key=lambda kv: -kv[1]["negatif"]) if v["negatif"] >= 2][:5] return out def _iso_ts(iso: str | None) -> float | None: if not iso: return None try: return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp() except ValueError: return None # --- fiche API + boucle de fond ---------------------------------------------- def fiche(source_id: str, con=None) -> dict | None: """Fiche gestionnaire pour l'API — tout vient de NOTRE base (0 SerpApi).""" own = con is None if own: con = db.connect() try: mgr = con.execute("SELECT * FROM managers WHERE source_id=?", (source_id,)).fetchone() if mgr is None: return None m = dict(mgr) n_active = con.execute( "SELECT COUNT(*) n FROM listings WHERE source=? AND active=1" " AND published=1 AND dup_of IS NULL", (source_id,)).fetchone()["n"] out = { "source_id": m["source_id"], "nom": m["name"], "site_web": m["website"], "telephone": m["phone"], "annonces_actives": n_active, "version": VERSION, } if m["gmaps_place_id"] and not m["resolve_failed"]: out["google_maps"] = { "nom": m["gmaps_name"], "adresse": m["gmaps_address"], "note": m["gmaps_rating"], "nombre_avis": m["gmaps_reviews"], "place_id": m["gmaps_place_id"], "confiance_association": m["match_confidence"], "signaux": json.loads(m["match_signals"] or "{}"), "statut": "inferred", "methode": ("fiche associée par similarité de nom, domaine " "web et ville — association refusée sous " f"{SEUIL_CONFIANCE:.2f} de confiance"), } try: out["avis"] = json.loads(m["review_stats"] or "null") except (ValueError, TypeError): out["avis"] = None out["avis_recents"] = [ {"note": r["rating"], "texte": r["text"], "date": r["published_at"] or r["relative_date_raw"], "auteur": r["author_name"], "reponse_proprietaire": bool(r["owner_response"]), "analyse": json.loads(r["analysis"] or "{}"), "statut": "observed"} for r in con.execute( "SELECT rating, text, published_at, relative_date_raw," " author_name, owner_response, analysis" " FROM manager_reviews WHERE manager_id=?" " ORDER BY published_at DESC, fetched_at DESC LIMIT 10", (m["id"],))] out["derniere_synchro"] = m["last_synced_at"] elif m["resolved_at"]: out["google_maps"] = { "statut": "non_associe", "note_methode": ("aucune fiche Google Maps n'a pu être " "associée avec une confiance suffisante — " "Lou-Ka préfère ne rien afficher plutôt que " "d'attribuer les avis d'une autre entreprise"), } return out finally: if own: con.close() def precompute(budget_s: float = 120.0, con=None) -> dict: """Boucle de fond : amorce, résout et synchronise dans un budget temps.""" if not serpapi_client.available(): return {"skipped": "SERPAPI_API_KEY absente"} own = con is None if own: con = db.connect() t0 = time.time() resolved = synced = errors = 0 try: seed(con) now = time.time() # 1) résolutions manquantes (ou échecs anciens à retenter) for r in con.execute( "SELECT id FROM managers WHERE resolved_at IS NULL" " OR (resolve_failed=1 AND resolved_at < ?)" " ORDER BY resolved_at IS NOT NULL, id", (now - RESOLVE_RETRY,)).fetchall(): if time.time() - t0 > budget_s: break try: if resolve(r["id"], con).get("ok"): resolved += 1 except serpapi_client.SerpApiError: errors += 1 break # panne API : ne pas insister ce tour-ci # 2) avis périmés (les jamais-synchronisés d'abord) for r in con.execute( "SELECT id FROM managers WHERE gmaps_data_id IS NOT NULL" " AND resolve_failed=0 AND (last_synced_at IS NULL" " OR last_synced_at < ?)" " ORDER BY last_synced_at IS NOT NULL, last_synced_at", (now - SYNC_TTL,)).fetchall(): if time.time() - t0 > budget_s: break try: if sync_reviews(r["id"], con).get("ok"): synced += 1 except serpapi_client.SerpApiError: errors += 1 break return {"resolved": resolved, "synced": synced, "errors": errors, "elapsed_s": round(time.time() - t0, 1)} finally: if own: con.close() if __name__ == "__main__": print(precompute())