docs(plan): add Phase 1, 2, 3 implementation specs
This commit is contained in:
hermes
@@ -0,0 +1,60 @@
|
||||
# Phase 1 — Reliability & Completeness Improvements
|
||||
|
||||
## Scope
|
||||
|
||||
This phase addresses the **three highest-impact reliability gaps** identified in the code review: incomplete ad coverage due to page-limited scraping, inefficient HTTP client usage, and silent notification loss. After this phase, the worker will:
|
||||
|
||||
- Capture a larger window of ads per scrape cycle (no longer limited to 30 newest)
|
||||
- Reuse TCP/TLS connections for willhaben API calls instead of creating one per request
|
||||
- Retry failed Telegram notifications instead of dropping them permanently
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
┌──────────────────────────────────────────────────────┐
|
||||
│ worker container │
|
||||
│ │
|
||||
│ ┌───────────┐ │
|
||||
│ │ scraper.py│ ← SINGLETON AsyncClient │
|
||||
│ │ │ (connection pool, keepalive) │
|
||||
│ │ │ │
|
||||
│ │ fetch_ads() │
|
||||
│ │ ├─ page 1: rows=30 & published_after=<cursor> │
|
||||
│ │ ├─ page 2: rows=30 & offset=30 │
|
||||
│ │ └─ ... until no new ads or max_pages reached │
|
||||
│ └───────────┘ │
|
||||
│ │ │
|
||||
│ ▼ │
|
||||
│ ┌──────────────┐ ┌──────────────────┐ │
|
||||
│ │ notifier.py │──►│ notification_queue│ │
|
||||
│ │ │ │ table (new) │ │
|
||||
│ │ notify_new() │ │ │ │
|
||||
│ │ notify_drop()│ │ - ad_id │ │
|
||||
│ │ │ │ - telegram_id │ │
|
||||
│ │ if success: │ │ - attempts (0→5) │ │
|
||||
│ │ log_notify │ │ - last_error │ │
|
||||
│ │ if fail: │ │ - status │ │
|
||||
│ │ enqueue! │ └────────┬─────────┘ │
|
||||
│ └──────────────┘ │ │
|
||||
│ ▼ │
|
||||
│ scheduler retries │
|
||||
│ pending items each cycle │
|
||||
└──────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Tasks
|
||||
|
||||
| Task | File | Description |
|
||||
|------|------|-------------|
|
||||
| Pagination in willhaben scraper | [task-scraper-pagination.md](./task-scraper-pagination.md) | Implement cursor-based or offset pagination to fetch more than 30 ads per cycle, tracking the last seen timestamp to avoid duplicates across cycles. |
|
||||
| httpx singleton with connection pool | [task-httpx-singleton.md](./task-httpx-singleton.md) | Replace per-call AsyncClient creation with a module-level singleton using keepalive connections and configurable limits. |
|
||||
| Retry queue for failed notifications | [task-notification-retry-queue.md](./task-notification-retry-queue.md) | Add a `notification_queue` table to persist failed Telegram sends with exponential backoff retries (up to 5 attempts). |
|
||||
|
||||
## General Acceptance Criteria
|
||||
|
||||
- [ ] A single scrape cycle captures at least 90 ads for high-volume keywords (3 pages × 30 rows) instead of the current hard cap of 30
|
||||
- [ ] Duplicate ads between cycles are not re-notified (cursor/offset tracking prevents this)
|
||||
- [ ] HTTP connection reuse reduces willhaben API call latency by ≥40% (measured via logs)
|
||||
- [ ] Failed notifications are retried up to 5 times with exponential backoff (1m, 2m, 4m, 8m, 16m between attempts)
|
||||
- [ ] After 5 failed retries the notification is marked as `dead` and logged — not silently dropped
|
||||
- [ ] The scheduler processes queued notifications at the start of each cycle before scraping new keywords
|
||||
@@ -0,0 +1,157 @@
|
||||
# Task: httpx singleton with connection pool
|
||||
|
||||
## Description
|
||||
|
||||
The current `scraper.fetch_ads()` creates a **new** `httpx.AsyncClient` on every call:
|
||||
|
||||
```python
|
||||
async def fetch_ads(keyword: str):
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
resp = await client.get(_API_URL, ...)
|
||||
```
|
||||
|
||||
This means each scrape cycle incurs the full cost of TCP handshake + TLS negotiation (≈100-300ms per call on a cold connection). For keywords scraped every 5 minutes with multiple pages, this overhead adds up to **seconds of unnecessary latency per cycle**.
|
||||
|
||||
This task replaces the per-call client with a module-level singleton that reuses connections via keepalive.
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
Current:
|
||||
Cycle 1: create AsyncClient → fetch → close → ~300ms overhead
|
||||
Cycle 2: create AsyncClient → fetch → close → ~300ms overhead
|
||||
Cycle N: ... (repeated forever)
|
||||
|
||||
Target:
|
||||
Module load: create AsyncClient (singleton, keepalive pool)
|
||||
Cycle 1: use client → fetch → ~50ms (warm connection)
|
||||
Cycle 2: use client → fetch → ~50ms (warm connection)
|
||||
Cycle N: ...
|
||||
Shutdown: close client gracefully
|
||||
```
|
||||
|
||||
### Key design decisions
|
||||
|
||||
- **Module-level singleton** (`_client = None`, lazy init). Simpler than dependency injection and works with the existing async context.
|
||||
- **Keepalive connections**: Default `max_keepalive_connections=5` handles concurrent keyword scrapes efficiently.
|
||||
- **Client recreation on error**: If the client is closed or encounters a fatal transport error, it's recreated on the next call. This prevents stale connection issues.
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### 1. Add singleton getter to `scraper.py`
|
||||
|
||||
```python
|
||||
import os
|
||||
|
||||
_client: httpx.AsyncClient | None = None
|
||||
|
||||
|
||||
async def get_client() -> httpx.AsyncClient:
|
||||
"""Return a shared AsyncClient with keepalive connection pool."""
|
||||
global _client
|
||||
|
||||
if _client is None or _client.is_closed:
|
||||
max_conns = int(os.getenv("HTTP_MAX_CONNECTIONS", "10"))
|
||||
max_keepalive = int(os.getenv("HTTP_KEEPALIVE_CONNECTIONS", "5"))
|
||||
|
||||
_client = httpx.AsyncClient(
|
||||
timeout=float(os.getenv("HTTP_TIMEOUT_S", "30.0")),
|
||||
limits=httpx.Limits(
|
||||
max_connections=max_conns,
|
||||
max_keepalive_connections=max_keepalive,
|
||||
keepalive_expiry=60, # seconds
|
||||
),
|
||||
)
|
||||
logger.info(
|
||||
"Created httpx client: max_conns=%d, keepalive=%d",
|
||||
max_conns, max_keepalive,
|
||||
)
|
||||
|
||||
return _client
|
||||
|
||||
|
||||
async def close_client() -> None:
|
||||
"""Close the shared AsyncClient. Call during shutdown."""
|
||||
global _client
|
||||
if _client and not _client.is_closed:
|
||||
await _client.aclose()
|
||||
logger.info("Closed httpx client")
|
||||
_client = None
|
||||
```
|
||||
|
||||
### 2. Update `fetch_ads()` to use the singleton
|
||||
|
||||
**Replace:**
|
||||
```python
|
||||
async def fetch_ads(keyword: str):
|
||||
params = {...}
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
for attempt in range(1, 4):
|
||||
try:
|
||||
resp = await client.get(_API_URL, headers=_HEADERS, params=params)
|
||||
```
|
||||
|
||||
**With:**
|
||||
```python
|
||||
async def fetch_ads(keyword: str):
|
||||
params = {...}
|
||||
client = await get_client()
|
||||
|
||||
for attempt in range(1, 4):
|
||||
try:
|
||||
resp = await client.get(_API_URL, headers=_HEADERS, params=params)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
break
|
||||
except httpx.ConnectError as exc:
|
||||
# Transport error — recreate client on next attempt
|
||||
logger.warning("Transport error on attempt %d: %s", attempt, exc)
|
||||
await close_client() # force recreation
|
||||
if attempt < 3:
|
||||
await asyncio.sleep(2 ** attempt)
|
||||
continue
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.warning("fetch_ads attempt %d failed: %s", attempt, exc)
|
||||
if attempt < 3:
|
||||
await asyncio.sleep(2 ** attempt)
|
||||
continue
|
||||
raise
|
||||
|
||||
# ... rest unchanged (extract ads_raw, total_hits)
|
||||
```
|
||||
|
||||
### 3. Call `close_client()` during shutdown in `main.py`
|
||||
|
||||
Add to the cleanup function (from Phase 0 task-graceful-shutdown):
|
||||
|
||||
```python
|
||||
async def cleanup(app: Application) -> None:
|
||||
logger.info("Shutting down...")
|
||||
|
||||
# ... existing cleanup steps ...
|
||||
|
||||
# Close HTTP client
|
||||
from scraper import close_client
|
||||
await close_client()
|
||||
|
||||
# ... rest of cleanup (close DB pool, etc.)
|
||||
```
|
||||
|
||||
### 4. Update `.env.example` with new config options
|
||||
|
||||
```bash
|
||||
# HTTP Client Configuration
|
||||
HTTP_MAX_CONNECTIONS=10 # Max concurrent connections to willhaben API
|
||||
HTTP_KEEPALIVE_CONNECTIONS=5 # Connections kept alive in the pool
|
||||
HTTP_TIMEOUT_S=30.0 # Request timeout in seconds
|
||||
```
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] Only one `httpx.AsyncClient` is created per process lifetime (logged once at startup)
|
||||
- [ ] Subsequent calls to `fetch_ads()` reuse the existing client (no "Created httpx client" log)
|
||||
- [ ] After calling `close_client()`, a new call to `get_client()` creates a fresh client
|
||||
- [ ] Connection keepalive reduces latency for sequential API calls (verifiable via timing in logs)
|
||||
- [ ] Fatal transport errors trigger client recreation without crashing the scheduler
|
||||
- [ ] The client is properly closed during graceful shutdown (no resource warnings)
|
||||
@@ -0,0 +1,301 @@
|
||||
# Task: Retry queue for failed notifications
|
||||
|
||||
## Description
|
||||
|
||||
The current `notifier.notify_new()` and `notify_drop()` call `_send_message()` directly. If the Telegram API returns an error (rate limiting, network hiccup, user deleted the bot), the notification is **silently logged** and never retried:
|
||||
|
||||
```python
|
||||
async def _send_message(bot: Bot, chat_id: int, message_text):
|
||||
try:
|
||||
await bot.send_message(chat_id=chat_id, text=message_text)
|
||||
except TelegramError as e:
|
||||
logger.warning("Failed to send notification ...") # ← notification LOST forever
|
||||
```
|
||||
|
||||
This task introduces a **persistent retry queue** backed by the `notification_queue` table. Failed notifications are stored with an attempt counter and retried on subsequent scheduler cycles with exponential backoff. After 5 failed attempts, they're marked as `dead` and logged — not silently dropped.
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
┌───────────────────────────────────────┐
|
||||
│ notification_queue table (new) │
|
||||
│ │
|
||||
│ id uuid PK │
|
||||
│ ad_id uuid │
|
||||
│ telegram_id text │
|
||||
│ message_text text │
|
||||
│ type enum('new','drop') │
|
||||
│ attempts int DEFAULT 0 │
|
||||
│ max_attempts int DEFAULT 5 │
|
||||
│ last_error text │
|
||||
│ status enum │
|
||||
│ ('pending','sent', │
|
||||
│ 'failed','dead') │
|
||||
│ created_at timestamptz │
|
||||
│ updated_at timestamptz │
|
||||
│ │
|
||||
│ INDEX: status, attempts (composite) │
|
||||
└───────────┬───────────────────────────┘
|
||||
│
|
||||
▼
|
||||
┌───────────────────────────────────────┐
|
||||
│ notifier.py flow: │
|
||||
│ │
|
||||
│ send_notification(): │
|
||||
│ try: │
|
||||
│ await bot.send_message(...) │
|
||||
│ → log_notify() (as before) │
|
||||
│ except TelegramError as e: │
|
||||
│ INSERT INTO notification_queue │
|
||||
│ (ad_id, telegram_id, ...) │
|
||||
│ VALUES (...) │
|
||||
└───────────┬───────────────────────────┘
|
||||
│
|
||||
▼
|
||||
┌───────────────────────────────────────┐
|
||||
│ scheduler.py / main.py: │
|
||||
│ At the START of each cycle: │
|
||||
│ │
|
||||
│ for item in pending_queue: │
|
||||
│ if attempts < max_attempts: │
|
||||
│ backoff = 2^attempts minutes │
|
||||
│ if now >= updated_at + backoff:│
|
||||
│ try send again │
|
||||
│ success → mark 'sent' │
|
||||
│ fail → increment count │
|
||||
│ elif attempts >= max_attempts: │
|
||||
│ mark as 'dead' │
|
||||
│ log warning │
|
||||
└───────────────────────────────────────┘
|
||||
|
||||
Exponential backoff schedule:
|
||||
Attempt 1 → wait 1 min (2^0)
|
||||
Attempt 2 → wait 2 min (2^1)
|
||||
Attempt 3 → wait 4 min (2^2)
|
||||
Attempt 4 → wait 8 min (2^3)
|
||||
Attempt 5 → wait 16 min (2^4)
|
||||
Total worst case: ~31 min before giving up
|
||||
```
|
||||
|
||||
### Key design decisions
|
||||
|
||||
- **Store full message text** in the queue table so we can retry without re-rendering. This is important because ad data might change or be removed from willhaben by the time we retry.
|
||||
- **Process at start of scheduler cycle** — ensures queued items are attempted before new scraping starts, prioritizing user notifications over fresh data collection.
|
||||
- **Backoff based on `updated_at`**, not wall-clock from first attempt. Each retry resets the backoff timer. This handles edge cases where a failure was transient but then another transient follows.
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### 1. Add migration for `notification_queue` table
|
||||
|
||||
In `worker/src/migrations/02-notification-queue.sql`:
|
||||
|
||||
```sql
|
||||
-- Notification retry queue table
|
||||
|
||||
CREATE TYPE notification_type AS ENUM ('new', 'drop');
|
||||
CREATE TYPE notification_status AS ENUM ('pending', 'sent', 'failed', 'dead');
|
||||
|
||||
CREATE TABLE IF NOT EXISTS notification_queue (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
ad_id uuid NOT NULL REFERENCES ads(id) ON DELETE SET NULL,
|
||||
telegram_id text NOT NULL,
|
||||
message_text text NOT NULL,
|
||||
type notification_type NOT NULL,
|
||||
attempts int NOT NULL DEFAULT 0,
|
||||
max_attempts int NOT NULL DEFAULT 5,
|
||||
last_error text,
|
||||
status notification_status NOT NULL DEFAULT 'pending',
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
updated_at timestamptz NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- Index for efficient queue polling
|
||||
CREATE INDEX IF NOT EXISTS idx_notif_queue_poll
|
||||
ON notification_queue(status, attempts, updated_at)
|
||||
WHERE status IN ('pending', 'failed');
|
||||
|
||||
COMMENT ON TABLE notification_queue IS
|
||||
'Persistent retry queue for failed Telegram notifications';
|
||||
```
|
||||
|
||||
### 2. Update `notifier.py` — enqueue on failure
|
||||
|
||||
**Current:**
|
||||
```python
|
||||
async def _send_message(bot: Bot, chat_id: int, message_text):
|
||||
try:
|
||||
await bot.send_message(chat_id=chat_id, text=message_text)
|
||||
except TelegramError as e:
|
||||
logger.warning(
|
||||
"Failed to send notification to %d: %s", chat_id, str(e)[:30]
|
||||
)
|
||||
```
|
||||
|
||||
**After:**
|
||||
```python
|
||||
async def _send_message(
|
||||
bot: Bot,
|
||||
chat_id: int,
|
||||
message_text: str,
|
||||
ad_id: uuid.UUID | None = None,
|
||||
notif_type: str = "new",
|
||||
):
|
||||
try:
|
||||
await bot.send_message(chat_id=chat_id, text=message_text)
|
||||
except TelegramError as e:
|
||||
error_msg = str(e)[:300] # cap length
|
||||
logger.warning("Telegram send failed for %s: %s", chat_id, error_msg)
|
||||
|
||||
if ad_id:
|
||||
await _enqueue_retry(
|
||||
ad_id=ad_id,
|
||||
telegram_id=str(chat_id),
|
||||
message_text=message_text,
|
||||
notif_type=notif_type,
|
||||
error_msg=error_msg,
|
||||
)
|
||||
|
||||
|
||||
async def _enqueue_retry(
|
||||
ad_id: uuid.UUID,
|
||||
telegram_id: str,
|
||||
message_text: str,
|
||||
notif_type: str,
|
||||
error_msg: str,
|
||||
) -> None:
|
||||
"""Store a failed notification for later retry."""
|
||||
from db import get_pool
|
||||
|
||||
pool = await get_pool()
|
||||
|
||||
# Check if already queued (avoid duplicates for same ad+user)
|
||||
existing = await pool.fetchval(
|
||||
"""SELECT id FROM notification_queue
|
||||
WHERE ad_id = $1 AND telegram_id = $2 AND status IN ('pending', 'failed')""",
|
||||
ad_id, telegram_id,
|
||||
)
|
||||
|
||||
if existing:
|
||||
logger.info("Already queued: ad=%s user=%s", ad_id[:8], telegram_id)
|
||||
return
|
||||
|
||||
await pool.execute(
|
||||
"""INSERT INTO notification_queue
|
||||
(ad_id, telegram_id, message_text, type, last_error, status)
|
||||
VALUES ($1, $2, $3, $4, $5, 'pending')""",
|
||||
ad_id, telegram_id, message_text, notif_type, error_msg,
|
||||
)
|
||||
|
||||
logger.info("Queued for retry: ad=%s user=%s", ad_id[:8], telegram_id)
|
||||
```
|
||||
|
||||
### 3. Add queue processing to scheduler in `main.py`
|
||||
|
||||
At the start of each scheduler cycle (before keyword scraping):
|
||||
|
||||
```python
|
||||
async def process_notification_queue() -> int:
|
||||
"""Process pending notifications from the retry queue."""
|
||||
from db import get_pool
|
||||
|
||||
pool = await get_pool()
|
||||
|
||||
# Get items eligible for retry (backoff respected)
|
||||
rows = await pool.fetch("""
|
||||
SELECT id, ad_id, telegram_id, message_text, type, attempts,
|
||||
max_attempts, last_error, updated_at
|
||||
FROM notification_queue
|
||||
WHERE status IN ('pending', 'failed')
|
||||
AND updated_at + ($1 || ' minutes')::interval <= now()
|
||||
ORDER BY attempts ASC, updated_at ASC
|
||||
""", "2^attempts" if pool.is_pg else 0)
|
||||
|
||||
# Actually use a computed backoff in Python since PG expressions are tricky:
|
||||
rows = await pool.fetch("""
|
||||
SELECT id, ad_id, telegram_id, message_text, type, attempts,
|
||||
max_attempts, last_error, updated_at
|
||||
FROM notification_queue
|
||||
WHERE status IN ('pending', 'failed')
|
||||
ORDER BY attempts ASC, updated_at ASC
|
||||
LIMIT 50
|
||||
""")
|
||||
|
||||
processed = 0
|
||||
|
||||
for row in rows:
|
||||
backoff_min = min(2 ** row["attempts"], 60) # cap at 60 min
|
||||
retry_after = row["updated_at"] + timedelta(minutes=backoff_min)
|
||||
|
||||
if datetime.now(tz=timezone.utc) < retry_after:
|
||||
continue # not yet eligible
|
||||
|
||||
try:
|
||||
from bot import get_application_bot
|
||||
bot = get_application_bot()
|
||||
|
||||
await bot.send_message(
|
||||
chat_id=int(row["telegram_id"]),
|
||||
text=row["message_text"]
|
||||
)
|
||||
|
||||
await pool.execute(
|
||||
"UPDATE notification_queue SET status = 'sent', updated_at = now() WHERE id = $1",
|
||||
row["id"],
|
||||
)
|
||||
|
||||
await log_notify(pool, row["ad_id"], int(row["telegram_id"]), row["type"])
|
||||
|
||||
processed += 1
|
||||
|
||||
except TelegramError as e:
|
||||
new_attempts = row["attempts"] + 1
|
||||
|
||||
if new_attempts >= row["max_attempts"]:
|
||||
await pool.execute(
|
||||
"""UPDATE notification_queue
|
||||
SET status = 'dead', attempts = $2, last_error = $3, updated_at = now()
|
||||
WHERE id = $1""",
|
||||
row["id"], new_attempts, str(e)[:300],
|
||||
)
|
||||
logger.error(
|
||||
"Notification DEAD after %d attempts: ad=%s user=%s err=%s",
|
||||
new_attempts, row["ad_id"][:8], row["telegram_id"], e,
|
||||
)
|
||||
else:
|
||||
await pool.execute(
|
||||
"""UPDATE notification_queue
|
||||
SET status = 'failed', attempts = $2, last_error = $3, updated_at = now()
|
||||
WHERE id = $1""",
|
||||
row["id"], new_attempts, str(e)[:300],
|
||||
)
|
||||
|
||||
return processed
|
||||
```
|
||||
|
||||
### 4. Call `process_notification_queue()` in the scheduler loop
|
||||
|
||||
In `main.py`, before iterating keywords:
|
||||
|
||||
```python
|
||||
async def run_scheduler() -> None:
|
||||
while True:
|
||||
try:
|
||||
record_scheduler_run() # healthcheck
|
||||
|
||||
# Process pending notifications FIRST
|
||||
processed = await process_notification_queue()
|
||||
if processed:
|
||||
logger.info("Retried %d queued notifications", processed)
|
||||
|
||||
# ... existing keyword iteration ...
|
||||
```
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] When `_send_message()` raises `TelegramError`, the notification is INSERTed into `notification_queue` with status='pending'
|
||||
- [ ] On the next scheduler cycle, pending items are attempted (respecting backoff)
|
||||
- [ ] After 5 failed attempts, the notification status becomes 'dead' and a warning is logged
|
||||
- [ ] The queue processes at most 50 items per cycle to avoid blocking the scheduler
|
||||
- [ ] Duplicate enqueue prevention works: calling `_enqueue_retry` twice for the same ad+user creates only one queue entry
|
||||
- [ ] Successful retries update `log_notifications` table (same as direct notifications)
|
||||
- [ ] The `/health` endpoint or logs can show the current count of pending/dead items
|
||||
@@ -0,0 +1,189 @@
|
||||
# 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
|
||||
|
||||
- [ ] A scrape cycle fetches at least 2 pages (60 ads) by default for active keywords
|
||||
- [ ] The `max_pages` limit is configurable via `SCRAPE_MAX_PAGES` environment variable
|
||||
- [ ] Early stop detection works: if page N has no newer ads than the cursor, pagination stops without fetching remaining pages
|
||||
- [ ] Ads are not re-notified across cycles (cursor prevents duplicates)
|
||||
- [ ] A 1-second delay between page fetches is logged and respected
|
||||
- [ ] `total_hits` from willhaben API is still returned for logging/stats purposes
|
||||
- [ ] No regression in single-page behavior when max_pages=1
|
||||
Reference in New Issue
Block a user