admin/Meal-Planner
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: phase r1+r2 recovery + r3-0 swiftly api ingestion

Browse Source 
R1 stabilization: pytest harness with transactional db fixture, smoke
+ alembic + auth + scrape + approval + swiftly tests, github actions
ci yaml. Bearer-token admin auth + signed-cookie session for family
ui mutations. Async POST /api/admin/scrape (BackgroundTasks, returns
202). Path canonicalization (no /list, /planned suffixes). DATABASE_URL
fail-fast on empty.

R2 deferred-risk spikes: live lucky california fetch (R2-A), full
email+per-voter approval click round trip with single-use enforcement
(R2-B, console email backend, sendgrid stub).

R3-0 phase 3 redesign: replaced playwright html scraper with requests
based swiftly json api client. 17 categories, ~10k products per scrape,
upsert by (source, external_id). 401 surfaces actionable token-refresh
message via ScrapeLog.error_message.

Pre-existing defects fixed: shopping_list.py syntax error blocking app
import, MealPlan.votes orphan relationship, JSONB(astext=True) invalid
kwarg, missing requests dep, calorie_target schema drift, every SQLEnum
needed values_callable, 0001 had empty downgrade(), seed had duplicate
ingredient rows.

Migrations added: 0003 grocery_item.description, 0004 family_profile.
calorie_target, 0005 grocery_item.external_id + source + composite index.

