#!/usr/bin/env python3
"""Vague 5 (expansion sectorielle) : nouvelles sources de fabricants.

Sources moissonnées (fiche d'évaluation datée du 2026-08-21 dans
docs/CONFORMITE.md ; découvertes via Serper/Tavily puis sondées AVANT
connexion — robots.txt lus, répertoires publics de promotion des membres) :

  modemtl             Répertoire de la mode locale montréalaise
                      (modemtl.com/repertoire-mode-locale-mtl/). Page
                      WordPress statique, ~52 liens directs vers les sites
                      des marques. robots.txt permissif (WP standard).
  boeufdici           Répertoire Bœuf d'ici — producteurs de bœuf vendant
                      à la ferme (boeufdici.com/repertoire/). Page WP
                      statique, ~43 liens directs. robots.txt permissif.
  afmq                Association des fabricants de meubles du Québec
                      (afmq.com/liste_compagnies). Liste -> fiches
                      /liste_compagnies/profil/<id>-<slug> (~100 profils),
                      site web extrait de chaque fiche (patron CTAQ de la
                      vague 4). robots.txt : /catalog /scripts /images
                      interdits — jamais requêtés.
  apiculteursduquebec Les Apiculteurs et Apicultrices du Québec — « route
                      du miel » (apiculteursduquebec.com). Répertoire
                      Drupal paginé par région (/membres/region/<15..31>) :
                      chaque carte membre expose nom, adresse (code postal)
                      et site web (lien du logo). robots.txt Drupal standard.
  vendors_epipresto   Champ `vendor` de la place de marché epipresto.ca
                      (connectée en vague 4, 273 vendeurs distincts) —
                      patron vendors_collectifs de la vague 4 : résolution
                      prudente du site officiel par candidats de domaine,
                      validés par correspondance du titre. ⚠️ vendeurs
                      hétérogènes (épiceries revendeuses ET producteurs) ->
                      prior E 0.4 « nature à vérifier », preuve QC exigée.

Écartées (raison consignée dans docs/CONFORMITE.md, 2026-08-21) :
ebenistes-quebec.com (coquille lead-gen, 0 fiche extractible),
microentreprendre.ca (répertoire JS sans liens vers les sites ; microcrédit
n'est pas une preuve de fabrication), index-design.ca (répertoire non
QC-only — fournisseurs internationaux), chantier.qc.ca (répertoire
admin-ajax ; entreprises d'économie sociale majoritairement non
fabricantes), apdiq.com / designmontreal.com (designers de services).

Usage :
  python3 scripts/wave5_discovery.py harvest [source ...]
  python3 scripts/wave5_discovery.py integrate   # additif, sérialisé
"""
import argparse
import concurrent.futures as cf
import html as htmllib
import json
import os
import re
import subprocess
import sys
import time
import unicodedata
from collections import Counter
from urllib.parse import urlparse

import requests

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RAW = os.path.join(ROOT, "data", "raw")
ENR = os.path.join(ROOT, "data", "enriched")
sys.path.insert(0, ROOT)
sys.path.insert(0, os.path.join(ROOT, "scripts"))

for line in open(os.path.join(ROOT, ".env")).read().splitlines():
    if "=" in line and not line.startswith("#"):
        k, _, v = line.partition("=")
        os.environ.setdefault(k.strip(), v.strip())

from verify import HDRS, POSTAL_RE, AREA_RE, verify_domain  # noqa: E402
from aggregate import BLOCK, norm_domain                    # noqa: E402

TAG_RE = re.compile(r"<[^>]+>")
SESS = requests.Session()

# domaines utilitaires vus sur les pages répertoires de la vague 5 (agences
# web des annuaires, CDN, boilerplate WordPress) — pas couverts par BLOCK
EXTRA_BLOCK = re.compile(
    r"(?i)(browsehappy\.com|cookiedatabase\.org|gmpg\.org|jarold\.ca|"
    r"viglob\.ca|igminformatique|themeforest|jsdelivr|cloudflare|"
    r"googleapis|gstatic|w3\.org|schema\.org|wp\.org|elementor|"
    # ritkey = agence des fiches AFMQ ; mtl.org = site touristique ;
    # achetezalaferme = place de marché (pas un producteur)
    r"ritkey\.com|mtl\.org|achetezalaferme)")

