willhaben-tracker/docs/phase-3/task-multi-marketplace.md

Task: Multi-marketplace abstraction layer

Description

Currently, scraper.py is tightly coupled to willhaben's API format and URL. Adding a second marketplace (e.g., Kleinanzeigen, Facebook Marketplace) would require extensive refactoring of the core logic — duplicating pagination, error handling, and notification code with subtle differences per source.

This task introduces an abstract base class for scrapers and a standardized ad schema, making it trivial to add new marketplaces by implementing only marketplace-specific parsing logic.

Architecture

┌───────────────────────────────────────┐
│  scraper.py (module)                  │
│                                       │
│  ┌───────────────────────────────────┐│
│  │ ScraperBase (ABC)                 ││
│  │                                   ││
│  │ Properties:                       ││
│  │   name        str                 ││
│  │   base_url    str                 ││
│  │   max_pages   int                 ││
│  │                                   ││
│  │ Abstract methods:                ││
│  │   build_query(url, params) → URL  ││
│  │   parse_page(html/json) → list    ││
│  │   normalize_ad(raw) → dict       ││
│  │                                   ││
│  │ Concrete methods (shared):        ││
│  │   fetch_ads(keyword, cursor)     ││
│  │   _fetch_with_retry(url)         ││
│  └───────────────────────────────────┘│
└──────┬────────────────────────────────┘
       │ inherits
       ▼
┌───────────────────────────────────────┐
│  scrapers/willhaben.py                │
│                                       │
│  class WillhabenScraper(ScraperBase): │
│    name = "willhaben"                 ││
│    base_url = ".../api/v1/ad-search" ││
│    build_query() → willhaben URL     ││
│    parse_page(json) → list           ││
│    normalize_ad(raw) → standard dict ││
└───────────────────────────────────────┘

Standard ad schema (dict):
{
    "id": str,            # marketplace-specific ID
    "marketplace": str,   # e.g. "willhaben"
    "title": str,         # ad title
    "price": int | None,  # price in cents
    "currency": str,      # e.g. "EUR"
    "url": str,           # full URL to the ad page
    "published_at": datetime | None,
    "location": {          │
        "city": str,       │
        "postcode": str | None,  │
        "region": str | None   │
    },                    │
    "attributes": dict     # marketplace-specific extras
}

Scheduler (in main.py):
  
  scrapers: list[ScraperBase] = [
      WillhabenScraper(),
      KleinanzeigenScraper(),  # future
  ]
  
  for scraper in scrapers:
      ads_raw, total_hits = await scraper.fetch_ads(keyword)
      # ... process with same pipeline (filters, notifications)

Key design decisions

• Abstract base class defines the contract. Concrete scrapers only implement what's different per marketplace — URL building, response parsing, and field normalization.
  • Alternative: Could use a plugin architecture with entry_points, but that adds significant complexity for what is currently expected to be ≤3 marketplaces.
• Standardized output schema ensures the downstream pipeline (filters, notifications) works identically regardless of source. Marketplace-specific fields are stored in attributes.
• Pagination logic lives in the base class. Most marketplaces use offset/limit pagination; the abstract method handles this generically. Special cases override _fetch_page().

Implementation Details

1. Create worker/src/scrapers/__init__.py

from .willhaben import WillhabenScraper

__all__ = ["WillhabenScraper"]


def get_scriper_by_name(name: str) -> "ScraperBase":
    """Factory function to instantiate scrapers by name."""
    registry = {
        "willhaben": WillhabenScraper,
    }
    
    cls = registry.get(name.lower())
    if not cls:
        raise ValueError(f"Unknown marketplace: {name}")
    
    return cls()

2. Create worker/src/scrapers/base.py (abstract base class)

import abc
import asyncio
import logging
from datetime import datetime, timezone
from typing import Any

logger = logging.getLogger(__name__)


class ScraperBase(abc.ABC):
    """Abstract base class for marketplace scrapers."""
    
    name: str = "unknown"
    base_url: str = ""
    max_pages: int = 2
    
    @abc.abstractmethod
    def build_query(self, keyword: str, offset: int) -> str:
        """Build the full API URL/endpoint for a keyword + offset."""
        ...
    
    @abc.abstractmethod
    def parse_page(self, response_content: Any) -> list[dict]:
        """Parse raw response into list of ad dicts (marketplace-specific)."""
        ...
    
    @abc.abstractmethod
    def normalize_ad(self, raw_ad: dict) -> dict:
        """Convert marketplace-specific format to standard schema."""
        ...
    
    async def fetch_ads(
        self,
        keyword: str,
        cursor_at: datetime | None = None,
        max_pages: int | None = None,
    ) -> tuple[list[dict], int]:
        """Fetch ads with pagination. Shared implementation."""
        pages = max_pages or self.max_pages
        all_ads: list[dict] = []
        total_hits = 0
        
        from ..scraper import get_client  # httpx singleton
        
        client = await get_client()
        
        for page in range(pages):
            url = self.build_query(keyword, offset=page * 30)
            
            try:
                content = await self._fetch_with_retry(client, url)
            except Exception as exc:
                logger.warning(
                    "%s: fetch failed at page %d for '%s': %s", 
                    self.name, page, keyword, exc
                )
                break
            
            raw_ads = self.parse_page(content)
            
            if not raw_ads:
                logger.info("%s: no more ads on page %d for '%s'", 
                           self.name, page, keyword)
                break
            
            # Normalize and filter by cursor
            normalized = []
            for raw in raw_ads:
                ad = self.normalize_ad(raw)
                if cursor_at and ad["published_at"] and ad["published_at"] <= cursor_at:
                    continue
                normalized.append(ad)
            
            all_ads.extend(normalized)
            
            # Politeness delay
            if page < pages - 1 and normalized:
                await asyncio.sleep(1.0)
        
        return all_ads, total_hits
    
    async def _fetch_with_retry(
        self, 
        client: Any, 
        url: str,
        max_retries: int = 3,
    ) -> Any:
        """Generic retry wrapper for HTTP fetches."""
        import httpx
        
        for attempt in range(max_retries):
            try:
                resp = await client.get(url)
                resp.raise_for_status()
                return resp.json() if "application/json" in (resp.headers.get("content-type") or "") else resp.text
            except httpx.ConnectError as exc:
                logger.warning("%s: transport error attempt %d: %s", 
                             self.name, attempt + 1, exc)
                if attempt < max_retries - 1:
                    await asyncio.sleep(2 ** attempt)
                else:
                    raise
    
    @property
    def headers(self) -> dict[str, str]:
        """HTTP headers for requests. Override per marketplace."""
        return {}

