Public Access
adminandClaude Opus 4.7
dfd79a9d08
fetch_category() now calls swiftly_auth.get_token() to mint a fresh Firebase JWT on demand when no explicit bearer_token override is pinned by tests. The cache short-circuit means the per-call mint overhead is ~zero in the steady state. - Removed the empty-token short-circuit; auto-mint makes it moot - Updated _AUTH_ERROR_MESSAGE: 401-after-mint now points at the spec (Lucky tightening anon-auth) rather than asking for manual capture - Replaced test_swiftly_auth_error_when_token_missing with a positive test that verifies fetch_category mints when bearer_token is None - bearer_token constructor arg preserved for the 401-path test Full suite: 92/92 green. Live verification via scripts/spike_swiftly_ingest.py --confirm-live deferred to next step per HANDOFF AM-2 halt boundary. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
332 lines
13 KiB
Python
332 lines
13 KiB
Python
"""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
|
|
from app.services.swiftly_auth import (
|
|
SwiftlyAuthMintError,
|
|
get_token as _get_swiftly_token,
|
|
)
|
|
|
|
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 returned 401 with a freshly minted Firebase JWT — Lucky may have "
|
|
"tightened anon-auth restrictions. See docs/specs/2026-05-06-swiftly-token-auto-mint.md."
|
|
)
|
|
|
|
|
|
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:
|
|
# When ``bearer_token`` is None, fetch_category() mints via Firebase
|
|
# REST through ``swiftly_auth.get_token()``. Tests pin a fixed token
|
|
# to exercise the 401 path without going through Firebase.
|
|
self.bearer_token = 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."""
|
|
token = self.bearer_token if self.bearer_token else _get_swiftly_token()
|
|
|
|
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 {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()
|