feat: Phase 3 — web dashboard, testing, and CI/CD
CI / lint-and-test (push) Has been cancelled

- Remove multi-marketplace from Phase 3
- Add FastAPI web UI on port 8766 with basic auth
- Add 6 Jinja2 templates (dashboard, keywords, users, ads, stats)
- Add pytest test suite (45 tests, 49% coverage)
- Add GitHub Actions CI/CD workflow
- Update docker-compose.yml to expose web UI port
- Update Dockerfile to include tests
This commit is contained in:
2026-07-10 22:28:18 +02:00
parent 0c20799f9a
commit 8151c530da
26 changed files with 1506 additions and 388 deletions
+28 -23
View File
@@ -1,12 +1,12 @@
# Phase 3 — Scalability & Advanced Features
# Phase 3 — Web Dashboard & Testing
## Scope
This phase introduces **structural improvements** that make the project maintainable, extensible, and testable. Currently, the entire system is a single async Python process with no tests and no CI/CD pipeline. After this phase:
This phase introduces **observability and reliability** improvements. Currently, the entire system is a single async Python process with no tests, no CI/CD pipeline, and no way to monitor what's happening without SSH-ing into the server. After this phase:
- A **Web Dashboard** provides real-time visibility into keywords, ads, users, and stats
- Automated tests provide confidence for every change (≥80% coverage)
- CI/CD pipeline runs on every push to validate code quality
- Multi-marketplace architecture enables adding new sources without modifying core logic
## Architecture
@@ -21,18 +21,17 @@ This phase introduces **structural improvements** that make the project maintain
│ │ │ ├── db.py (asyncpg pool mgmt) │
│ │ │ ├── bot.py (Telegram handlers) │
│ │ │ ├── notifier.py (message sending) │
│ │ │ ├── scraper.py (base scraper class) │
│ │ │ ├── scrapers/
│ │ │ │ ├── __init__.py │
│ │ │ │ ├── willhaben.py (willhaben-specific) │
│ │ │ │ └── base.py (abstract base class) │
│ │ │ ├── scraper.py (willhaben scraper)
│ │ │ ├── web.py (FastAPI dashboard)
│ │ │ ├── health.py (healthcheck endpoint) │
│ │ │ ── migrate.py (migration runner) │
│ │ │ ── migrate.py (migration runner) │
│ │ │ └── templates/ (Jinja2 HTML templates) │
│ │ ├── tests/ │
│ │ │ ├── conftest.py │
│ │ │ ├── test_scraper.py │
│ │ │ ├── test_notifier.py │
│ │ │ ── ...
│ │ │ ── test_filters.py
│ │ │ └── test_web.py │
│ │ ├── Dockerfile │
│ │ └── requirements.txt │
│ ├── .github/ │
@@ -42,17 +41,17 @@ This phase introduces **structural improvements** that make the project maintain
│ └── docker-compose.yml │
└──────────────────────────────────────────────────────┘
Multi-marketplace abstraction:
Web Dashboard (FastAPI, port 8766):
ScraperBase (abstract):
- async fetch_ads(keyword) → list[dict]
- async parse_response(html/json) → list[dict]
- normalize_ad(raw) → dict with standard keys
WillhabenScraper(ScraperBase):
- implements willhaben-specific URL, headers, parsing
GET / → Dashboard (keywords overview, stats summary)
GET /keywords → Keywords list with status, filters, subscribers
GET /keywords/<id> → Keyword detail (recent ads, price history, scrape logs)
GET /users → Users list with settings
GET /ads → Recent ads with search/filter
GET /stats → JSON stats (extends existing /stats endpoint)
Future: KleinAnzeigenScraper, MobileScraper, ...
Auth: Basic Auth via WEB_UI_USERNAME / WEB_UI_PASSWORD env vars
Templates: Jinja2 with inline CSS (zero external dependencies)
CI/CD Pipeline (.github/workflows/ci.yml):
@@ -71,6 +70,8 @@ Tests Structure:
- test_scraper_pagination() — verify pagination logic with mock responses
- test_price_filters() — verify filter functions
- test_notification_retry() — verify retry queue behavior
- test_mute_digest() — verify mute hours and digest buffering
- test_web_endpoints() — verify web UI routes
Integration tests:
- Test against real willhaben API (rate-limited, cached)
@@ -81,13 +82,17 @@ Tests Structure:
| Task | File | Description |
|------|------|-------------|
| Multi-marketplace abstraction layer | [task-multi-marketplace.md](./task-multi-marketplace.md) | Refactor `scraper.py` into a base class + per-marketplace implementations. Introduces a standard ad schema and factory for registering new sources. |
| Web Dashboard (FastAPI + Jinja2) | [task-web-ui.md](./task-web-ui.md) | Add a read-only web dashboard for monitoring keywords, ads, users, and stats. Runs on port 8766 with basic auth. |
| Test suite with pytest (≥80% coverage) | [task-testing-pytest.md](./task-testing-pytest.md) | Add comprehensive unit tests covering scraper parsing, notification logic, price/postcode filters, retry queue, and scheduler flow. Configure coverage thresholds. |
## General Acceptance Criteria
- [ ] Web Dashboard is accessible at `http://<host>:8766` with basic auth
- [ ] Dashboard shows keywords with status, filters, subscribers, and last scrape time
- [ ] Dashboard shows recent ads with price, location, and keyword
- [ ] Dashboard shows users with mute/digest settings
- [ ] Dashboard shows stats (ads indexed, notifications sent, queue status)
- [ ] CI pipeline runs on every push to `main` and feature branches — fails if lint or coverage checks are not met
- [ ] Code coverage is ≥80% across all source files in `worker/src/`
- [ ] Multi-marketplace abstraction works — adding a new marketplace requires only creating one file under `scrapers/` with no changes to core logic
- [ ] All existing functionality (willhaben scraping, notifications) continues to work after refactoring
- [ ] The `/health` endpoint exposes test results or coverage stats (optional enhancement)
- [ ] All existing functionality (willhaben scraping, Telegram notifications, health server) continues to work
- [ ] Health server still works on port 8765 (no regression)
-363
View File
@@ -1,363 +0,0 @@
# 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`
```python
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)
```python
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:
```python
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:
```python
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
```bash
# 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
+77
View File
@@ -0,0 +1,77 @@
# Task: Web Dashboard (FastAPI + Jinja2)
## Description
Currently, the only way to monitor the system is via Telegram bot commands or SSH into the server. This task adds a read-only web dashboard for real-time visibility into keywords, ads, users, and stats.
## Architecture
```
┌──────────────────────────────────────────────┐
│ FastAPI App (port 8766) │
│ │
│ Auth: Basic Auth (WEB_UI_USERNAME/PASSWORD) │
│ Templates: Jinja2 with inline CSS │
│ │
│ Routes: │
│ GET / → Dashboard │
│ GET /keywords → Keywords list │
│ GET /keywords/<id> → Keyword detail │
│ GET /users → Users list │
│ GET /ads → Recent ads │
│ GET /stats → JSON stats │
└──────────┬───────────────────────────────────┘
┌──────────────────────────────────────────────┐
│ PostgreSQL (asyncpg pool) │
│ │
│ Queries: │
│ - Keywords with status, filters, subs │
│ - Recent ads with price, location │
│ - Users with mute/digest settings │
│ - Stats (counts, queue status) │
└──────────────────────────────────────────────┘
```
## Implementation Details
### 1. Add dependencies
In `worker/requirements.txt`:
```
fastapi==0.115.0
uvicorn==0.30.0
jinja2==3.1.4
```
### 2. Create `worker/src/web.py`
- FastAPI app with Jinja2 template engine
- Basic auth middleware using `WEB_UI_USERNAME` / `WEB_UI_PASSWORD` env vars
- Routes that query the DB via `get_pool()` from `db.py`
- Each route returns HTML via Jinja2 templates
### 3. Create `worker/src/templates/`
- `base.html` — Base layout with sidebar navigation, dark theme
- `dashboard.html` — Keywords overview + stats summary cards
- `keywords.html` — Table of keywords with status, filters, subscribers
- `keyword_detail.html` — Keyword detail with recent ads, price history, scrape logs
- `users.html` — Users list with mute/digest settings
- `ads.html` — Recent ads with search/filter
### 4. Integrate into `main.py`
- Start uvicorn server on port 8766 alongside existing aiohttp health server on 8765
- Graceful shutdown includes web server cleanup
## Acceptance Criteria
- [ ] Web UI accessible at `http://<host>:8766` with basic auth
- [ ] Dashboard shows keywords with status, filters, subscribers, last scrape
- [ ] Dashboard shows recent ads with price, location, keyword
- [ ] Dashboard shows users with mute/digest settings
- [ ] Dashboard shows stats (ads indexed, notifications sent, queue status)
- [ ] Health server still works on port 8765 (no regression)
- [ ] Telegram bot still works (no regression)