# ==============================================================================
# Author: Simon-Pierre Boucher <contact@spboucher.ai>
# File:   restoka/connectors/siteresto.py
# Desc:   Connecteur « sites web des restos » (Palier 3) — capte les MENUS
#         MAISON des restaurants découverts par OSM qui publient un site web.
#
#         Pipeline par site (budget par passage, incrémental via detail_cache) :
#           1. GET la page d'accueil (requests direct, UA RestoKaBot).
#           2. DÉTECTION DE PLATEFORME (CLAUDE.md §9 P3) : si le site embarque
#              UEAT (order.ueat.io/integration/<GUID> ou *.order-online.ai),
#              le GUID est consigné dans data/ueat-discovered.json — le
#              connecteur UEAT (Palier 1, menu structuré parfait) le couvrira
#              au prochain passage. Aucune fiche émise ici.
#           3. Sinon : trouver le lien « menu/carte », récupérer la page
#              (Firecrawl en secours pour les sites JS, budget limité) ou le
#              PDF du menu.
#           4. Structuration par Claude Haiku 4.5 (restoka/menullm.py) —
#              sorties structurées JSON + validation STRICTE des prix.
#           5. Émettre une fiche `site-resto` clonée de la fiche OSM avec le
#              menu ; la déduplication masque la fiche OSM derrière elle.
#
#         Contexte de prix produit : DINE-IN (menu affiché par le resto
#         lui-même — la référence idéale, §6.2). price_source: site-resto.
#         Les fiches déjà captées sont RÉÉMISES depuis la base à chaque
#         passage (délai de grâce) et re-crawlées après REFRESH_DAYS.
# ==============================================================================
from __future__ import annotations

import datetime
import json
import os
import re
import sys
import urllib.parse
from pathlib import Path

from ..hours import parse_jsonld_hours
from ..normalize import normalize_phone
from ..schema import Restaurant
from .base import BaseConnector

DATA_DIR = Path(__file__).resolve().parent.parent.parent / "data"
UEAT_DISCOVERED = DATA_DIR / "ueat-discovered.json"

# Budgets par passage (incrémental : la couverture grandit à chaque sync)
MAX_SITES_PER_RUN = int(os.environ.get("SITE_RESTO_BUDGET", "250"))
MAX_FIRECRAWL_PER_RUN = int(os.environ.get("SITE_RESTO_FIRECRAWL", "40"))
WORKERS = int(os.environ.get("SITE_RESTO_WORKERS", "6"))   # crawls parallèles
REFRESH_DAYS = 30                 # re-crawl des menus captés (cadence §14)

# verrou d'écriture du fichier des intégrations UEAT découvertes (workers //)
_UEAT_FILE_LOCK = __import__("threading").Lock()

# plateformes de réservation reconnues (lien « réserver » sur le site du resto)
_RESERVATION_DOMAINS = (
    "opentable.com", "opentable.ca", "libroreserve.com", "restoloco.com",
    "restoloco.ca", "resy.com", "sevenrooms.com", "bookenda.com",
    "zenchef.com", "guestonline.io", "tablein.com",
)

_UEAT_RE = re.compile(
    r"(?:order\.ueat\.io/integration/|order-online\.ai.{0,200}?integration/)"
    r"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})", re.I)
_ORDER_ONLINE_RE = re.compile(r"https?://([a-z0-9-]+)\.order-online\.ai", re.I)

# autres plateformes de commande (CLAUDE.md §9 P1) détectables sur le site du
# resto — consignées dans data/<plateforme>-discovered.json, couvertes par
# leurs connecteurs dédiés (gloriafood.py, chownow.py, squareonline.py ;
# Clover : consigné comme candidat, pas de connecteur — SPA opaque)
_GLF_RUID_RE = re.compile(r'data-glf-ruid="([0-9a-f-]{30,40})"', re.I)
_GLF_CUID_RE = re.compile(r'data-glf-cuid="([0-9a-f-]{30,40})"', re.I)
_CHOWNOW_RE = re.compile(
    r"(?:direct\.chownow\.com/order/|ordering\.chownow\.com/order/"
    r"|api\.chownow\.com/api/restaurant/|chownow\.com/order/)(\d+)", re.I)
