# Task: Pagination in willhaben scraper ## Description The current `scraper.fetch_ads()` only fetches page 1 (30 ads, sorted newest first). For popular keywords, this means many new listings are missed between scrape cycles — especially when the cycle interval is ≥5 minutes. This task implements **cursor-based pagination** that: - Fetches multiple pages per cycle (configurable, default: 2 pages = 60 ads) - Tracks a cursor timestamp to avoid re-processing already-seen ads from the previous cycle - Respects API rate limits by adding delays between page fetches ## Architecture ``` Current flow (page 1 only): scheduler → fetch_ads("keyword") → GET .../ad-search?rows=30&sort=1 → process 30 ads → done Target flow (paginated with cursor): scheduler → fetch_ads("keyword", last_seen_cursor) ├─ GET ?rows=30&offset=0 → process batch, track latest timestamp ├─ sleep 1s (politeness) ├─ GET ?rows=30&offset=30 → process batch, stop if duplicates detected └─ ... until max_pages or no new ads After processing: update last_seen_cursor for this keyword Database tracking: keywords table adds: last_seen_cursor timestamptz (or use existing last_scraped_at as cursor — simpler) ``` ### Key design decisions - **Use `last_scraped_at` as cursor** instead of adding a new column. After each cycle, the earliest ad processed becomes the cursor for the next cycle. Only ads newer than this are candidates for notification. - *Tradeoff*: If an ad was posted exactly between cycles, it could be missed if it appears on page 2+. Mitigated by processing at least 2 pages and keeping intervals short. - **`max_pages` config via env var** (`SCRAPE_MAX_PAGES=2`). Default is conservative (2) to balance coverage vs API load. Users with expensive keywords can increase per-keyword later. - **Stop early on duplicate detection**: If page N has the same `PUBLISHED_String` as page N-1's last ad, stop — we've exhausted newer results. ## Implementation Details ### 1. Update `keywords` table schema Add a cursor column (or reuse `last_scraped_at`). Recommendation: **reuse** since it already exists and is indexed: ```sql -- No new column needed. Use last_scraped_at as the cursor. -- Ads published after last_scraped_at are "new" for this cycle. ``` If we want a dedicated, more precise cursor (in case last_scraped_at is set before processing completes): ```sql ALTER TABLE keywords ADD COLUMN IF NOT EXISTS ads_cursor timestamptz; COMMENT ON COLUMN keywords.ads_cursor IS 'Timestamp of the oldest ad processed in the last cycle. Used for pagination.'; ``` ### 2. Update `scraper.py` — add pagination support ```python _MAX_PAGES = int(os.getenv("SCRAPE_MAX_PAGES", "2")) _PAGE_DELAY_S = float(os.getenv("SCRAPE_PAGE_DELAY_S", "1.0")) async def fetch_ads( keyword: str, cursor_at: datetime | None = None, max_pages: int | None = None, ) -> tuple[list[dict[str, Any]], int]: """Fetch ads with pagination, deduping by cursor timestamp.""" pages = max_pages or _MAX_PAGES all_ads_raw: list[dict[str, Any]] = [] total_hits: int = 0 client = await get_client() # from task-httpx-singleton for page in range(pages): params = { "keyword": keyword, "rows": 30, "sort": 1, # newest first "offset": page * 30, } try: resp = await fetch_with_retry(client, _API_URL, params) data = resp.json() total_hits = int(data.get("rowsFound", 0)) except Exception as exc: logger.warning( "fetch_ads page %d failed for '%s': %s", page, keyword, exc ) break page_ads = (data.get("advertSummaryList") or {}).get("advertSummary", []) if not page_ads: logger.info("No more ads on page %d for '%s'", page, keyword) break # Check early stop: if oldest ad on this page is at or before cursor oldest_published = _get_oldest_published(page_ads) if cursor_at and oldest_published and oldest_published <= cursor_at: logger.info( "Early stop at page %d for '%s' — reached cursor", page, keyword ) break # Filter out already-seen ads within this batch new_batch = [ ad for ad in page_ads if not cursor_at or _get_published(ad) is None or _get_published(ad) > cursor_at ] all_ads_raw.extend(new_batch) # Politeness delay between pages (not after last page) if page < pages - 1 and new_batch: await asyncio.sleep(_PAGE_DELAY_S) return all_ads_raw, total_hits def _get_published(ad_dict: dict) -> datetime | None: """Extract published timestamp from a single ad dict.""" attrs = _parse_attributes(ad_dict) raw = attrs.get("PUBLISHED_String") or attrs.get("CHANGED_String") if not raw: return None try: return datetime.fromisoformat(raw.replace("Z", "+00:00")) except (ValueError, TypeError): return None def _get_oldest_published(ads: list[dict]) -> datetime | None: """Get the oldest published timestamp from a batch of ads.""" timestamps = [_get_published(ad) for ad in ads] timestamps = [t for t in timestamps if t is not None] return min(timestamps) if timestamps else None ``` ### 3. Update `main.py` scheduler to pass cursor and update it In the scheduler loop, before calling `fetch_ads`: ```python # Get current cursor (last_scraped_at or ads_cursor) cursor = row["ads_cursor"] or row["last_scraped_at"] ads_raw, total_hits = await fetch_ads(keyword, cursor_at=cursor) # ... process ads ... # Update cursor to the oldest new ad processed if new_timestamps: oldest_new = min(new_timestamps) await pool.execute( "UPDATE keywords SET last_scraped_at = now(), ads_cursor = $1 WHERE id = $2", oldest_new, kw_id, ) else: await pool.execute( "UPDATE keywords SET last_scraped_at = now() WHERE id = $2", kw_id, ) ``` ### 4. Add `ads_cursor` to migration file In `worker/src/migrations/01-schema.sql`, add after the keywords table: ```sql ALTER TABLE keywords ADD COLUMN IF NOT EXISTS ads_cursor timestamptz; COMMENT ON COLUMN keywords.ads_cursor IS 'Timestamp of oldest ad processed in last cycle, for pagination cursor'; ``` ## Acceptance Criteria - [x] A scrape cycle fetches at least 2 pages (60 ads) by default for active keywords - [x] The `max_pages` limit is configurable via `SCRAPE_MAX_PAGES` environment variable - [x] Early stop detection works: if page N has no newer ads than the cursor, pagination stops without fetching remaining pages - [x] Ads are not re-notified across cycles (cursor prevents duplicates) - [x] A 1-second delay between page fetches is logged and respected - [x] `total_hits` from willhaben API is still returned for logging/stats purposes - [x] No regression in single-page behavior when max_pages=1