foodlinkk-command-center/tools-api/app/connectors/reclamefolder.py

"""Fetch live supermarket folders from reclamefolder.nl — all chains."""
from __future__ import annotations

import json
import re
import xml.etree.ElementTree as ET
from datetime import date, datetime, timezone
from typing import Any, Optional
from urllib.parse import unquote
from urllib.request import HTTPCookieProcessor, Request, build_opener

from app.db import execute, fetch_all, fetch_one

BASE = "https://www.reclamefolder.nl"
SUPERMARKT_URL = f"{BASE}/categorieen/supermarkt/"
SITEMAP_RETAILERS = f"{BASE}/sitemap/retailers.xml"
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"

SUPERMARKT_KEYWORDS = (
    "supermarkt", "markt", "ah", "jumbo", "lidl", "aldi", "plus", "dirk",
    "coop", "boni", "vomar", "deka", "ekoplaza", "mitra", "spar", "hoogvliet",
    "food", "gall", "poiesz", "nettorama",
)

_opener = None


def _get_opener():
    global _opener
    if _opener is None:
        _opener = build_opener(HTTPCookieProcessor())
    return _opener


def _fetch_html(url: str) -> str:
    headers = {"User-Agent": USER_AGENT, "Accept-Language": "nl-NL,nl;q=0.9"}
    html = _get_opener().open(Request(url, headers=headers), timeout=35).read().decode("utf-8", "ignore")
    if "__NEXT_DATA__" not in html:
        cb = re.search(r"decodeURIComponent\('([^']+)'\)", html)
        if cb:
            html = _get_opener().open(Request(unquote(cb.group(1)), headers=headers), timeout=35).read().decode("utf-8", "ignore")
    return html


def _fetch_page_props(url: str) -> dict[str, Any]:
    html = _fetch_html(url)
    match = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.+?)</script>', html)
    if not match:
        return {}
    data = json.loads(match.group(1))
    return data.get("props", {}).get("pageProps", {}) or {}


def _parse_day(raw: Optional[str]) -> Optional[date]:
    if not raw:
        return None
    try:
        return datetime.fromisoformat(raw.replace("Z", "+00:00")).date()
    except Exception:
        return None


def _folder_item(row: dict[str, Any], source: str = "category") -> Optional[dict[str, Any]]:
    retailer = row.get("retailer") or {}
    if source == "retailer_page":
        chain = retailer.get("name") or row.get("name", "")
        edition_id = row.get("id")
        valid_label = ""
        cover = row.get("cover") or {}
        folder_name = row.get("name") or "Folder"
    else:
        chain = retailer.get("name")
        edition_id = row.get("editionId") or row.get("id")
        valid_label = row.get("validLabel") or ""
        cover = row.get("cover") or {}
        folder_name = valid_label or "Folder"

    if not edition_id or not chain:
        return None

    valid_to = _parse_day(row.get("validThru"))
    today = datetime.now(timezone.utc).date()
    if valid_to and valid_to < today:
        return None

    permaname = retailer.get("permaname") or ""
    url = f"{BASE}/f/folders/{edition_id}/"
    title = f"{chain} — {folder_name}" if folder_name != chain else f"{chain} folder ({valid_label})".strip(" ()")

    return {
        "chain": chain,
        "title": title,
        "folder_path": url,
        "folder_label": valid_label or folder_name or chain,
        "description": f"Reclamefolder.nl · {valid_label or folder_name}".strip(" ·"),
        "valid_from": _parse_day(row.get("validFrom")),
        "valid_to": valid_to,
        "image_url": cover.get("imageUrl") if isinstance(cover, dict) else None,
        "status": "active",
        "promo_type": row.get("name") or "folder",
        "source": "reclamefolder.nl",
        "metadata": {
            "edition_id": str(edition_id),
            "version_id": str(row.get("versionId") or row.get("id") or ""),
            "retailer_permaname": permaname,
            "retailer_url": f"{BASE}/winkels/{permaname}/" if permaname else "",
            "folder_type": row.get("name") or "folder",
        },
    }


def fetch_retailer_slugs() -> list[str]:
    try:
        req = Request(SITEMAP_RETAILERS, headers={"User-Agent": USER_AGENT})
        data = _get_opener().open(req, timeout=45).read()
        text = data.decode("utf-8", "ignore")
        slugs = set()
        for m in re.finditer(r"https://www\.reclamefolder\.nl/winkels/([a-z0-9-]+)/", text):
            slug = m.group(1)
            if "vestiging" in slug:
                continue
            if any(k in slug for k in SUPERMARKET_KEYWORDS):
                slugs.add(slug)
        return sorted(slugs)
    except Exception:
        return [
            "albert-heijn", "jumbo", "lidl", "plus", "dirk", "aldi", "dekamarkt",
            "ekoplaza", "vomar", "coop-supermarkten", "boni-supermarkt", "mitra",
        ]


def fetch_supermarkt_folders(include_all_retailers: bool = True) -> list[dict[str, Any]]:
    seen: set[str] = set()
    items: list[dict[str, Any]] = []

    props = _fetch_page_props(SUPERMARKT_URL)
    for row in props.get("foldersFromProps") or []:
        item = _folder_item(row, "category")
        if item and item["metadata"]["edition_id"] not in seen:
            seen.add(item["metadata"]["edition_id"])
            items.append(item)

    if include_all_retailers:
        for slug in fetch_retailer_slugs():
            try:
                rprops = _fetch_page_props(f"{BASE}/winkels/{slug}/")
                retailer = rprops.get("retailer") or {}
                for folder in retailer.get("folders") or []:
                    folder = dict(folder)
                    folder["retailer"] = retailer
                    item = _folder_item(folder, "retailer_page")
                    if item and item["metadata"]["edition_id"] not in seen:
                        seen.add(item["metadata"]["edition_id"])
                        items.append(item)
            except Exception:
                continue

    items.sort(key=lambda x: (x.get("chain") or "", x.get("valid_to") or date.max))
    return items


def sync_to_db() -> dict[str, Any]:
    folders = fetch_supermarkt_folders()
    execute(
        "UPDATE promo_campaigns SET status = 'expired', updated_at = NOW() WHERE source = 'reclamefolder.nl'"
    )
    inserted = 0
    chains: set[str] = set()
    for f in folders:
        fetch_one(
            """INSERT INTO promo_campaigns
               (chain, title, folder_path, folder_label, description, valid_from, valid_to,
                status, promo_type, image_url, source, metadata)
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb)
               RETURNING id""",
            (
                f["chain"], f["title"], f["folder_path"], f["folder_label"],
                f["description"], f["valid_from"], f["valid_to"],
                f["status"], f["promo_type"], f.get("image_url"),
                f["source"], json.dumps(f["metadata"]),
            ),
        )
        inserted += 1
        chains.add(f["chain"])
    return {
        "ok": True,
        "source": "reclamefolder.nl",
        "synced": inserted,
        "chains": len(chains),
        "items": folders,
    }


def list_cached(limit: int = 200) -> list[dict[str, Any]]:
    rows = fetch_all(
        """SELECT * FROM promo_campaigns
           WHERE source = 'reclamefolder.nl' AND status = 'active'
           ORDER BY valid_to ASC NULLS LAST, chain ASC
           LIMIT %s""",
        (limit,),
    )
    return [dict(r) for r in rows]


def list_chains() -> list[str]:
    rows = fetch_all(
        """SELECT DISTINCT chain FROM promo_campaigns
           WHERE source = 'reclamefolder.nl' AND status = 'active' AND chain IS NOT NULL
           ORDER BY chain"""
    )
    return [r["chain"] for r in rows if r.get("chain")]