SOCIALS = ("facebook", "instagram", "linkedin", "youtube", "tiktok",
           "pinterest", "twitter", "x.com")


def strip_tags(s):
    return re.sub(r"\s+", " ", TAG_RE.sub(" ", htmllib.unescape(s or ""))).strip()


def get(url, timeout=25):
    try:
        r = SESS.get(url, headers=HDRS, timeout=timeout, allow_redirects=True)
        if r.status_code == 200:
            return r.text
    except Exception:
        pass
    return ""


def write_jsonl(name, records):
    os.makedirs(RAW, exist_ok=True)
    path = os.path.join(RAW, name + ".jsonl")
    with open(path, "w") as f:
        for r in records:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print(f"[harvest] {name}: {len(records)} entrées -> {path}", flush=True)


def _bad(url, dom):
    return (not dom or BLOCK.search(dom) or BLOCK.search(url)
            or EXTRA_BLOCK.search(url) or any(s in dom for s in SOCIALS))


# ------------------------------------------------ répertoires à liens directs
def harvest_single_page(source, page_url, own_domain, region_hint, evidence):
    html = get(page_url, timeout=40)
    records, seen = [], set()
    for url, label in re.findall(r'<a\s[^>]*href="(https?://[^"]+)"[^>]*>(.*?)</a>',
                                 html, re.S | re.I):
        url = htmllib.unescape(url)
        host = (urlparse(url).netloc or "").lower()
        if not host or own_domain in host:
            continue
        dom = norm_domain(url)
        if _bad(url, dom) or dom in seen:
            continue
        seen.add(dom)
        label = strip_tags(label)
        generic = label.lower() in {"web", "site", "site web", "voir le site",
                                    "en savoir plus", "visiter", "boutique"}
        records.append({
            "name": label if 2 < len(label) < 80 and not generic else "",
            "domain": dom, "url": url, "region_hint": region_hint,
            "category_hint": "", "evidence": evidence, "query": page_url,
        })
    write_jsonl(source, records)


# ----------------------------------------------------------------------- AFMQ
AFMQ_BASE = "https://www.afmq.com"
AFMQ_PROFIL_RE = re.compile(r'href="(/liste_compagnies/profil/[^"#?]+)"')


def harvest_afmq():
    """Liste des membres -> fiches profil -> site web (patron CTAQ vague 4)."""
    html = get(AFMQ_BASE + "/liste_compagnies", timeout=40)
    paths = sorted(set(AFMQ_PROFIL_RE.findall(html)))
    print(f"[afmq] {len(paths)} fiches profil à visiter", flush=True)
    records = []
    for i, path in enumerate(paths, 1):
        page = get(AFMQ_BASE + path, timeout=30)
        time.sleep(0.4)
        if i % 25 == 0:
            print(f"  [afmq] {i}/{len(paths)}", flush=True)
        if not page:
            continue
        name = ""
        m = re.search(r"<h1[^>]*>(.*?)</h1>", page, re.S | re.I)
        if m:
            name = strip_tags(m.group(1))
        if not name:
            name = path.rsplit("-", 1)[-1].replace("_", " ").title()
        webs = []
        for u in re.findall(r'href="(https?://[^"]+)"', page):
            dom = norm_domain(u)
            if _bad(u, dom) or "afmq" in (dom or ""):
                continue
            if u not in webs:
                webs.append(u)
        if not webs:
            continue
        text = strip_tags(page[:150000])
        postal = POSTAL_RE.search(text)
        phone = AREA_RE.search(text)
        cat = re.search(r"Catégorie\(s\) de produits:\s*(.{0,120})", text)
        records.append({
            "url": AFMQ_BASE + path,
            "title": name,
            "h1": name,
            "websites": webs[:2],
            "socials": [],
            "postal_prefix": postal.group(0)[:3] if postal else None,
            "phone": phone.group(0) if phone else None,
            "regions_mentioned": [],
            "text_sample": ("Fabricant de meubles membre AFMQ — "
                            + (cat.group(1) if cat else ""))[:200],
            "category_hint": "maison",
        })
    write_jsonl("afmq", records)