3. Create worker/src/scrapers/willhaben.py (refactor existing logic)

Move the current willhaben-specific code from scraper.py into this implementation:

import logging
from datetime import datetime, timezone
from typing import Any

from .base import ScraperBase

logger = logging.getLogger(__name__)


class WillhabenScraper(ScraperBase):
    name = "willhaben"
    base_url = "https://api.willhaben.at/external/api/v1/ad-search"
    
    @property
    def headers(self) -> dict[str, str]:
        return {
            "Accept": "application/json",
            "User-Agent": "Mozilla/5.0 (compatible; WillhabenTracker/1.0)",
        }
    
    def build_query(self, keyword: str, offset: int = 0) -> str:
        """Build willhaben API URL with keyword + pagination."""
        import urllib.parse
        
        params = {
            "keyword": keyword,
            "rows": 30,
            "sort": 1,  # newest first
            "offset": offset,
        }
        
        return f"{self.base_url}?{urllib.parse.urlencode(params)}"
    
    def parse_page(self, response_content: dict) -> list[dict]:
        """Parse willhaben JSON response into raw ad dicts."""
        ads_list = (response_content.get("advertSummaryList") or {}).get(
            "advertSummary", []
        )
        
        total_hits = int(response_content.get("rowsFound", 0))
        
        return ads_list
    
    def normalize_ad(self, raw_ad: dict) -> dict:
        """Convert willhaben ad format to standard schema."""
        # Extract attributes from the nested format
        attrs = self._parse_attributes(raw_ad)
        
        # Extract ID
        ad_id_raw = raw_ad.get("id", "")
        
        # Extract title (handle various formats)
        title_raw = raw_ad.get("title") or raw_ad.get("Title", {})
        title = title_raw.get("Value", title_raw) if isinstance(title_raw, dict) else str(title_raw or "")
        
        # Extract price
        price_str = attrs.get("PRICE_String") or attrs.get("priceString", "")
        try:
            price_cents = int(float(price_str.replace(".", ""))) if price_str else None
        except (ValueError, TypeError):
            price_cents = None
        
        # Extract published date
        pub_str = attrs.get("PUBLISHED_String") or attrs.get("publishedString", "")
        published_at = None
        try:
            published_at = datetime.fromisoformat(pub_str.replace("Z", "+00:00"))
        except (ValueError, TypeError):
            pass
        
        # Extract location
        city = attrs.get("LOCATION_CityName") or ""
        postcode = attrs.get("LOCATION_ZIP") or ""
        region = attrs.get("LOCATION_Region") or ""
        
        return {
            "id": ad_id_raw,
            "marketplace": self.name,
            "title": title.strip(),
            "price": price_cents,
            "currency": "EUR",
            "url": raw_ad.get("linkUrl", ""),
            "published_at": published_at,
            "location": {
                "city": city,
                "postcode": postcode if postcode else None,
                "region": region if region else None,
            },
            "attributes": attrs,  # preserve marketplace-specific fields
        }
    
    @staticmethod
    def _parse_attributes(ad_dict: dict) -> dict:
        """Parse willhaben's nested attribute format into flat dict."""
        result = {}
        
        for attr_group in ad_dict.get("attributes", []):
            if not isinstance(attr_group, dict):
                continue
            
            group_name = attr_group.get("name") or ""
            
            for item in attr_group.get("items", []):
                key = f"{group_name}_{item['name']}" if group_name else item["name"]
                result[key] = item.get("valueString", "")
        
        return result

4. Update main.py scheduler to use the new scraper factory

Replace direct calls to fetch_ads(keyword) with:

from scrapers import get_scriper_by_name

# At startup:
scrapers_config = os.getenv("SCRAPERS", "willhaben").split(",")
active_scrapers = [get_scriper_by_name(s) for s in scrapers_config]

# In scheduler loop:
for scraper in active_scrapers:
    ads_raw, total_hits = await scraper.fetch_ads(keyword, cursor_at=cursor)
    
    # ... process with existing pipeline (price filter, postcode filter, etc.)

5. Add .env.example configuration for scrapers

# Marketplace sources to scrape (comma-separated)
SCRAPERS=willhaben

Acceptance Criteria

• WillhabenScraper produces identical output to the current scraper.py implementation (no regression in ad extraction)
• Adding a new marketplace requires only: creating one file under scrapers/, registering it in __init__.py, and listing it in SCRAPERS env var
• The standardized ad schema includes all fields needed by the downstream pipeline (price, postcode, published_at, URL)
• Pagination logic works correctly through the base class for willhaben
• Error handling (retries, timeouts) continues to work with the new abstraction
• All existing bot commands and notifications function identically after refactoring