Verified: 31/31 pytest green, alembic upgrade->downgrade->upgrade clean,
frontend npm run build clean, live scrape 9,960 grocery_item rows in 36s.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
admin
2026-05-05 14:08:19 -07:00
co-authored by Claude Opus 4.7
parent b9434967ed
commit 8e89f793d5
58 changed files with 3594 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/scraper/__init__.py
+2 -2
View File
@@ -1,4 +1,4 @@
from .base import BaseScraper
from .lucky_ca_scraper import LuckyCaliforniaScraper
from .lucky_ca_scraper import LuckyCaliforniaScraper, SwiftlyAuthError
__all__ = ["BaseScraper", "LuckyCaliforniaScraper"]
__all__ = ["BaseScraper", "LuckyCaliforniaScraper", "SwiftlyAuthError"]
backend/app/scraper/lucky_ca_scraper.py
+292 -147
View File
@@ -1,180 +1,325 @@
"""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
from typing import Dict, Any, List, Optional
import time
import urllib.parse
from datetime import datetime
from urllib.parse import urljoin
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, Iterator, List, Optional, Tuple
import requests
from bs4 import BeautifulSoup
from .base import SeleniumScraper
from app.config import settings
logger = logging.getLogger(__name__)
class LuckyCaliforniaScraper(SeleniumScraper):
def __init__(self, base_url: str = "https://luckysupermarkets.com"):
super().__init__(base_url=base_url, rate_limit_seconds=3.0)
self.ingredients_cache = {}
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]:
logger.info("Starting Lucky California scrape")
result = {
"source": "lucky_california",
"""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": datetime.now().isoformat(),
"items_scraped": 0,
"items": []
"started_at": started,
"completed_at": datetime.now().isoformat(),
"items_scraped": len(items),
"items": items,
"status": "success",
}
try:
featured_items = self.scrape_featured_coupons()
result["items"].extend(featured_items)
result["items_scraped"] = len(featured_items)
result["completed_at"] = datetime.now().isoformat()
result["status"] = "success"
logger.info(f"Lucky California scrape complete: {result['items_scraped']} items")
except Exception as e:
logger.error(f"Lucky California scrape failed: {e}")
result["status"] = "failed"
result["error_message"] = str(e)
result["completed_at"] = datetime.now().isoformat()
return result
def scrape_featured_coupons(self) -> List[Dict[str, Any]]:
url = f"{self.base_url}/coupons/Coupon%2Flu-featured-in-ad"
logger.info(f"Scraping featured coupons from {url}")
page = self.get_browser_page(url)
content = page.content()
page.close()
soup = BeautifulSoup(content, "html.parser")
items = []
coupon_items = soup.find_all("div", class_=re.compile(r"coupon|item|product", re.I))
if not coupon_items:
headline = soup.find("h1")
if headline:
logger.info(f"Page loaded, headline: {headline.get_text().strip()}")
titles = soup.find_all(["h2", "h3", "a"], string=re.compile(r"\$[\d\.]+"))
for title_elem in titles[:20]:
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:
item = self._parse_coupon_item(title_elem)
if item:
items.append(item)
except Exception as e:
logger.debug(f"Failed to parse item: {e}")
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
logger.info(f"Found {len(items)} coupon items")
# ------------------------------------------------------------------
# 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
def _parse_coupon_item(self, element) -> Optional[Dict[str, Any]]:
text = element.get_text().strip()
# ------------------------------------------------------------------
# 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.
price_match = re.search(r'\$[\d,]+\.?\d*', text)
if not price_match:
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_str = price_match.group().replace("$", "").replace(",", "")
try:
price = float(price_str)
except ValueError:
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
name_elem = element.find_parent("a") or element.find_parent("div")
name = text.split("$")[0].strip() if "$" in text else text
name = re.sub(r'\s+', " ", name).strip()[:200]
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
if not name or len(name) < 3:
return None
image_url: Optional[str] = None
primary_image = product.get("primaryImage")
if isinstance(primary_image, dict):
image_url = primary_image.get("url")
image_url = None
img_elem = element.find_parent().find("img") if element.find_parent() else None
if img_elem and img_elem.get("src"):
image_url = img_elem["src"]
brand = product.get("brand")
description = product.get("description")
product_url = None
link_elem = element.find_parent("a") if element.find_parent() else element.find("a")
if link_elem and link_elem.get("href"):
product_url = urljoin(self.base_url, link_elem["href"])
item = {
"name": name,
"current_price": price,
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": product_url,
"is_on_sale": True,
"product_url": None, # Swiftly does not expose a public product URL
"scraped_at": datetime.now().isoformat(),
"scraped_url": self.base_url
"scraped_url": f"{cls.__name__}:{source_slug}" if source_slug else None,
}
return item
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _parse_price(text: Optional[str]) -> Tuple[Optional[Decimal], Optional[str]]:
"""Extract ``(price, unit)`` from strings like ``"$3.49 /lb"``.
def scrape_produce(self) -> List[Dict[str, Any]]:
url = f"{self.base_url}/coupons/Coupon%2Flu-produce"
logger.info(f"Scraping produce from {url}")
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
page = self.get_browser_page(url)
content = page.content()
page.close()
@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
soup = BeautifulSoup(content, "html.parser")
items = []
for item_elem in soup.find_all("div", class_=re.compile(r"product|item|coupon")):
try:
item = self._parse_coupon_item(item_elem)
if item:
item["aisle"] = "Produce"
items.append(item)
except Exception:
continue
return items
def scrape_meat_seafood(self) -> List[Dict[str, Any]]:
url = f"{self.base_url}/coupons/Coupon%2Flu-meat-seafood"
logger.info(f"Scraping meat & seafood from {url}")
page = self.get_browser_page(url)
content = page.content()
page.close()
soup = BeautifulSoup(content, "html.parser")
items = []
for item_elem in soup.find_all("div", class_=re.compile(r"product|item|coupon")):
try:
item = self._parse_coupon_item(item_elem)
if item:
item["aisle"] = "Meat"
items.append(item)
except Exception:
continue
return items
def scrape_dairy_eggs(self) -> List[Dict[str, Any]]:
url = f"{self.base_url}/coupons/Coupon%2Flu-dairy-eggs"
logger.info(f"Scraping dairy & eggs from {url}")
page = self.get_browser_page(url)
content = page.content()
page.close()
soup = BeautifulSoup(content, "html.parser")
items = []
for item_elem in soup.find_all("div", class_=re.compile(r"product|item|coupon")):
try:
item = self._parse_coupon_item(item_elem)
if item:
item["aisle"] = "Dairy"
items.append(item)
except Exception:
continue
return items
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()