# ------------------------------------------------------- Apiculteurs du Québec
API_BASE = "https://www.apiculteursduquebec.com"
# ids Drupal des vues régionales (constatés sur le sélecteur de la page
# « entreprises apicoles d'ici ») -> nom canonique du registre
API_REGIONS = {
    15: "Abitibi-Témiscamingue", 16: "Bas-Saint-Laurent",
    17: "Capitale-Nationale", 18: "Centre-du-Québec",
    19: "Chaudière-Appalaches", 20: "Côte-Nord", 21: "Estrie",
    22: "Gaspésie–Îles-de-la-Madeleine", 23: "Lanaudière",
    24: "Laurentides", 25: "Laval", 26: "Mauricie", 27: "Montérégie",
    28: "Montréal", 29: "Nord-du-Québec", 30: "Outaouais",
    31: "Saguenay–Lac-Saint-Jean",
}
API_ROW_RE = re.compile(r'<div class="views-row">(.*?)(?=<div class="views-row">|</section>)', re.S)
API_NAME_RE = re.compile(r'field-adresse-organization-1[^>]*>.*?<a href="[^"]*">([^<]+)</a>', re.S)
API_ADDR_RE = re.compile(r'views-field-field-adresse"[^>]*><div class="field-content">(.*?)</div>', re.S)


def harvest_apiculteurs():
    records, seen = [], set()
    for rid, region in API_REGIONS.items():
        for page in range(0, 12):
            url = f"{API_BASE}/membres/region/{rid}" + (f"?page={page}" if page else "")
            html = get(url, timeout=30)
            time.sleep(0.5)
            rows = API_ROW_RE.findall(html) if html else []
            if not rows:
                break
            found = 0
            for row in rows:
                webs = []
                for u in re.findall(r'href="(https?://[^"]+)"', row):
                    dom = norm_domain(u)
                    if _bad(u, dom) or "apiculteursduquebec" in (dom or ""):
                        continue
                    if u not in webs:
                        webs.append(u)
                if not webs:
                    continue      # membre sans site web -> inconnectable
                nm = API_NAME_RE.search(row)
                name = strip_tags(nm.group(1)) if nm else ""
                key = norm_domain(webs[0])
                if key in seen:
                    continue
                seen.add(key)
                found += 1
                addr = API_ADDR_RE.search(row)
                text = strip_tags(addr.group(1)) if addr else ""
                postal = POSTAL_RE.search(text)
                records.append({
                    "url": url,
                    "title": name,
                    "h1": name,
                    "websites": webs[:2],
                    "socials": [],
                    "postal_prefix": postal.group(0)[:3].replace(" ", "") if postal else None,
                    "phone": None,
                    "regions_mentioned": [region],
                    "text_sample": text[:200],
                    "category_hint": "erable",
                })
            if not found and page > 0:
                break
        print(f"  [apiculteurs] {region}: cumul {len(records)}", flush=True)
    write_jsonl("apiculteursduquebec", records)


# --------------------------------------------------- vendors d'EPIPRESTO
COLLECTIFS = ["epipresto.ca"]
STOP_TOKENS = {"inc", "enr", "ltee", "les", "the", "and", "et", "de", "du",
               "la", "le", "des", "by", "par", "studio", "atelier", "co"}
# vendeurs génériques de la place de marché (pas des marques)
VENDOR_SKIP = re.compile(r"(?i)(général|autre choix|mini march|pharmacie|"
                         r"dépanneur|depanneur|épicerie|epicerie|march[ée] )")


def _slug(name):
    s = unicodedata.normalize("NFKD", name).encode("ascii", "ignore").decode()
    return re.sub(r"[^a-z0-9]", "", s.lower())


def _norm_name(name):
    s = unicodedata.normalize("NFKD", name).encode("ascii", "ignore").decode().lower()
    return " ".join(t for t in re.findall(r"[a-z0-9]+", s) if t not in STOP_TOKENS)


def _curl(url, timeout=20):
    """GET via curl (mécanisme du projet pour éviter le 429 TLS de requests)."""
    try:
        r = subprocess.run(["curl", "-sL", "-A", HDRS["User-Agent"], "--max-time",
                            str(timeout), "--max-filesize", "400000", url],
                           capture_output=True, timeout=timeout + 10)
        return r.stdout.decode("utf-8", "replace")
    except Exception:
        return ""


