Architecture

Pipeline Flow

The system is built as a chain of decoupled workers connected by RabbitMQ queues. Each worker is an independent process (or container) that consumes from one queue, processes, and publishes to the next.

flowchart LR
    subgraph Producer
        P[producer.py]
    end
    subgraph Queues
        UQ[url.queue]
        FQ[fetch.queue]
        RQ[raw.queue]
        PQ[parsed.queue]
    end
    subgraph Workers
        DW[dedup_worker]
        FW[fetch_worker]
        PW[parse_worker]
        SW[store_worker]
    end
    subgraph Storage
        S3[Garage S3]
        PG[PostgreSQL]
        J[JSON/CSV files]
    end
    subgraph LLM
        L[Chunker]
        E[Embedder]
        V[pgvector]
        A[Search API]
    end

    P -->|"🟢 URLs"| UQ
    UQ --> DW
    DW -->|"✅ unique URLs"| FQ
    FQ --> FW
    FW -->|"📄 raw HTML/JSON"| RQ
    RQ --> PW
    PW -->|"📊 parsed items"| PQ
    PQ --> SW
    SW --> S3 & PG & J
    S3 --> L
    L -->|"chunks"| E
    E -->|"vectors"| V
    V --> A

Queue Chain

Queue	Producer	Consumer	Payload
url.queue	Producer scripts	Dedup worker	URL + metadata
fetch.queue	Dedup worker	Fetch worker	Unique URLs to scrape
raw.queue	Fetch worker	Parse worker	Raw HTML/JSON response
parsed.queue	Parse worker	Store worker	Structured items
dead_letter.queue	Any	—	Failed/unprocessable messages

Message Format

Messages are serialized with msgpack for compactness and speed:

# Core message structure
{
    "url": str,
    "job_id": str,
    "stage": str,           # "listing", "detail", etc.
    "profile": str,         # site profile name
    "source_url": str | None,  # parent URL (for followed links)
    "retry_count": int,
    "payload": dict | None,    # stage-specific metadata
}

Prefect Orchestration (Production)

In production, a Prefect flow wraps the full lifecycle:

flowchart LR
    A[Preflight] --> B[Verify Structure]
    B -->|"⛔ changed/unverified"| C[Skip - Alert]
    B -->|"✅ ok"| D[Producer]
    D --> E[Wait for Drain]
    E --> F[LLM Pipeline]
    F --> G[Report + Slack]

The flow verifies that: 1. All queues have active consumers (workers are up) 2. The site structure hasn't drifted from its baseline fingerprint 3. The scrape completes and queues drain before the LLM step begins

Plugin System

Each pipeline stage uses a plugin interface registered via decorators:

plugins/
├── base.py                 # Abstract base classes
├── fetchers/
│   ├── cloakbrowser_fetcher.py   # Anti-bot browser
│   └── wreq_fetcher.py           # Lightweight HTTP
├── parsers/
│   ├── generic_parser.py         # CSS-selector based
│   ├── jsonld_parser.py          # JSON-LD extraction
│   ├── apollo_json_parser.py     # Apollo GraphQL state
│   ├── g2_listing_parser.py      # G2-specific
│   ├── hermes_listing_parser.py  # Hermès-specific
│   ├── idealista_detail.py       # Idealista-specific
│   └── trustpilot_nextdata_parser.py  # Next.js state
└── storage/
    ├── garage_s3_storage.py
    ├── postgres_storage.py
    ├── json_storage.py
    ├── csv_storage.py
    └── markdown_storage.py

[!NOTE] See the Plugins section for detailed documentation on each plugin interface and how to add new ones.

Core Modules

Module	Responsibility
core/settings.py	Pydantic-based configuration from .env
core/profiles.py	YAML profile loading and validation
core/queue.py	RabbitMQ connection, channel, publish/consume
core/redis.py	Redis connection pool
core/dedup.py	Bloom-filter deduplication via RedisBloom
core/browser_pool.py	CloakBrowser session lifecycle management
core/rate_limiter.py	Per-stage rate limiting
core/retry.py	Configurable retry with backoff strategies
core/job_cursor.py	Resume cursor for interrupted scrapes
core/job_metrics.py	Per-job success/block/error counters
core/site_metrics.py	Cumulative per-site statistics
core/site_versioning.py	Structural fingerprinting and comparison
core/registry.py	Plugin decorator registry
core/alerts.py	Slack webhook alerting with cooldowns