#!/usr/bin/env python3 """Re-sondage agressif pour augmenter le nombre de boutiques connectables. Phase A — boutiques actives SANS endpoint catalogue : re-tester directement les 4 endpoints (Shopify products.json via curl, WooCommerce Store API, Wix access-tokens + app Stores, Squarespace ?format=json), même si la plateforme détectée est « unknown/wordpress » (thèmes headless, signatures manquées). Phase B — candidats INACTIFS (sites « morts ») : nouvelle tentative directe, puis via Scrapfly (asp anti-bot) — beaucoup de « morts » sont en fait des murs anti-bot qui bloquent python-requests. Met à jour data/verify_cache/.json ; relancer ensuite build_registry.py puis sync. """ import concurrent.futures as cf import json import os import re import subprocess import sys import time import requests ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.join(ROOT, "scripts")) sys.path.insert(0, ROOT) CACHE = os.path.join(ROOT, "data", "verify_cache") # .env 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 detect_platform, MADE_RE, POSTAL_RE, AREA_RE, QC_WORD_RE, CART_RE, SOCIAL_RE, HDRS # noqa: E402 from fabrika.connectors.scrapfly import scrapfly_get, available # noqa: E402 WIX_STORES_APP = "1380b703-ce81-ff05-f115-39571d94dfcd" def curl_get(url, timeout=20): p = subprocess.run(["curl", "-sS", "-L", "--compressed", "--max-time", str(timeout), "-A", HDRS["User-Agent"].split(" FabriKaBot")[0], "-w", "\n%{http_code}", url], capture_output=True, text=True, errors="replace") body, _, code = p.stdout.rpartition("\n") return (int(code) if code.isdigit() else 0), body def probe_endpoints(domain): """Teste tous les endpoints catalogue ; retourne (platform, endpoint) ou (None, None).""" base = f"https://{domain}" # Shopify (via curl : la moitié des blocages TLS disparaissent) code, body = curl_get(f"{base}/products.json?limit=1") if code == 200 and body.lstrip().startswith("{") and '"products"' in body[:200]: return "shopify", "/products.json" # WooCommerce Store API try: r = requests.get(f"{base}/wp-json/wc/store/v1/products?per_page=1", headers=HDRS, timeout=15) if r.status_code == 200 and r.text.strip().startswith("["): return "woocommerce", "/wp-json/wc/store/v1/products" except Exception: pass # Wix Stores try: r = requests.get(f"{base}/_api/v1/access-tokens", headers=HDRS, timeout=15, allow_redirects=True) if r.status_code == 200 and WIX_STORES_APP in (r.text or ""): return "wix", "/_api/wix-ecommerce-storefront-web/api" except Exception: pass # Squarespace for path in ("/shop", "/boutique", "/store"): try: r = requests.get(f"{base}{path}?format=json-pretty", headers=HDRS, timeout=12) if r.status_code == 200 and '"items"' in r.text[:5000]: return "squarespace", path + "?format=json" except Exception: pass return None, None def phase_a(): reg = json.load(open(os.path.join(ROOT, "data", "stores.json")))["stores"] targets = [s["id"] for s in reg if not s.get("catalog_endpoint")] print(f"[A] {len(targets)} boutiques actives sans endpoint — re-sondage direct") found = 0 def work(dom): plat, ep = probe_endpoints(dom) if not ep: return None cpath = os.path.join(CACHE, dom + ".json") try: rec = json.load(open(cpath)) except Exception: rec = {"domain": dom, "active": True, "final_domain": dom} rec["platform"] = plat rec["catalog_endpoint"] = ep json.dump(rec, open(cpath, "w")) return dom, plat with cf.ThreadPoolExecutor(12) as ex: for res in ex.map(work, targets): if res: found += 1 print(f" + {res[0]} -> {res[1]}", flush=True) print(f"[A] nouveaux connectables: {found}") def build_record_from_html(domain, status, final_url, html): text = html[:400000] title = re.search(r"]*>(.*?)", text, re.S | re.I) rec = { "domain": domain, "checked_at": time.strftime("%Y-%m-%d"), "status": status, "final_url": final_url or f"https://{domain}", "active": True, "final_domain": domain, "title": re.sub(r"\s+", " ", title.group(1)).strip()[:200] if title else "", "platform": detect_platform(text, {}), "catalog_endpoint": None, "catalog_count_hint": None, "has_cart": bool(CART_RE.search(text)), "made_in_qc_wording": bool(MADE_RE.search(text)), "qc_postal": (POSTAL_RE.search(text) or [None]) and (POSTAL_RE.search(text).group(0) if POSTAL_RE.search(text) else None), "qc_phone": AREA_RE.search(text).group(0) if AREA_RE.search(text) else None, "mentions_quebec": bool(QC_WORD_RE.search(text)), "tld_quebec": domain.endswith(".quebec") or domain.endswith(".qc.ca"), "socials": list(dict.fromkeys(SOCIAL_RE.findall(text)))[:4], "language": None, "via": "scrapfly", } return rec def phase_b(limit=None): if not available(): print("[B] SCRAPFLY_API_KEY manquant — phase B sautée") return cands = [json.loads(l)["domain"] for l in open(os.path.join(ROOT, "data", "enriched", "candidates.jsonl"))] dead = [] for dom in cands: cpath = os.path.join(CACHE, dom + ".json") try: rec = json.load(open(cpath)) if not rec.get("active"): dead.append(dom) except Exception: dead.append(dom) if limit: dead = dead[:limit] print(f"[B] {len(dead)} candidats inactifs — retentative directe puis Scrapfly") revived = 0 def work(dom): # 1) direct rapide (les échecs transitoires) try: r = requests.get(f"https://{dom}", headers=HDRS, timeout=12, allow_redirects=True) if r.status_code == 200 and len(r.text) > 2000: return dom, 200, r.url, r.text, "direct" except Exception: pass # 2) scrapfly anti-bot try: status, content = scrapfly_get(f"https://{dom}") if status == 200 and len(content) > 2000: return dom, 200, f"https://{dom}", content, "scrapfly" except Exception: pass return None with cf.ThreadPoolExecutor(6) as ex: for res in ex.map(work, dead): if not res: continue dom, status, final_url, html, via = res rec = build_record_from_html(dom, status, final_url, html) rec["via"] = via # sonde les endpoints catalogue dans la foulée plat, ep = probe_endpoints(dom) if ep: rec["platform"], rec["catalog_endpoint"] = plat, ep json.dump(rec, open(os.path.join(CACHE, dom + ".json"), "w")) revived += 1 print(f" ✚ {dom} (via {via}, plat={rec['platform'] or '-'}, ep={rec['catalog_endpoint'] or '-'})", flush=True) print(f"[B] ressuscités: {revived}/{len(dead)}") if __name__ == "__main__": what = sys.argv[1] if len(sys.argv) > 1 else "ab" if "a" in what: phase_a() if "b" in what: phase_b()