def harvest_vendors_epipresto(cap=400):
    from fabrika import db as fdb
    con = fdb.connect()
    rows = con.execute(
        "SELECT vendor, COUNT(*) AS n, GROUP_CONCAT(DISTINCT store_id) AS sids "
        "FROM products WHERE active=1 AND vendor<>'' AND store_id IN (%s) "
        "GROUP BY vendor ORDER BY n DESC" % ",".join("?" * len(COLLECTIFS)),
        COLLECTIFS).fetchall()
    known_names = {_norm_name(r[0]) for r in
                   con.execute("SELECT name FROM stores") if r[0]}
    known_doms = {r[0] for r in con.execute("SELECT id FROM stores")}
    con.close()
    cand_path = os.path.join(ENR, "candidates.jsonl")
    if os.path.exists(cand_path):
        with open(cand_path) as f:
            known_doms |= {json.loads(l)["domain"] for l in f}

    records, probes = [], 0
    for vendor, n, sids in rows:
        v = vendor.strip()
        if len(v) < 3 or len(v) > 60 or _norm_name(v) in known_names:
            continue
        if VENDOR_SKIP.search(v):
            continue
        slug = _slug(v)
        if not (4 <= len(slug) <= 30):
            continue
        if probes >= cap:
            break
        found = None
        for dom in (slug + ".com", slug + ".ca"):
            if dom in known_doms or BLOCK.search(dom):
                continue
            probes += 1
            html = _curl(f"https://{dom}")
            time.sleep(0.3)
            if not html:
                continue
            title_m = re.search(r"<title[^>]*>(.*?)</title>", html, re.S | re.I)
            title = _norm_name(strip_tags(title_m.group(1))[:120]) if title_m else ""
            vt = set(_norm_name(v).split())
            if vt and title and len(vt & set(title.split())) / len(vt) >= 0.6:
                found = dom
                break
        if found:
            records.append({
                "name": v,
                "domain": found,
                "url": f"https://{found}",
                "region_hint": "",
                "category_hint": "",
                "evidence": f"Vendeur de la place de marché EPIPRESTO ({n} produits) "
                            f"— site officiel résolu et validé par titre",
                "query": f"vendor:{v}",
            })
    print(f"[vendors_epipresto] {probes} sondes, {len(records)} sites résolus", flush=True)
    write_jsonl("vendors_epipresto", records)


HARVESTERS = {
    "modemtl": lambda: harvest_single_page(
        "modemtl", "https://modemtl.com/repertoire-mode-locale-mtl/",
        "modemtl.com", "Montréal",
        "Marque de mode locale répertoriée par ModeMTL (répertoire de la "
        "mode montréalaise)"),
    "boeufdici": lambda: harvest_single_page(
        "boeufdici", "https://boeufdici.com/repertoire/",
        "boeufdici.com", "",
        "Producteur membre du répertoire Bœuf d'ici (vente de bœuf à la ferme)"),
    "afmq": harvest_afmq,
    "apiculteursduquebec": harvest_apiculteurs,
    "vendors_epipresto": harvest_vendors_epipresto,
}

WAVE5_SOURCES = list(HARVESTERS)


def harvest(only=None):
    os.makedirs(RAW, exist_ok=True)
    for name, fn in HARVESTERS.items():
        if only and name not in only:
            continue
        print(f"[harvest] === {name} ===", flush=True)
        fn()


