# ============================================================================== # Author: Simon-Pierre Boucher # File: restoka/permits.py # Desc: Enrichissement RACJ — permis d'alcool en vigueur (Régie des alcools, # des courses et des jeux, Données Québec, licence CC-BY 4.0). # Télécharge le registre CSV des permis de détaillant d'alcool # (bar, restaurant pour vendre/servir…), regroupe par établissement et # croise de façon CONSERVATRICE avec les restaurants (mêmes règles que # les inspections MAPAQ : nom exact + lieu, ou postal + civique + nom # similaire — jamais fusionner deux établissements). Résultat dans # details.permis_alcool (catégories de permis, capacité, titulaire). # Gratuit, sans clé API. Cadence hebdomadaire (guard 6 jours). # ============================================================================== from __future__ import annotations import csv import io import json import re import sys import time from datetime import date import requests from .connectors.base import USER_AGENT from .inspections import (_CIVIC_RE, _NUMBERED_CO_RE, _name_similar, core_name, norm_name) from .regions import strip_accents SOURCE_ID = "racj" REFRESH_DAYS = 6 # au plus une fois par cycle hebdo CSV_URL = ("https://www.donneesquebec.ca/recherche/dataset/" "d817c9f7-76c7-44af-882d-0d673056ef86/resource/" "6b69360c-af8d-4c57-b5bf-1c38d5461de3/download/" "racj-alcool-detaillant.csv") def _download() -> list[dict]: resp = requests.get(CSV_URL, headers={"User-Agent": USER_AGENT}, timeout=120) resp.raise_for_status() text = None for enc in ("utf-8-sig", "latin-1"): try: text = resp.content.decode(enc) break except UnicodeDecodeError: continue if text is None: raise RuntimeError("encodage CSV RACJ inconnu") return list(csv.DictReader(io.StringIO(text))) def group_establishments(records: list[dict]) -> list[dict]: """Regroupe les lignes CSV (une par local/terrasse/permis) par établissement (NoEtablissement) : catégories de permis, capacité totale, nom, adresse, ville, code postal, titulaire.""" by_no: dict[str, dict] = {} for rec in records: no = (rec.get("NoEtablissement") or "").strip() if not no: continue e = by_no.setdefault(no, { "no": no, "nom": (rec.get("RaisonSociale") or "").strip(), "titulaire": (rec.get("Titulaire") or "").strip(), "adresse": (rec.get("Adresse") or "").strip(), "ville": (rec.get("Ville") or "").strip(), "postal": (rec.get("CodePostal") or "").replace(" ", "").upper(), "categories": set(), "capacite": 0, "permis": set(), }) cat = (rec.get("Categorie") or "").strip() if cat: e["categories"].add(cat) e["permis"].add((rec.get("NoPermis") or "").strip()) try: e["capacite"] += int(rec.get("Capacite") or 0) except ValueError: pass return list(by_no.values()) def match(con, establishments: list[dict]) -> dict: """Croisement CONSERVATEUR permis <-> restaurants (mêmes règles que inspections.match) : Règle A « nom+lieu » : nom commercial normalisé IDENTIQUE ET (code postal identique OU même ville + même numéro civique). Règle B « adresse+nom » : code postal identique ET numéro civique identique ET similarité de nom (hors tokens d'adresse/ville). """ restos = con.execute( "SELECT uid, name, chain, city, postal_code, address FROM restaurants" " WHERE active=1 AND dup_of IS NULL").fetchall() by_name: dict[str, list] = {} by_postal: dict[str, list] = {} for r in restos: info = { "uid": r["uid"], "nname": norm_name(r["name"]), "nchain": norm_name(r["chain"] or ""), "ncity": strip_accents((r["city"] or "").lower()).strip(), "postal": (r["postal_code"] or "").replace(" ", "").upper(), } mm = _CIVIC_RE.match(r["address"] or "") info["civic"] = mm.group(1) if mm else "" if info["nname"]: by_name.setdefault(info["nname"], []).append(info) core = core_name(info["nname"]) if core and core != info["nname"]: by_name.setdefault(core, []).append(info) if info["postal"]: by_postal.setdefault(info["postal"], []).append(info) today = date.today().isoformat() matched = 0 seen_uids: set[str] = set() for est in establishments: names = [] if est["nom"]: names.append(norm_name(est["nom"])) tit = est["titulaire"] if tit and not _NUMBERED_CO_RE.match(strip_accents(tit.lower())): names.append(norm_name(tit)) names = [n for n in names if n] names += [c for c in (core_name(n) for n in names) if c and c not in names] if not names: continue ncity = strip_accents(est["ville"].lower()).strip() civic_m = _CIVIC_RE.match(est["adresse"]) civic = civic_m.group(1) if civic_m else "" postal = est["postal"] hit, how = None, "" # Règle A : nom exact + code postal ou ville+civique for n in names: for info in by_name.get(n, []): same_place = ((postal and info["postal"] == postal) or (ncity and info["ncity"] and (ncity == info["ncity"] or ncity in info["ncity"] or info["ncity"] in ncity) and civic and info["civic"] == civic)) if same_place: hit, how = info["uid"], "nom+lieu" break if hit: break # Règle B : code postal + civique + similarité de nom if hit is None and postal and civic: addr_tokens = set(re.sub(r"[^a-z0-9]+", " ", strip_accents( est["adresse"].lower())).split()) | {ncity} for info in by_postal.get(postal, []): if info["civic"] != civic: continue if any(_name_similar(n, info["nname"], addr_tokens) or (info["nchain"] and _name_similar(n, info["nchain"], addr_tokens)) for n in names): hit, how = info["uid"], "adresse+nom" break if hit is None or hit in seen_uids: continue # jamais deux établissements sur 1 fiche seen_uids.add(hit) from . import db as _db _db.merge_details(con, hit, {"permis_alcool": { "categories": sorted(est["categories"]), "nb_permis": len(est["permis"]), "capacite": est["capacite"] or None, "titulaire": est["titulaire"], "matched_by": how, "source": "RACJ / Données Québec (CC-BY 4.0)", "maj": today, }}) matched += 1 con.commit() total = con.execute( "SELECT COUNT(*) c FROM restaurants WHERE active=1 AND dup_of IS NULL" " AND details LIKE '%\"permis_alcool\"%'").fetchone()["c"] return {"matched": matched, "with_permit": total} def sync(con=None, force: bool = False) -> dict | None: """Télécharge le registre RACJ, regroupe et croise. Cadence hebdo.""" from . import db own = con is None if own: con = db.connect() try: last = con.execute( "SELECT MAX(ts) ts FROM sync_log WHERE source=? AND ok=1", (SOURCE_ID,)).fetchone()["ts"] if not force and last and time.time() - last < REFRESH_DAYS * 86400: return None # déjà à jour cette semaine records = _download() establishments = group_establishments(records) m = match(con, establishments) stats = {"rows": len(records), "etablissements": len(establishments), **m} con.execute( "INSERT INTO sync_log (source, ts, found, added, updated, removed," " ok, message, stats) VALUES (?,?,?,?,0,0,1,?,?)", (SOURCE_ID, time.time(), len(establishments), m["matched"], f"permis d'alcool croisés : {m['with_permit']} resto(s)", json.dumps(stats, ensure_ascii=False))) con.commit() print(f"[resto-ka] racj: {stats}") return stats except Exception as exc: db.log_failure(con, SOURCE_ID, str(exc)) print(f"[resto-ka] racj: erreur non bloquante: {exc}", file=sys.stderr) return {"error": str(exc)} finally: if own: con.close()