Public Access
302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""Scraper service.
|
|
|
|
Two entry points:
|
|
|
|
- ``enqueue_scrape(db, source, scrape_type, background_tasks)``: synchronously
|
|
inserts a ``ScrapeLog`` row in status ``STARTED`` (the existing enum has no
|
|
``pending`` member; ``STARTED`` is reused for the queued state) and registers
|
|
``_run_scrape_in_background`` to fire after the response is sent.
|
|
- ``_run_scrape_in_background(log_id, source, scrape_type)``: runs in a
|
|
FastAPI background task with its OWN ``SessionLocal()`` (the request-scoped
|
|
``db`` is closed by the time this fires). Writes terminal status
|
|
(``SUCCESS``/``FAILED``) and ``error_message``.
|
|
|
|
``ScraperService.run_scrape`` is preserved for direct/test invocation; the
|
|
``/api/admin/scrape`` endpoint now goes through ``enqueue_scrape``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Dict, Optional
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import BackgroundTasks
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import SessionLocal
|
|
from app.models import GroceryItem, Ingredient, ScrapeLog, ScrapeStatus
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API: enqueue + background runner
|
|
# ---------------------------------------------------------------------------
|
|
def enqueue_scrape(
|
|
db: Session,
|
|
*,
|
|
source: str,
|
|
scrape_type: str,
|
|
background_tasks: BackgroundTasks,
|
|
) -> ScrapeLog:
|
|
"""Create the ScrapeLog row, commit, and schedule the background scrape.
|
|
|
|
Returns the persisted ``ScrapeLog`` instance (refreshed). The actual
|
|
scraping work runs after FastAPI sends the 202 response.
|
|
"""
|
|
log = ScrapeLog(
|
|
id=uuid4(),
|
|
source=source,
|
|
scrape_type=scrape_type,
|
|
status=ScrapeStatus.STARTED,
|
|
started_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(log)
|
|
db.commit()
|
|
db.refresh(log)
|
|
|
|
background_tasks.add_task(
|
|
_run_scrape_in_background, log.id, source, scrape_type
|
|
)
|
|
return log
|
|
|
|
|
|
def _run_scrape_in_background(
|
|
log_id: UUID, source: str, scrape_type: str
|
|
) -> None:
|
|
"""Background entry point. Opens a fresh DB session — the request-scoped
|
|
session is gone by the time this runs.
|
|
"""
|
|
db: Session = SessionLocal()
|
|
try:
|
|
log = db.query(ScrapeLog).filter(ScrapeLog.id == log_id).first()
|
|
if log is None:
|
|
logger.error("ScrapeLog %s vanished before background run", log_id)
|
|
return
|
|
|
|
# No "running" state in the enum; STARTED already covers in-flight.
|
|
# Mark started_at fresh in case there was lag between enqueue and run.
|
|
try:
|
|
service = ScraperService(db)
|
|
saved_count, items_found = service._do_scrape(source=source)
|
|
log.status = ScrapeStatus.SUCCESS
|
|
log.items_scraped = saved_count
|
|
log.completed_at = datetime.now(timezone.utc)
|
|
log.duration_seconds = int(
|
|
(log.completed_at - log.started_at).total_seconds()
|
|
)
|
|
try:
|
|
from app.services.matcher import run_match_job
|
|
run_match_job(db, source_filter="lucky_california")
|
|
except Exception as e: # matcher failure must not flip scrape to FAILED
|
|
import logging
|
|
logging.exception("matcher failed after successful scrape: %s", e)
|
|
db.commit()
|
|
logger.info(
|
|
"Background scrape %s complete: %s/%s items saved",
|
|
log_id, saved_count, items_found,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 — must catch all to mark failed
|
|
logger.exception("Background scrape %s failed", log_id)
|
|
db.rollback()
|
|
# Re-fetch in case the rollback detached the instance.
|
|
log = db.query(ScrapeLog).filter(ScrapeLog.id == log_id).first()
|
|
if log is not None:
|
|
log.status = ScrapeStatus.FAILED
|
|
log.error_message = str(exc)
|
|
log.completed_at = datetime.now(timezone.utc)
|
|
if log.started_at:
|
|
log.duration_seconds = int(
|
|
(log.completed_at - log.started_at).total_seconds()
|
|
)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Service class (kept for direct/test use)
|
|
# ---------------------------------------------------------------------------
|
|
class ScraperService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def run_scrape(
|
|
self, source: str = "lucky_california", scrape_type: str = "weekly_ad"
|
|
) -> Dict[str, Any]:
|
|
"""Synchronous end-to-end scrape (legacy path). Creates the log row,
|
|
runs the scrape, commits terminal status. Used by direct callers and
|
|
tests; the API endpoint goes through ``enqueue_scrape``.
|
|
"""
|
|
log = ScrapeLog(
|
|
id=uuid4(),
|
|
source=source,
|
|
scrape_type=scrape_type,
|
|
status=ScrapeStatus.STARTED,
|
|
started_at=datetime.now(timezone.utc),
|
|
)
|
|
self.db.add(log)
|
|
self.db.commit()
|
|
|
|
logger.info("Starting %s %s scrape", source, scrape_type)
|
|
|
|
try:
|
|
saved_count, items_found = self._do_scrape(source=source)
|
|
log.status = ScrapeStatus.SUCCESS
|
|
log.items_scraped = saved_count
|
|
log.completed_at = datetime.now(timezone.utc)
|
|
log.duration_seconds = int(
|
|
(log.completed_at - log.started_at).total_seconds()
|
|
)
|
|
self.db.commit()
|
|
logger.info("Scrape complete: %s items saved", saved_count)
|
|
return {
|
|
"scrape_id": str(log.id),
|
|
"status": "success",
|
|
"items_scraped": saved_count,
|
|
"items_found": items_found,
|
|
}
|
|
except Exception as e: # noqa: BLE001
|
|
logger.error("Scrape failed: %s", e)
|
|
log.status = ScrapeStatus.FAILED
|
|
log.error_message = str(e)
|
|
log.completed_at = datetime.now(timezone.utc)
|
|
log.duration_seconds = int(
|
|
(log.completed_at - log.started_at).total_seconds()
|
|
)
|
|
self.db.commit()
|
|
return {
|
|
"scrape_id": str(log.id),
|
|
"status": "failed",
|
|
"error": str(e),
|
|
}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internals
|
|
# ------------------------------------------------------------------
|
|
def _do_scrape(self, *, source: str) -> tuple[int, int]:
|
|
"""Run the scraper and persist items. Returns (saved, found).
|
|
|
|
Raises any scraper exception (including ``SwiftlyAuthError``) to
|
|
the caller for status mapping.
|
|
"""
|
|
from app.scraper import LuckyCaliforniaScraper
|
|
|
|
scraper = LuckyCaliforniaScraper()
|
|
saved_count = 0
|
|
found_count = 0
|
|
try:
|
|
for item_data in scraper.fetch_all():
|
|
found_count += 1
|
|
if self._save_grocery_item(item_data) is not None:
|
|
saved_count += 1
|
|
finally:
|
|
scraper.cleanup()
|
|
return saved_count, found_count
|
|
|
|
def _save_grocery_item(
|
|
self, item_data: Dict[str, Any]
|
|
) -> Optional[GroceryItem]:
|
|
name = (item_data.get("name") or "").strip()
|
|
if not name:
|
|
return None
|
|
|
|
name_lower = name.lower()
|
|
source = item_data.get("source") or "lucky_california"
|
|
external_id = item_data.get("external_id")
|
|
|
|
ingredient = self.db.query(Ingredient).filter(
|
|
Ingredient.name_lower == name_lower
|
|
).first()
|
|
|
|
if not ingredient:
|
|
ingredient = Ingredient(
|
|
id=uuid4(),
|
|
name=name,
|
|
name_lower=name_lower,
|
|
aisle=item_data.get("aisle"),
|
|
typical_price=item_data.get("current_price"),
|
|
)
|
|
self.db.add(ingredient)
|
|
self.db.flush()
|
|
|
|
# Idempotency:
|
|
# 1) (source, external_id) when both present (Swiftly path);
|
|
# 2) fall back to (name, scraped_url) for legacy rows from R2-A.
|
|
existing: Optional[GroceryItem] = None
|
|
if external_id:
|
|
existing = self.db.query(GroceryItem).filter(
|
|
GroceryItem.source == source,
|
|
GroceryItem.external_id == external_id,
|
|
).first()
|
|
if existing is None:
|
|
existing = self.db.query(GroceryItem).filter(
|
|
GroceryItem.name == name,
|
|
GroceryItem.scraped_url == item_data.get("scraped_url"),
|
|
).first()
|
|
|
|
if existing:
|
|
existing.ingredient_id = ingredient.id
|
|
existing.brand = item_data.get("brand")
|
|
existing.current_price = item_data.get("current_price")
|
|
existing.regular_price = item_data.get("regular_price")
|
|
existing.unit = item_data.get("unit")
|
|
existing.aisle = item_data.get("aisle")
|
|
existing.image_url = item_data.get("image_url")
|
|
existing.product_url = item_data.get("product_url")
|
|
existing.description = item_data.get("description")
|
|
existing.is_on_sale = bool(item_data.get("is_on_sale", False))
|
|
existing.scraped_at = datetime.now(timezone.utc)
|
|
existing.scraped_url = item_data.get("scraped_url")
|
|
existing.external_id = external_id
|
|
existing.source = source
|
|
self.db.flush()
|
|
return existing
|
|
|
|
grocery_item = GroceryItem(
|
|
id=uuid4(),
|
|
ingredient_id=ingredient.id,
|
|
name=name,
|
|
brand=item_data.get("brand"),
|
|
current_price=item_data.get("current_price"),
|
|
regular_price=item_data.get("regular_price"),
|
|
unit=item_data.get("unit"),
|
|
aisle=item_data.get("aisle"),
|
|
image_url=item_data.get("image_url"),
|
|
product_url=item_data.get("product_url"),
|
|
description=item_data.get("description"),
|
|
is_on_sale=bool(item_data.get("is_on_sale", False)),
|
|
sale_start_date=item_data.get("sale_start_date"),
|
|
sale_end_date=item_data.get("sale_end_date"),
|
|
in_season=item_data.get("in_season", False),
|
|
scraped_at=datetime.now(timezone.utc),
|
|
scraped_url=item_data.get("scraped_url"),
|
|
external_id=external_id,
|
|
source=source,
|
|
)
|
|
|
|
self.db.add(grocery_item)
|
|
self.db.flush()
|
|
return grocery_item
|
|
|
|
def get_sale_items(self, limit: int = 50) -> list:
|
|
items = self.db.query(GroceryItem).filter(
|
|
GroceryItem.is_on_sale == True # noqa: E712 — SQLAlchemy idiom
|
|
).order_by(GroceryItem.scraped_at.desc()).limit(limit).all()
|
|
|
|
return [
|
|
{
|
|
"id": str(item.id),
|
|
"name": item.name,
|
|
"current_price": float(item.current_price) if item.current_price else None,
|
|
"regular_price": float(item.regular_price) if item.regular_price else None,
|
|
"aisle": item.aisle,
|
|
"image_url": item.image_url,
|
|
"product_url": item.product_url,
|
|
"description": item.description,
|
|
"scraped_at": item.scraped_at.isoformat() if item.scraped_at else None,
|
|
}
|
|
for item in items
|
|
]
|