def integrate():
    """Intégration additive au registre + DB (mêmes garde-fous que la vague 4)."""
    from datetime import date
    import build_registry as br
    from fabrika import db as fdb

    for s in WAVE5_SOURCES:
        assert s in br.SOURCE_PRIORS, f"prior manquant dans build_registry: {s}"

    subprocess.run([sys.executable, os.path.join(ROOT, "scripts", "aggregate.py")],
                   check=True)

    cands = {}
    with open(os.path.join(ENR, "candidates.jsonl")) as f:
        for line in f:
            c = json.loads(line)
            cands[c["domain"]] = c

    verified_path = os.path.join(ENR, "verified.jsonl")
    verified = {}
    with open(verified_path) as f:
        for line in f:
            v = json.loads(line)
            verified[v["domain"]] = v

    reg_path = os.path.join(ROOT, "data", "stores.json")
    reg = json.load(open(reg_path))
    existing_ids = {s["id"] for s in reg["stores"]}

    new_domains = sorted(d for d in cands
                         if d not in verified and d not in existing_ids)
    print(f"[integrate] {len(new_domains)} nouveaux domaines à vérifier", flush=True)

    new_recs = []
    with cf.ThreadPoolExecutor(12) as ex:
        for i, rec in enumerate(ex.map(verify_domain, new_domains)):
            new_recs.append(rec)
            if (i + 1) % 100 == 0:
                print(f"  verify {i+1}/{len(new_domains)}", flush=True)
    with open(verified_path, "a") as f:
        for r in new_recs:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
            verified[r["domain"]] = r

    added, skipped_dup, skipped_qc, skipped_dead = [], 0, 0, 0
    for dom in new_domains:
        cand, ver = cands[dom], verified.get(dom, {})
        if not ver.get("active"):
            skipped_dead += 1
            continue
        final_dom = ver.get("final_domain") or dom
        if final_dom in existing_ids:
            skipped_dup += 1
            continue
        qc_signal = any(ver.get(k) for k in ("qc_postal", "qc_phone", "tld_quebec",
                                             "mentions_quebec", "made_in_qc_wording"))
        qc_source = any(s in br.QC_ONLY_SOURCES for s in cand.get("sources", []))
        if not qc_signal and not qc_source:
            skipped_qc += 1
            continue
        cls, conf, ev = br.classify(cand, ver)
        default_cat = next((br.SOURCE_PRIORS[s][3] for s in br.PRIORITY
                            if s in cand.get("sources", []) and br.SOURCE_PRIORS[s][3]), None)
        platform = ver.get("platform") or ""
        catalog_endpoint = ver.get("catalog_endpoint") or ""
        if platform == "wix" and not catalog_endpoint:
            catalog_endpoint = "/_api/wix-ecommerce-storefront-web/api"
        fu = urlparse(ver.get("final_url") or f"https://{final_dom}")
        store = {
            "id": final_dom,
            "name": br.clean_name(cand, ver),
            "url": f"{fu.scheme}://{fu.netloc}",
            "platform": platform,
            "catalog_endpoint": catalog_endpoint,
            "city": "",
            "region": br.pick_region(cand) or br.region_from_postal(cand, ver),
            "postal_prefix": cand.get("postal_prefix") or (ver.get("qc_postal") or "")[:3] or None,
            "phone": cand.get("phone") or ver.get("qc_phone"),
            "origin_class": cls,
            "origin_confidence": conf,
            "origin_evidence": ev,
            "categories": [default_cat] if default_cat else [],
            "socials": (cand.get("socials") or [])[:4] or ver.get("socials", []),
            "discovery_sources": cand.get("sources", []),
            "discovery_source_urls": cand.get("source_pages", [])[:5],
            "language": ver.get("language"),
            "ecommerce": bool(ver.get("has_cart") or catalog_endpoint),
            "verification_date": ver.get("checked_at") or str(date.today()),
            "status": "verified" if (conf >= 0.6 and ver.get("mentions_quebec")) else "probable",
            "enabled": bool(catalog_endpoint),
        }
        existing_ids.add(final_dom)
        added.append(store)

    reg["stores"].extend(added)
    reg["count"] = len(reg["stores"])
    reg["generated"] = str(date.today())
    json.dump(reg, open(reg_path, "w"), ensure_ascii=False, indent=1)

    con = fdb.connect()
    for s in added:
        fdb.upsert_store(con, s)
    con.commit(); con.close()

    enabled = [s["id"] for s in added if s["enabled"]]
    per_plat = Counter(s["platform"] or "(aucune)" for s in added)
    per_src = Counter(src for s in added for src in s["discovery_sources"])
    per_cls = Counter(s["origin_class"] for s in added)
    print(f"[integrate] boutiques ajoutées: {len(added)} | connectables (enabled): {len(enabled)}")
    print(f"[integrate] écartées — mortes/injoignables: {skipped_dead}, "
          f"dédup domaine final: {skipped_dup}, sans preuve QC: {skipped_qc}")
    print("[integrate] par plateforme:", dict(per_plat.most_common()))
    print("[integrate] par source:", dict(per_src.most_common()))
    print("[integrate] par classe:", dict(per_cls))
    with open(os.path.join(ROOT, "data", "wave5_new_enabled.txt"), "w") as f:
        f.write("\n".join(enabled) + "\n")
    if enabled:
        print("[integrate] à synchroniser: python run.py sync $(cat data/wave5_new_enabled.txt)")


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("cmd", choices=["harvest", "integrate"])
    ap.add_argument("only", nargs="*", help="sources précises (harvest)")
    args = ap.parse_args()
    harvest(args.only or None) if args.cmd == "harvest" else integrate()