feat: implement Lucky California scraper with Playwright + BeautifulSoup

- Add BaseScraper with rate limiting, retries, session management
- Add LuckyCaliforniaScraper with Playwright for dynamic content
- Add ScraperService to save scraped items to grocery_item table
- Connect /api/admin/scrape to ScraperService
- Update ORIENTATION.md phase table
This commit is contained in:
2026-05-04 20:50:27 -07:00
parent c735d21661
commit 933a0cc9db
6 changed files with 500 additions and 26 deletions
+180
View File
@@ -0,0 +1,180 @@
import logging
import re
from typing import Dict, Any, List, Optional
from datetime import datetime
from urllib.parse import urljoin
from bs4 import BeautifulSoup
from .base import SeleniumScraper
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 = {}
def scrape(self) -> Dict[str, Any]:
logger.info("Starting Lucky California scrape")
result = {
"source": "lucky_california",
"scrape_type": "weekly_ad",
"started_at": datetime.now().isoformat(),
"items_scraped": 0,
"items": []
}
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]:
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}")
continue
logger.info(f"Found {len(items)} coupon items")
return items
def _parse_coupon_item(self, element) -> Optional[Dict[str, Any]]:
text = element.get_text().strip()
price_match = re.search(r'\$[\d,]+\.?\d*', text)
if not price_match:
return None
price_str = price_match.group().replace("$", "").replace(",", "")
try:
price = float(price_str)
except ValueError:
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]
if not name or len(name) < 3:
return None
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"]
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,
"image_url": image_url,
"product_url": product_url,
"is_on_sale": True,
"scraped_at": datetime.now().isoformat(),
"scraped_url": self.base_url
}
return item
def scrape_produce(self) -> List[Dict[str, Any]]:
url = f"{self.base_url}/coupons/Coupon%2Flu-produce"
logger.info(f"Scraping produce 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"] = "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