_SQUARE_SITE_RE = re.compile(r"https?://([a-z0-9-]+\.square\.site)", re.I)
_CLOVER_RE = re.compile(r"clover\.com/online-ordering/([a-z0-9-]+)", re.I)
# plateformes dont le connecteur dédié capte un MENU structuré : quand l'une
# d'elles est détectée, site-resto ne crawle pas le menu maison (comme UEAT)
_MENU_PLATFORMS = ("gloriafood", "chownow", "square")
_MENU_WORD_RE = re.compile(r"menu|carte|nos-plats|nos_plats|food", re.I)
_PRICE_HINT_RE = re.compile(r"\d{1,3}[.,]\d{2}\s*\$|\$\s*\d{1,3}[.,]\d{2}|\d{1,3}\s*\$")


def _domain(url: str) -> str:
    try:
        return urllib.parse.urlparse(url).netloc.lower().removeprefix("www.")
    except ValueError:
        return ""


class SiteRestoConnector(BaseConnector):
    source_id = "site-resto"
    request_delay = 1.0
    timeout = 20
    use_detail_cache = False      # cache géré à la main (statuts par domaine)

    # -- accès base -------------------------------------------------------------
    def _con(self):
        from .. import db
        if self._detail_con is None:
            self._detail_con = db.connect()
        return self._detail_con

    def _get_status(self, domain: str) -> tuple[dict | None, float]:
        row = self._con().execute(
            "SELECT payload, fetched_at FROM detail_cache"
            " WHERE source=? AND external_id=?",
            (self.source_id, domain)).fetchone()
        if row is None:
            return None, 0.0
        try:
            return json.loads(row["payload"] or "{}"), row["fetched_at"] or 0.0
        except ValueError:
            return None, 0.0

    def _set_status(self, domain: str, payload: dict) -> None:
        from .. import db
        db.put_cached_detail(self._con(), self.source_id, domain, "v1", payload)

    # -- fetch HTML ---------------------------------------------------------------
    def _get_html(self, url: str) -> str:
        try:
            resp = self.get(url, allow_redirects=True)
        except Exception:
            return ""
        ctype = resp.headers.get("content-type", "")
        if "pdf" in ctype:
            return ""
        try:
            return resp.text[:800_000]
        except Exception:
            return ""

    def _get_pdf(self, url: str) -> bytes:
        try:
            resp = self.get(url, allow_redirects=True)
            if len(resp.content) <= 8_000_000:
                return resp.content
        except Exception:
            pass
        return b""

    @staticmethod
    def _html_text(html: str) -> str:
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        for tag in soup(["script", "style", "nav", "footer", "noscript",
                         "header", "svg", "form"]):
            tag.decompose()
        return re.sub(r"\n{3,}", "\n\n", soup.get_text("\n", strip=True))

    def _menu_links(self, html: str, base_url: str) -> tuple[list[str], list[str]]:
        """(pages menu HTML, PDF de menu) trouvés sur la page d'accueil."""
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        pages, pdfs = [], []
        base_dom = _domain(base_url)
        for a in soup.find_all("a", href=True):
            href = a["href"].strip()
            label = f"{href} {a.get_text(' ', strip=True)}"
            if not _MENU_WORD_RE.search(label):
                continue
            full = urllib.parse.urljoin(base_url, href)
            if full.lower().split("?")[0].endswith(".pdf"):
                if full not in pdfs:
                    pdfs.append(full)
            elif _domain(full) == base_dom and full not in pages:
                pages.append(full)
        return pages[:2], pdfs[:2]

    # -- extraction des photos du resto (page d'accueil) ------------------------------
    @staticmethod
    def _extract_images(html: str, base_url: str) -> list[str]:
        """Jusqu'à 6 photos du resto extraites de sa page d'accueil.

        og:image/twitter:image d'abord (l'image que le resto met lui-même de
        l'avant), puis les <img> de la page (src/srcset/lazy-load). Filtres :
        logos, icônes, pictos de paiement, placeholders et miniatures
        (suffixe WordPress -LxH < 300 px) sont écartés ; les variantes de
        taille d'une même image sont dédupliquées sur le nom de base.
        """
        from bs4 import BeautifulSoup
        if not html:
            return []
        _BAD_RE = re.compile(
            r"logo|icon|favicon|placeholder|sprite|badge|paiement|payment|"
            r"visa|mastercard|interac|avatar|emoji|pixel|spinner|loader|"
            r"footer|bandeau|widget|captcha", re.I)
        _EXT = r"(?:jpe?g|png|webp|avif)"
        _EXT_RE = re.compile(rf"\.{_EXT}(?:\?|$)", re.I)
        _THUMB_RE = re.compile(rf"-(\d{{2,4}})x(\d{{2,4}})\.{_EXT}", re.I)

        def _ok(u: str) -> bool:
            if not u.startswith("http") or not _EXT_RE.search(u):
                return False
            if _BAD_RE.search(u):
                return False
            m = _THUMB_RE.search(u)
            if m and max(int(m.group(1)), int(m.group(2))) < 300:
                return False
            return True

        def _base_key(u: str) -> str:
            # même photo en plusieurs tailles : clé = URL sans suffixe de
            # variante (-LxH WordPress, -p-500 Webflow) ni extension
            u = u.split("?")[0].lower()
            u = re.sub(rf"-\d{{2,4}}x\d{{2,4}}(?=\.{_EXT}$)", "", u)
            u = re.sub(rf"-p-\d{{2,4}}(?=\.{_EXT}$)", "", u)
            return re.sub(rf"\.{_EXT}$", "", u)

        soup = BeautifulSoup(html, "html.parser")
        found: list[str] = []
        for name, attrs in (("meta", {"property": "og:image"}),
                            ("meta", {"name": "twitter:image"})):
            for tag in soup.find_all(name, attrs=attrs):
                u = urllib.parse.urljoin(base_url, (tag.get("content") or "").strip())
                if _ok(u):
                    found.append(u)
        for img in soup.find_all("img"):
            try:  # pictos et flèches déclarés étroits dans le markup
                if img.get("width") and int(str(img["width"]).rstrip("px")) < 200:
                    continue
            except (ValueError, TypeError):
                pass
            cands = [img.get("src") or "", img.get("data-src") or "",
                     img.get("data-lazy-src") or ""]
            srcset = img.get("srcset") or img.get("data-srcset") or ""
            if srcset:  # prendre la plus grande variante du srcset
                parts = [p.strip().split(" ")[0] for p in srcset.split(",")]
                if parts:
                    cands.append(parts[-1])
            for c in cands:
                u = urllib.parse.urljoin(base_url, c.strip())
                if _ok(u):
                    found.append(u)
                    break
        out: list[str] = []
        seen: set[str] = set()
        for u in found:
            k = _base_key(u)
            if k in seen:
                continue
            seen.add(k)
            out.append(u)
            if len(out) >= 6:
                break
        return out

    # -- extraction contact/horaires/réservation (page d'accueil) ---------------------
    @staticmethod
    def _extract_contact(html: str, base_url: str) -> dict:
        """{"phone", "hours", "reservation_url"} extraits du HTML d'accueil.

        - téléphone : lien tel: d'abord (fiable), sinon regex nord-américaine
          sur le texte visible -> E.164 (+1XXXXXXXXXX) ;
        - réservation : premier lien vers une plateforme connue (OpenTable,
          Libro, Resto Loco…) ;
        - horaires : JSON-LD schema.org (LocalBusiness/Restaurant,
          openingHoursSpecification ou openingHours) -> structuré par jour.
        """
        from bs4 import BeautifulSoup
        out = {"phone": "", "hours": None, "reservation_url": ""}
        if not html:
            return out
        soup = BeautifulSoup(html, "html.parser")
        tel_numbers: list[str] = []
        for a in soup.find_all("a", href=True):
            href = a["href"].strip()
            if href.lower().startswith("tel:"):
                p = normalize_phone(href[4:])
                if p and p not in tel_numbers:
                    tel_numbers.append(p)
            if not out["reservation_url"] and href.startswith("http"):
                dom = _domain(href)
                if any(dom == d or dom.endswith("." + d)
                       for d in _RESERVATION_DOMAINS):
                    out["reservation_url"] = href
        # >3 numéros distincts = page « trouver une succursale » d'une chaîne :
        # impossible d'attribuer LE bon numéro à CE resto — on s'abstient
        if 1 <= len(tel_numbers) <= 3:
            out["phone"] = tel_numbers[0]
        # horaires structurés via JSON-LD (LocalBusiness / Restaurant)
        for script in soup.find_all("script", type="application/ld+json"):
            try:
                data = json.loads(script.string or "")
            except (ValueError, TypeError):
                continue
            nodes = data if isinstance(data, list) else [data]
            more = []
            for n in nodes:
                if isinstance(n, dict) and isinstance(n.get("@graph"), list):
                    more.extend(n["@graph"])
            for node in nodes + more:
                if not isinstance(node, dict):
                    continue
                if not out["hours"]:
                    out["hours"] = parse_jsonld_hours(node)
                if not out["phone"]:
                    out["phone"] = normalize_phone(node.get("telephone"))
            if out["hours"] and out["phone"]:
                break
        if not out["phone"]:      # secours : regex sur le texte visible
            text = SiteRestoConnector._html_text(html)[:20_000]
            m = re.search(r"(?:t[eé]l|phone|appelez|r[eé]servation)[^\n]{0,40}?"
                          r"(\(?\d{3}\)?[\s.\-]\d{3}[\s.\-]\d{4})", text, re.I)
            out["phone"] = normalize_phone(m.group(1) if m else "")
        return out

    # -- traitement d'un resto -------------------------------------------------------
    def _discover_ueat(self, html: str, final_url: str) -> str | None:
        """GUID d'intégration UEAT si le site embarque la plateforme."""
        m = _UEAT_RE.search(html)
        if m:
            return m.group(1)
        m = _ORDER_ONLINE_RE.search(html)
        if m:   # lien vers la page de commande marque blanche : suivre pour le GUID
            sub = self._get_html(f"https://{m.group(1)}.order-online.ai/")
            m2 = _UEAT_RE.search(sub)
            if m2:
                return m2.group(1)
        return None

    def _record_ueat(self, guid: str, name: str, domain: str) -> None:
        with _UEAT_FILE_LOCK:
            self._record_ueat_locked(guid, name, domain)

    def _discover_platforms(self, html: str) -> list[dict]:
        """Plateformes de commande tierces (hors UEAT) embarquées sur le site."""
        out: list[dict] = []
        m = _GLF_RUID_RE.search(html)
        if m:
            cuid = _GLF_CUID_RE.search(html)
            entry = {"name": "gloriafood", "key": m.group(1)}
            if cuid:
                entry["cuid"] = cuid.group(1)
            out.append(entry)
        m = _CHOWNOW_RE.search(html)
        if m:
            out.append({"name": "chownow", "key": m.group(1)})
        m = _SQUARE_SITE_RE.search(html)
        if m:
            out.append({"name": "square", "key": m.group(1).lower()})
        m = _CLOVER_RE.search(html)
        if m:
            out.append({"name": "clover", "key": m.group(1)})
        return out

    def _record_platform(self, platform: dict, name: str, domain: str,
                         uid: str) -> None:
        """Consigne une intégration découverte dans
        data/<plateforme>-discovered.json (avec l'uid du resto porteur pour
        les connecteurs d'enrichissement gloriafood/square)."""
        path = DATA_DIR / f"{platform['name']}-discovered.json"
        with _UEAT_FILE_LOCK:
            data = {"integrations": []}
            if path.exists():
                try:
                    data = json.loads(path.read_text(encoding="utf-8"))
                except ValueError:
                    pass
            if any(i.get("key") == platform["key"]
                   for i in data.get("integrations", [])):
                return
            entry = {"key": platform["key"], "uid": uid,
                     "label": f"{name} (découvert via {domain})"}
            if platform.get("cuid"):
                entry["cuid"] = platform["cuid"]
            data.setdefault("integrations", []).append(entry)
            path.write_text(json.dumps(data, ensure_ascii=False, indent=1),
                            encoding="utf-8")
            print(f"[resto-ka] site-resto: plateforme {platform['name']} "
                  f"découverte chez {name} ({platform['key']}) -> {path.name}")

    def _record_ueat_locked(self, guid: str, name: str, domain: str) -> None:
        data = {"integrations": []}
        if UEAT_DISCOVERED.exists():
            try:
                data = json.loads(UEAT_DISCOVERED.read_text(encoding="utf-8"))
            except ValueError:
                pass
        if any(i.get("key") == guid for i in data["integrations"]):
            return
        data["integrations"].append(
            {"key": guid, "label": f"{name} (découvert via {domain})"})
        UEAT_DISCOVERED.write_text(
            json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
        print(f"[resto-ka] site-resto: plateforme UEAT découverte chez {name}"
              f" ({guid}) -> ueat-discovered.json")

    def _crawl_menu(self, website: str, firecrawl_budget: list[int]) -> dict | None:
        """Retourne {"sections": [...], "url": <page>, "contact": {...}},
        {"ueat": guid, "contact": ...} ou {"contact": ...} (pas de menu mais
        téléphone/horaires/réservation captés) ou None."""
        from .. import menullm
        home = self._get_html(website)
        if not home:
            return None
        contact = self._extract_contact(home, website)
        images = self._extract_images(home, website)
        platforms = self._discover_platforms(home)
        guid = self._discover_ueat(home, website)
        if guid:
            return {"ueat": guid, "contact": contact, "images": images,
                    "platforms": platforms}
        if any(p["name"] in _MENU_PLATFORMS for p in platforms):
            # une plateforme à menu structuré couvre ce resto : son connecteur
            # dédié prendra le relais (comme UEAT) — pas de crawl maison
            return {"contact": contact, "images": images,
                    "platforms": platforms}
        pages, pdfs = self._menu_links(home, website)
        # 1) pages HTML de menu
        for page_url in pages:
            html = self._get_html(page_url)
            text = self._html_text(html) if html else ""
            if len(_PRICE_HINT_RE.findall(text)) < 3 and firecrawl_budget[0] > 0:
                try:   # site JS : rendu via Firecrawl (budget limité)
                    firecrawl_budget[0] -= 1
                    rendered = self.get_rendered(page_url)
                    if rendered:
                        text = self._html_text(rendered)
                except Exception:
                    pass
            if len(_PRICE_HINT_RE.findall(text)) >= 3:
                menu = menullm.menu_from_text(text)
                if menu:
                    return {"sections": menu["sections"], "url": page_url,
                            "contact": contact, "images": images,
                            "platforms": platforms}
        # 2) PDF de menu
        for pdf_url in pdfs:
            pdf = self._get_pdf(pdf_url)
            if pdf[:5] == b"%PDF-":
                menu = menullm.menu_from_pdf(pdf)
                if menu:
                    return {"sections": menu["sections"], "url": pdf_url,
                            "contact": contact, "images": images,
                            "platforms": platforms}
        # 3) la page d'accueil elle-même contient parfois le menu
        text = self._html_text(home)
        if len(_PRICE_HINT_RE.findall(text)) >= 8:
            menu = menullm.menu_from_text(text)
            if menu:
                return {"sections": menu["sections"], "url": website,
                        "contact": contact, "images": images,
                        "platforms": platforms}
        if any(v for v in contact.values()) or images or platforms:
            # pas de menu, mais du contact/photos/plateformes captés
            return {"contact": contact, "images": images,
                    "platforms": platforms}
        return None

    # -- reconstruction des fiches déjà captées ------------------------------------------
    def _row_to_restaurant(self, row, menus_by_uid: dict) -> Restaurant:
        menu = menus_by_uid.get(row["uid"])
        return Restaurant(
            source=self.source_id, external_id=row["external_id"],
            name=row["name"], url=row["url"], chain=row["chain"],
            cuisines=json.loads(row["cuisines"] or "[]"),
            establishment_type=row["establishment_type"] or "",
            price_range=row["price_range"] or "",
            address=row["address"] or "", city=row["city"] or "",
            region=row["region"] or "", postal_code=row["postal_code"] or "",
            lat=row["lat"], lng=row["lng"], phone=row["phone"] or "",
            website=row["website"] or "",
            hours=json.loads(row["hours"] or "{}"),
            services=json.loads(row["services"] or "[]"),
            dietary_options=json.loads(row["dietary_options"] or "[]"),
            languages=json.loads(row["languages"] or '["fr"]'),
            images=json.loads(row["images"] or "[]"),
            menu=menu,
        )

    def _existing(self) -> list[Restaurant]:
        con = self._con()
        menus_by_uid = {}
        for m in con.execute(
                "SELECT uid, price_context, price_source, currency, captured_at,"
                " sections FROM menus WHERE uid LIKE 'site-resto:%'"):
            menus_by_uid[m["uid"]] = {
                "currency": m["currency"] or "CAD",
                "price_context": m["price_context"],
                "price_source": m["price_source"],
                "captured_at": m["captured_at"],
                "sections": json.loads(m["sections"] or "[]"),
            }
        return [self._row_to_restaurant(r, menus_by_uid) for r in con.execute(
            "SELECT * FROM restaurants WHERE source='site-resto'")]

    # -- fetch --------------------------------------------------------------------------
    def fetch(self) -> list[Restaurant]:
        import time
        con = self._con()
        out: dict[str, Restaurant] = {r.uid: r for r in self._existing()}

        candidates = con.execute(
            "SELECT * FROM restaurants WHERE source='osm' AND active=1"
            " AND dup_of IS NULL AND website<>''"
            " AND NOT EXISTS (SELECT 1 FROM menus m WHERE m.uid=restaurants.uid)"
            " ORDER BY region<>'' DESC, city").fetchall()

        firecrawl_budget = [MAX_FIRECRAWL_PER_RUN
                            if os.environ.get("FIRECRAWL_API_KEY") else 0]
        stats = {"crawled": 0, "menus": 0, "ueat": 0, "plateformes": 0,
                 "vides": 0, "contacts": 0, "photos": 0}
        now = time.time()
        # enrichissements à appliquer APRÈS sync_source (la fiche site-resto
        # n'existe en base qu'après l'upsert) — consommé par ingest.run
        self.pending_enrichment: list[tuple[str, dict]] = []

        # 1re passe (fil principal) : réémission depuis le cache + sélection
        # des sites à crawler (budget) — aucune E/S réseau ici.
        to_crawl: list[tuple] = []
        for row in candidates:
            website = row["website"]
            if not website.startswith("http"):
                website = "https://" + website
            domain = _domain(website)
            if not domain:
                continue
            status, fetched_at = self._get_status(domain)
            if status is not None and now - fetched_at < REFRESH_DAYS * 86400:
                # menu déjà capté mais pas encore en base (run interrompu) :
                # réémettre depuis le cache — le connecteur est idempotent
                uid = f"{self.source_id}:{row['external_id']}"
                if (status.get("status") == "menu" and status.get("sections")
                        and uid not in out):
                    resto = self._row_to_restaurant(row, {})
                    resto.source = self.source_id
                    resto.url = status.get("url") or website
                    if status.get("images") and not resto.images:
                        resto.images = status["images"]
                    resto.menu = {
                        "currency": "CAD", "price_context": "dine-in",
                        "price_source": self.source_id,
                        "captured_at": status.get("captured_at") or
                        datetime.datetime.now(datetime.timezone.utc)
                        .strftime("%Y-%m-%dT%H:%M:%SZ"),
                        "sections": status["sections"],
                    }
                    resto.cuisines, resto.price_range, resto.dietary_options = [], "", []
                    out[uid] = resto
                continue                       # déjà tenté récemment
            if len(to_crawl) < MAX_SITES_PER_RUN:
                to_crawl.append((row, website, domain))

        # 2e passe : crawl + structuration Claude en PARALLÈLE (WORKERS fils).
        # Chaque fil a son propre connecteur (session requests dédiée) et ne
        # touche JAMAIS la base — les écritures restent dans le fil principal.
        from concurrent.futures import ThreadPoolExecutor

        def _crawl_one(args):
            row, website, domain = args
            worker = SiteRestoConnector()
            try:
                return row, domain, website, worker._crawl_menu(website,
                                                                firecrawl_budget)
            except Exception as exc:            # un site cassé ne bloque pas le lot
                print(f"[resto-ka] site-resto: {domain} erreur: {exc}",
                      file=sys.stderr)
                return row, domain, website, None

        with ThreadPoolExecutor(max_workers=max(1, WORKERS)) as pool:
            results = pool.map(_crawl_one, to_crawl)
            for row, domain, website, result in results:
                stats["crawled"] += 1
                # téléphone / horaires / lien de réservation captés sur le
                # site : COALESCE conservateur sur la fiche OSM d'origine
                # (fil principal seulement — les workers ne touchent pas la BD)
                contact = (result or {}).get("contact") or {}
                if any(v for v in contact.values()):
                    from .. import db as _db
                    _db.enrich_contact(
                        con, row["uid"], phone=contact.get("phone") or "",
                        hours=contact.get("hours"),
                        details={"reservation_url":
                                 contact.get("reservation_url")}
                        if contact.get("reservation_url") else None)
                    con.commit()
                    stats["contacts"] += 1
                # photos captées sur le site du resto : posées sur la fiche
                # OSM d'origine si elle n'en a pas (COALESCE conservateur)
                images = (result or {}).get("images") or []
                if images:
                    con.execute(
                        "UPDATE restaurants SET images=? WHERE uid=? AND"
                        " (images IS NULL OR images='' OR images='[]')",
                        (json.dumps(images), row["uid"]))
                    con.commit()
                    stats["photos"] += 1
                # plateformes de commande découvertes : consignées ici (fil
                # principal seulement — les workers n'écrivent ni BD ni fichiers)
                for p in (result or {}).get("platforms") or []:
                    self._record_platform(p, row["name"], domain, row["uid"])
                if result and result.get("ueat"):
                    self._record_ueat(result["ueat"], row["name"], domain)
                    self._set_status(domain, {"status": "ueat",
                                              "guid": result["ueat"]})
                    stats["ueat"] += 1
                    continue
                blocking = [p for p in (result or {}).get("platforms") or []
                            if p["name"] in _MENU_PLATFORMS]
                if blocking and not (result and result.get("sections")):
                    # menu servi par une plateforme connue : le connecteur
                    # dédié (gloriafood/chownow/square) prendra le relais
                    self._set_status(domain, {"status": "platform",
                                              "platforms":
                                              [p["name"] for p in blocking]})
                    stats["plateformes"] += 1
                    continue
                if not result or not result.get("sections"):
                    self._set_status(domain, {"status": "no_menu"})
                    stats["vides"] += 1
                    continue
                captured_at = datetime.datetime.now(datetime.timezone.utc) \
                    .strftime("%Y-%m-%dT%H:%M:%SZ")
                resto = self._row_to_restaurant(row, {})
                resto.source = self.source_id
                resto.url = result["url"]
                resto.menu = {
                    "currency": "CAD",
                    "price_context": "dine-in",  # menu maison : la référence (§6.2)
                    "price_source": self.source_id,
                    "captured_at": captured_at,
                    "sections": result["sections"],
                }
                resto.cuisines = []              # reclassées avec le texte du menu
                resto.price_range = ""
                resto.dietary_options = []
                if images and not resto.images:
                    resto.images = images
                # contact capté sur le site : complète la fiche émise (COALESCE)
                if contact.get("phone") and not (resto.phone or "").strip():
                    resto.phone = contact["phone"]
                if contact.get("hours"):
                    resto.hours = {**contact["hours"], **(resto.hours or {})}
                if contact.get("reservation_url"):
                    self.pending_enrichment.append(
                        (resto.uid, {"details": {"reservation_url":
                                                 contact["reservation_url"]}}))
                out[resto.uid] = resto
                self._set_status(domain, {"status": "menu", "url": result["url"],
                                          "captured_at": captured_at,
                                          "sections": result["sections"],
                                          "images": images})
                stats["menus"] += 1

        print(f"[resto-ka] site-resto: {stats} (workers={WORKERS})")
        return list(out.values())