Meal-Planner/backend/app/scraper/lucky_ca_scraper.py

"""Lucky California / Swiftly product API client.

Replaces the prior Playwright + HTML coupon-card path (R2-A) with the
underlying JSON API the website itself calls. The API exposes the full
inventory per category — no rendering, no Chromium, and far more items
than the visible coupon strip (256 in `Product/meat_seafood` vs the 11
the HTML parser saw).

Two endpoints (no public docs; reverse-engineered from the network panel
on luckysupermarkets.com — see ``.agent/context.md`` "Swiftly API"):

  GET https://luckysupermarkets.com/categories
      → HTML page; anchors with ``class="swiftlyCouponCategory"`` carry
        ``href="/categories/Product%2F<slug>"``.

  GET https://prod.swiftlyapi.net/search/api/v1/products/categories
      ?cat=<slug>&store=<store_id>&limit=10000
      Authorization: Bearer <SWIFTLY_BEARER_TOKEN>
      → ``{"products": {"info": {...}, "items": [...], "facets": [...]}}``

The bearer token expires roughly hourly. On 401 we raise
``SwiftlyAuthError`` so the background runner records the error_message
that asks the admin to refresh ``SWIFTLY_BEARER_TOKEN`` and retry.
"""
from __future__ import annotations

import logging
import re
import time
import urllib.parse
from datetime import datetime
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, Iterator, List, Optional, Tuple

import requests
from bs4 import BeautifulSoup

from app.config import settings

logger = logging.getLogger(__name__)


class SwiftlyAuthError(Exception):
    """Raised when the Swiftly API returns 401.

    The exception message is surfaced verbatim to the ScrapeLog row by
    ``_run_scrape_in_background``; keep it actionable.
    """


_USER_AGENT = (
    "MealPlannerBot/1.0 (+https://mealplanner.local; contact peter@research.bike)"
)

_AUTH_ERROR_MESSAGE = (
    "SWIFTLY_BEARER_TOKEN expired — request a fresh token from the user "
    "(capture from luckysupermarkets.com network tab on a /search/api/v1 request)"
)


