`exec/elitis_feed.py`

136 lines
#!/usr/bin/env python3
"""Elitis feed-first scraper (D). $0 - local curl + node-eval of the elitis.fr
__NUXT__ hydration payload. NO Shopify writes, NO Browserbase, NO paid vision.

Pipeline:
  1. sitemap-en.xml -> wall/panoramic/HP-contract collection URLs (~217).
  2. fetch each (polite delay), extract the window.__NUXT__ IIFE blob.
  3. node extract_nuxt.js -> [{name, slug, n, images:[originals]}] per page.
  4. aggregate pattern -> {collection, images} ; write elitis_feed.json.
  5. coverage: accent-normalized match vs our Needs-Image drafts + missing patterns.

Real brand domain is elitis.fr (NOT elitis.com = an IT consultancy).
Images come from backoffice.elitis.fr/media/cache/resolve/original/... (public)."""
import sys
import os
import re
import json
import subprocess
import time
import urllib.request
import unicodedata

HERE = os.path.dirname(__file__)
ROOT = os.path.join(HERE, "..")
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
SITEMAP = "https://elitis.fr/sitemap-en.xml"
WALL_CATS = ("walls", "panoramic", "high-performance-contract")
FEED_OUT = os.path.join(ROOT, "elitis_feed.json")
NUXT_JS = os.path.join(HERE, "extract_nuxt.js")


def norm(s):
    s = unicodedata.normalize("NFKD", s or "")
    s = "".join(c for c in s if not unicodedata.combining(c))
    return " ".join(s.casefold().split())


def fetch(url, timeout=30):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read().decode("utf-8", "replace")


def collection_urls():
    xml = fetch(SITEMAP)
    urls = re.findall(r"<loc>(https://elitis\.fr/en/collections/[^<]+)</loc>", xml)
    out = [u for u in urls if u.split("/collections/", 1)[1].split("/")[0] in WALL_CATS
           and u.count("/collections/") and len(u.split("/collections/")[1].split("/")) == 2]
    return sorted(set(out))


def extract_page(html):
    m = re.search(r"window\.__NUXT__\s*=\s*", html)
    if not m:
        return []
    end = html.find("</script>", m.end())
    blob = html[m.end():end].strip().rstrip(";")
    p = subprocess.run(["node", NUXT_JS], input=blob, capture_output=True, text=True, timeout=60)
    try:
        data = json.loads(p.stdout)
    except Exception:
        return []
    return data if isinstance(data, list) else []


NAV = {"walls", "fabric", "accessoire", "panoramic", "outdoor", "in-outdoor",
       "high-performance-contract", "high performance contract"}


def run_scrape():
    urls = collection_urls()
    print(f"wall/panoramic/HP collection pages: {len(urls)}")
    feed = {}   # norm(pattern) -> {name, collection, images:set}
    errs = 0
    for i, u in enumerate(urls):
        coll = u.rsplit("/", 1)[1]
        try:
            pats = extract_page(fetch(u))
        except Exception as e:
            errs += 1
            print(f"  [err] {coll}: {str(e)[:70]}")
            continue
        for p in pats:
            nm = (p.get("name") or "").strip()
            if not nm or norm(nm) in {norm(x) for x in NAV}:
                continue
            key = norm(nm)
            slot = feed.setdefault(key, {"name": nm, "collection": coll, "images": []})
            for img in p.get("images", []):
                if img not in slot["images"]:
                    slot["images"].append(img)
        if (i + 1) % 25 == 0:
            print(f"  ...{i+1}/{len(urls)} patterns={len(feed)} errs={errs}")
        time.sleep(0.3)
    out = {k: {"name": v["name"], "collection": v["collection"],
               "n_images": len(v["images"]), "images": v["images"]}
           for k, v in feed.items()}
    json.dump(out, open(FEED_OUT, "w"), ensure_ascii=False, indent=1)
    print(f"\nWROTE {FEED_OUT}: {len(out)} patterns, "
          f"{sum(x['n_images'] for x in out.values())} image URLs, errs={errs}")
    return out


def coverage(feed=None):
    if feed is None:
        feed = json.load(open(FEED_OUT))
    feed_norm = set(feed.keys())
    # Needs-Image draft patterns: derive from staging join + the 47 missing list is implicit
    sj = os.path.join(ROOT, "OUT_577_staging_join.csv")
    ndc = os.path.join(ROOT, "OUT_577_no_catalog_data.csv")
    import csv
    want = {}
    for path, tag in [(sj, "staging"), (ndc, "no-data")]:
        if not os.path.exists(path):
            continue
        for r in csv.DictReader(open(path)):
            pat = r.get("pattern") or ""
            if pat:
                want.setdefault(norm(pat), (pat, tag))
    hit = {k: v for k, v in want.items() if k in feed_norm}
    miss = {k: v for k, v in want.items() if k not in feed_norm}
    print(f"\nCOVERAGE vs list patterns needing data: {len(want)} distinct")
    print(f"  COVERED by feed (>=1 image): {len(hit)}")
    print(f"  still missing               : {len(miss)}")
    print("  -- still-missing patterns --")
    for k, (pat, tag) in sorted(miss.items()):
        print(f"    {pat!r} [{tag}]")


if __name__ == "__main__":
    if "--coverage" in sys.argv:
        coverage()
    else:
        feed = run_scrape()
        coverage(feed)