feat(ebay): async sold-top jobs with polling, shared sold-scrape cache, and UI refresh

Author: antoine-gourgue

api/routes/ebay_market_route.py

@@ -6,6 +6,7 @@
 from typing import Annotated, Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel, Field
 
 from app_types.ebay_browse import (
     ConditionFilter,
@@ -22,6 +23,7 @@
 from services.ebay_sold_scrape_rate_limit import acquire_sold_scrape_slot
 from services.ebay_sold_scrape_service import ebay_fr_sold_search_url, scrape_sold_listings
 from services.ebay_sold_top_service import aggregate_top_sold
+from services.ebay_sold_top_worker import get_job, peek_items_sample, submit_job
 
 logger = logging.getLogger(__name__)
 
@@ -179,14 +181,33 @@ async def sold_scrape_html(
     Window goes up to ``720`` hours (30 days).
     """
     app = get_settings()
+
+    # If the worker has a fresh cached top result for the same (q, window),
+    # reuse its items_sample — saves an eBay roundtrip *and* the rate-limit
+    # slot, which matters when the user just searched in Top mode and
+    # switches to List mode.
+    cached_sample = peek_items_sample(q=q.strip(), window_hours=window_hours)
+    if cached_sample is not None:
+        return {
+            "query": q.strip(),
+            "window_hours": window_hours,
+            "items": cached_sample[:limit],
+            "error": None,
+            "ebay_sold_search_url": ebay_fr_sold_search_url(
+                q=q.strip(), page_size=min(60, max(limit, 10)),
+            ),
+            "source": "ebay_html_scrape_cached_from_top",
+            "cached": True,
+        }
+
     retry_after = await acquire_sold_scrape_slot(user.id, app.ebay_sold_scrape_min_interval_seconds)
     if retry_after > 0:
         iv = app.ebay_sold_scrape_min_interval_seconds
         raise HTTPException(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
             detail=(
-                f"Limite : une recherche « vendus eBay » toutes les {iv:g} s "
-                f"(réessayez dans {retry_after} s)."
+                f"Rate limit: at most one eBay sold-search every {iv:g} s "
+                f"(retry in {retry_after} s)."
             ),
             headers={"Retry-After": str(retry_after)},
         )
@@ -198,64 +219,94 @@ async def sold_scrape_html(
         "error": err,
         "ebay_sold_search_url": ebay_fr_sold_search_url(q=q.strip(), page_size=min(60, max(limit, 10))),
         "source": "ebay_html_scrape",
+        "cached": False,
     }
 
 
-@router.get("/sold-top", response_model=None)
-async def sold_top(
+class SoldTopSubmitBody(BaseModel):
+    """Body for ``POST /ebay/market/sold-top`` — schedules a background scrape."""
+
+    q: str = Field(min_length=2, max_length=256)
+    window_hours: float = Field(default=168, ge=1, le=720)
+    pages: int = Field(default=10, ge=1, le=20)
+    scrape_limit: int = Field(default=600, ge=10, le=1000)
+    top_limit: int = Field(default=20, ge=1, le=100)
+    min_count: int = Field(default=1, ge=1, le=20)
+
+
+@router.post("/sold-top", response_model=None, status_code=status.HTTP_202_ACCEPTED)
+async def sold_top_submit(
     user: Annotated[User, Depends(get_current_user)],
-    q: Annotated[str, Query(min_length=2, max_length=256)],
-    window_hours: Annotated[float, Query(ge=1, le=720)] = 168,
-    pages: Annotated[int, Query(ge=1, le=5)] = 2,
-    scrape_limit: Annotated[int, Query(ge=10, le=300)] = 180,
-    top_limit: Annotated[int, Query(ge=1, le=100)] = 30,
-    min_count: Annotated[int, Query(ge=1, le=20)] = 1,
+    body: SoldTopSubmitBody,
 ) -> dict[str, Any]:
     """
-    Top des cartes les plus vendues dans la fenêtre, agrégées depuis le
-    scrape HTML public eBay.fr.
+    Submit a background top-sold scrape job and return its ``job_id``
+    (consumed via ``GET /ebay/market/sold-top/{job_id}``).
 
-    Le résultat est trié par ``count`` (puis valeur cumulée). Fenêtre par
-    défaut : 7 jours (168 h) ; valeurs autorisées de 1 h à 720 h (30 j).
-    ``pages`` (1-5) déclenche autant de requêtes paginées vers eBay (60
-    annonces / page), avec déduplication par ``item_id``. Même rate-limit
-    utilisateur que ``/sold-scrape`` — un appel utilisateur peut donc
-    générer plusieurs requêtes vers eBay.
+    When a fresh cached result (TTL 15 min) exists for the same parameters,
+    the job comes back already in ``status="completed"`` with its
+    ``result`` populated — no eBay scrape triggered. The per-user rate-limit
+    only fires when an actual scrape is launched.
     """
     app = get_settings()
-    retry_after = await acquire_sold_scrape_slot(user.id, app.ebay_sold_scrape_min_interval_seconds)
-    if retry_after > 0:
-        iv = app.ebay_sold_scrape_min_interval_seconds
-        raise HTTPException(
-            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail=(
-                f"Limite : une recherche « vendus eBay » toutes les {iv:g} s "
-                f"(réessayez dans {retry_after} s)."
-            ),
-            headers={"Retry-After": str(retry_after)},
-        )
-    items, err = await scrape_sold_listings(
-        q=q.strip(),
-        window_hours=window_hours,
-        limit=scrape_limit,
-        pages=pages,
+    job = submit_job(
+        user_id=user.id,
+        q=body.q.strip(),
+        window_hours=body.window_hours,
+        pages=body.pages,
+        scrape_limit=body.scrape_limit,
+        top_limit=body.top_limit,
+        min_count=body.min_count,
         app=app,
     )
-    grouped = aggregate_top_sold(items, min_count=min_count, limit_per_category=top_limit)
+
+    cache_hit = job.status == "completed" and job.result is not None
+    if not cache_hit:
+        retry_after = await acquire_sold_scrape_slot(
+            user.id, app.ebay_sold_scrape_min_interval_seconds,
+        )
+        if retry_after > 0:
+            iv = app.ebay_sold_scrape_min_interval_seconds
+            raise HTTPException(
+                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                detail=(
+                    f"Rate limit: at most one eBay sold-search every {iv:g} s "
+                    f"(retry in {retry_after} s)."
+                ),
+                headers={"Retry-After": str(retry_after)},
+            )
+
     return {
-        "query": q.strip(),
-        "window_hours": window_hours,
-        "pages_requested": pages,
-        "total_observed": len(items),
-        "cards": grouped["cards"],
-        "graded": grouped["graded"],
-        "sealed": grouped["sealed"],
-        "groups_count": {
-            "cards": len(grouped["cards"]),
-            "graded": len(grouped["graded"]),
-            "sealed": len(grouped["sealed"]),
-        },
-        "error": err,
-        "ebay_sold_search_url": ebay_fr_sold_search_url(q=q.strip(), page_size=60),
-        "source": "ebay_html_scrape_aggregated",
+        **job.to_public(),
+        "ebay_sold_search_url": ebay_fr_sold_search_url(q=body.q.strip(), page_size=60),
+        "cached": cache_hit,
+    }
+
+
+@router.get("/sold-top/{job_id}", response_model=None)
+async def sold_top_status(
+    user: Annotated[User, Depends(get_current_user)],
+    job_id: str,
+) -> dict[str, Any]:
+    """
+    Return the current state of a ``sold-top`` job.
+
+    The client polls this endpoint while ``status`` is ``pending`` or
+    ``running``. Once ``completed`` (or ``failed``), ``result`` is populated
+    and polling can stop. A job may only be read by its creator.
+    """
+    job = get_job(job_id)
+    if job is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Unknown or expired job.",
+        )
+    if job.user_id != user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="This job does not belong to you.",
+        )
+    return {
+        **job.to_public(),
+        "ebay_sold_search_url": ebay_fr_sold_search_url(q=job.q, page_size=60),
     }

api/services/ebay_sold_scrape_service.py

@@ -19,7 +19,7 @@
 import time
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, Awaitable, Callable
 from urllib.parse import urlencode
 
 from bs4 import BeautifulSoup
@@ -477,6 +477,7 @@ async def scrape_sold_listings(
     limit: int = 50,
     pages: int = 1,
     app: AppSettings | None = None,
+    on_page_done: Callable[[int, int], Awaitable[None]] | None = None,
 ) -> tuple[list[dict[str, Any]], str | None]:
     """
     Return ``(items_as_dicts, error_message_or_none)``.
@@ -487,9 +488,13 @@ async def scrape_sold_listings(
     along with the matching error message. A small politeness pause sits
     between pages.
 
+    ``on_page_done`` is awaited after each successfully parsed page with
+    ``(page_num, total_unique_observed_so_far)`` — useful for surfacing
+    progress to a long-running job consumer.
+
     On success ``error_message_or_none`` is ``None``.
     """
-    pages_total = max(1, min(int(pages), 5))
+    pages_total = max(1, min(int(pages), 20))
     page_size = min(60, max(limit, 10))
 
     raw_rows: list[Any] = []
@@ -537,6 +542,12 @@ async def scrape_sold_listings(
             raw_rows.append(r)
             new_in_page += 1
 
+        if on_page_done is not None:
+            try:
+                await on_page_done(page_num, len(raw_rows))
+            except Exception:
+                logger.exception("on_page_done callback raised — ignoring")
+
         # eBay quietly stops returning new listings past the available pages —
         # break early once a page yields nothing new.
         if new_in_page == 0 and page_num > 1:

api/services/ebay_sold_top_service.py

@@ -54,9 +54,7 @@
 
 #: Strong sealed-product signals (matched on the diacritic-stripped lowercased
 #: title). Order does not matter — first hit classifies the listing.
-_SEALED_SIGNALS: tuple[str, ...] = (
-    "scelle",
-    "sealed",
+_SEALED_STRONG_SIGNALS: tuple[str, ...] = (
     "etb",
     "elite trainer",
     "trainer box",
@@ -83,6 +81,20 @@
     "pokebox",
 )
 
+#: Weak sealed hints often used for single cards in blister/sleeve.
+#: We only use these when no card-level hint is detected.
+_SEALED_WEAK_SIGNALS: tuple[str, ...] = (
+    "scelle",
+    "sealed",
+)
+
+#: Single-card set/promo codes commonly seen in French listings
+#: (e.g. "SWSH291", "SVP 052", "MEP031", "TG07").
+_CARD_CODE_RX = re.compile(
+    r"\b(?:svp|swsh|tg|gg|mep|sm|xy|bw|dp|hgss|sw|svp)\s*-?\s*\d{1,3}\b",
+    re.IGNORECASE,
+)
+
 _STOPWORDS: frozenset[str] = frozenset(
     {
         "pokemon",
@@ -175,28 +187,43 @@ def _significant_tokens(title_norm: str) -> list[str]:
     return out
 
 
-def _classify(title_norm: str, *, has_grade: bool, has_card_number: bool) -> Category:
+def _classify(
+    title_norm: str,
+    *,
+    has_grade: bool,
+    has_card_number: bool,
+    has_card_code: bool,
+) -> Category:
     """
     Decide if the listing is a graded card, a sealed product, or a raw card.
 
     Priority:
 
     1. ``has_grade`` — graded items always win, even if they happen to mention
        a sealed-product keyword (graded sealed boosters are ultra-rare).
-    2. ``has_card_number`` — a ``\\d+/\\d+`` pattern in the title denotes a
-       single card. Listings using the word « scellé(e) » in this case mean
-       *the card is sealed in plastic*, not « unopened booster product » — so
-       we never classify them as sealed.
-    3. Sealed-product keyword (ETB, display, blister, …) — sealed.
-    4. Default — raw card.
+    2. ``has_card_number`` or ``has_card_code`` — single-card hints (``12/102``,
+       ``SWSH291``, ``SVP052``, ``TG07``, …). Listings using « scellé(e) » in
+       this case mean *the card is sealed in plastic*, not unopened sealed
+       product — we keep them as cards.
+    3. Strong sealed-product keywords (ETB, display, booster box, coffret…) — sealed.
+    4. Weak sealed hints ("scellé"/"sealed"):
+       - if single-card hints are present (promo/code set), keep as card;
+       - otherwise sealed.
+    5. Default — raw card.
     """
     if has_grade:
         return "graded"
-    if has_card_number:
+    if has_card_number or has_card_code:
         return "cards"
-    for sig in _SEALED_SIGNALS:
+    for sig in _SEALED_STRONG_SIGNALS:
         if sig in title_norm:
             return "sealed"
+    has_weak_sealed = any(sig in title_norm for sig in _SEALED_WEAK_SIGNALS)
+    if has_weak_sealed:
+        has_card_hint = "promo" in title_norm or bool(_CARD_CODE_RX.search(title_norm))
+        if has_card_hint:
+            return "cards"
+        return "sealed"
     return "cards"
 
 
@@ -213,10 +240,12 @@ def _build_fingerprint(title: str) -> tuple[str, str | None, Category]:
         norm = _GRADE_RX.sub(" ", norm)
 
     card_match = _CARD_NUMBER_RX.search(norm)
+    has_card_code = _CARD_CODE_RX.search(norm) is not None
     category = _classify(
         norm,
         has_grade=grade is not None,
         has_card_number=card_match is not None,
+        has_card_code=has_card_code,
     )
     tokens = _significant_tokens(norm)