class LuckyCaliforniaScraper:
    """Swiftly product-API client for the Lucky California banner.

    The class name is preserved (``LuckyCaliforniaScraper``) so existing
    import sites in ``app.scraper.__init__`` and
    ``ScraperService._do_scrape`` keep working. Internally it is no
    longer a ``BaseScraper`` subclass: that class wires Playwright +
    ``_get`` retry-with-swallow, neither of which we want here. The
    client uses two ``requests.Session`` objects so the bearer header
    is scoped strictly to the API host (the public categories page
    is unauthenticated).
    """

    SOURCE = "lucky_california"
    SLUG_HREF_RE = re.compile(
        r'<a\b(?=[^>]*\bclass="swiftlyCouponCategory")(?=[^>]*\bhref="([^"]+)")[^>]*>',
        re.IGNORECASE,
    )

    def __init__(
        self,
        *,
        bearer_token: Optional[str] = None,
        store_id: Optional[str] = None,
        api_base: Optional[str] = None,
        categories_url: Optional[str] = None,
        rate_limit_seconds: float = 0.75,
        timeout: int = 60,
    ) -> None:
        self.bearer_token = bearer_token or settings.SWIFTLY_BEARER_TOKEN
        self.store_id = store_id or settings.LUCKY_STORE_ID
        self.api_base = (api_base or settings.SWIFTLY_API_BASE).rstrip("/")
        self.categories_url = categories_url or settings.SWIFTLY_CATEGORIES_URL
        self.rate_limit_seconds = rate_limit_seconds
        self.timeout = timeout
        self._last_request = 0.0

        self.public_session = requests.Session()
        self.public_session.headers.update(
            {
                "User-Agent": _USER_AGENT,
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "en-US,en;q=0.5",
            }
        )

        self.api_session = requests.Session()
        self.api_session.headers.update(
            {
                "User-Agent": _USER_AGENT,
                "Accept": "application/json",
            }
        )
        # base_url retained for code paths that still introspect it.
        self.base_url = "https://luckysupermarkets.com"

    # ------------------------------------------------------------------
    # Lifecycle (no-op; preserves cleanup() contract from BaseScraper).
    # ------------------------------------------------------------------
    def cleanup(self) -> None:
        try:
            self.public_session.close()
        finally:
            self.api_session.close()

    # ------------------------------------------------------------------
    # Public scrape entrypoints
    # ------------------------------------------------------------------
    def scrape(self) -> Dict[str, Any]:
        """Synchronous top-level scrape.

        Kept for the legacy ``ScraperService.run_scrape`` path used by
        tests. Walks every category and returns ``{"items": [...], ...}``
        with mapped product dicts ready for ``_save_grocery_item``.
        """
        started = datetime.now().isoformat()
        items: List[Dict[str, Any]] = list(self.fetch_all())
        return {
            "source": self.SOURCE,
            "scrape_type": "weekly_ad",
            "started_at": started,
            "completed_at": datetime.now().isoformat(),
            "items_scraped": len(items),
            "items": items,
            "status": "success",
        }

    def fetch_all(self) -> Iterator[Dict[str, Any]]:
        """Yield mapped product dicts across every discovered category."""
        slugs = self.discover_categories()
        logger.info("Swiftly: discovered %d categories", len(slugs))
        for slug in slugs:
            try:
                products = self.fetch_category(slug)
            except SwiftlyAuthError:
                # Hard fail — token must be refreshed before any further work.
                raise
            except requests.RequestException as exc:
                logger.warning("Swiftly: skipping %s after error: %s", slug, exc)
                continue
            aisle = self._aisle_from_slug(slug)
            for product in products:
                mapped = self.map_product(product, aisle=aisle, source_slug=slug)
                if mapped is not None:
                    yield mapped

    # ------------------------------------------------------------------
    # Category discovery
    # ------------------------------------------------------------------
    def discover_categories(self) -> List[str]:
        """Fetch the categories page and return the list of API slugs.

        The slugs look like ``Product/meat_seafood``. Order is preserved
        from the HTML (which is the order shown to the user).
        """
        self._rate_limit()
        resp = self.public_session.get(self.categories_url, timeout=self.timeout)
        resp.raise_for_status()
        return self.parse_categories_html(resp.text)

    @classmethod
    def parse_categories_html(cls, html: str) -> List[str]:
        """Pure parser used by tests against a saved fixture."""
        slugs: List[str] = []
        seen: set[str] = set()
        for href in cls.SLUG_HREF_RE.findall(html):
            m = re.match(r"^/categories/(.+)$", href)
            if not m:
                continue
            slug = urllib.parse.unquote(m.group(1))
            if slug not in seen:
                seen.add(slug)
                slugs.append(slug)
        return slugs

    # ------------------------------------------------------------------
    # Per-category fetch
    # ------------------------------------------------------------------
    def fetch_category(self, slug: str) -> List[Dict[str, Any]]:
        """Fetch every product in a category. Raises ``SwiftlyAuthError`` on 401."""
        if not self.bearer_token:
            raise SwiftlyAuthError(_AUTH_ERROR_MESSAGE)

        self._rate_limit()
        url = f"{self.api_base}/search/api/v1/products/categories"
        params = {"cat": slug, "store": self.store_id, "limit": 10000}
        headers = {"Authorization": f"Bearer {self.bearer_token}"}
        resp = self.api_session.get(
            url, params=params, headers=headers, timeout=self.timeout
        )
        if resp.status_code == 401:
            raise SwiftlyAuthError(_AUTH_ERROR_MESSAGE)
        resp.raise_for_status()
        payload = resp.json()
        return self.parse_category_response(payload)

    @staticmethod
    def parse_category_response(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Pure parser used by tests against a saved fixture."""
        if not isinstance(payload, dict):
            return []
        products = payload.get("products") or {}
        items = products.get("items")
        if not isinstance(items, list):
            return []
        return items

    # ------------------------------------------------------------------
    # Field mapping
    # ------------------------------------------------------------------
    @classmethod
    def map_product(
        cls,
        product: Dict[str, Any],
        *,
        aisle: Optional[str] = None,
        source_slug: Optional[str] = None,
    ) -> Optional[Dict[str, Any]]:
        """Convert one Swiftly product dict to a grocery_item-ready dict.

        Returns ``None`` for products with no parseable price or no name —
        those are usually placeholder/unavailable rows.
        """
        external_id = product.get("id")
        name = (product.get("name") or "").strip()
        if not name:
            return None

        price_block = (product.get("price") or {}).get("ok") or {}
        reg_price, reg_unit = cls._parse_price(price_block.get("regPriceText"))
        promo = price_block.get("promoArea") or {}
        sale_price, sale_unit = cls._parse_price(promo.get("promoText"))

        if reg_price is None and sale_price is None:
            # No usable price; skip rather than persist garbage.
            return None

        unit = sale_unit or reg_unit
        is_on_sale = sale_price is not None and reg_price is not None and sale_price < reg_price
        # current_price = "what the customer pays today" → sale_price when on sale.
        current_price = sale_price if is_on_sale else reg_price

        image_url: Optional[str] = None
        primary_image = product.get("primaryImage")
        if isinstance(primary_image, dict):
            image_url = primary_image.get("url")

        brand = product.get("brand")
        description = product.get("description")

        return {
            "external_id": str(external_id) if external_id is not None else None,
            "source": cls.SOURCE,
            "name": name[:300],
            "brand": (brand or None),
            "description": description,
            "current_price": current_price,
            "regular_price": reg_price,
            "sale_price": sale_price,
            "is_on_sale": bool(is_on_sale),
            "unit": unit,
            "aisle": aisle,
            "image_url": image_url,
            "product_url": None,  # Swiftly does not expose a public product URL
            "scraped_at": datetime.now().isoformat(),
            "scraped_url": f"{cls.__name__}:{source_slug}" if source_slug else None,
        }

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------
    @staticmethod
    def _parse_price(text: Optional[str]) -> Tuple[Optional[Decimal], Optional[str]]:
        """Extract ``(price, unit)`` from strings like ``"$3.49 /lb"``.

        Returns ``(None, None)`` when ``text`` is empty or unparseable.
        """
        if not text:
            return None, None
        m = re.search(r"\$\s*([\d,]+(?:\.\d+)?)", text)
        if not m:
            return None, None
        raw = m.group(1).replace(",", "")
        try:
            price = Decimal(raw)
        except (InvalidOperation, ValueError):
            return None, None
        unit_match = re.search(r"/\s*([A-Za-z]+)", text)
        unit = unit_match.group(1).lower() if unit_match else None
        return price, unit

    @staticmethod
    def _aisle_from_slug(slug: str) -> Optional[str]:
        """``Product/meat_seafood`` → ``meat_seafood``."""
        if not slug:
            return None
        if "/" in slug:
            return slug.rsplit("/", 1)[1]
        return slug

    def _rate_limit(self) -> None:
        elapsed = time.time() - self._last_request
        if elapsed < self.rate_limit_seconds:
            time.sleep(self.rate_limit_seconds - elapsed)
        self._last_request = time.time()