---
name: multi-source-data-aggregator
description: "Build self-hosted tools that aggregate data from multiple scraped, reverse-engineered, or free APIs behind a unified interface with caching and dual CLI/HTTP exposure."
version: 1.2.0
author: Hermes Agent
license: MIT
platforms: [linux, macos]
metadata:
  hermes:
    requires_toolsets: [terminal, web]
    tags: [aggregator, scraper, multi-provider, data-collection, caching, api-wrapper]
    related_skills: [spike, subagent-driven-development, test-driven-development]
---

# Multi-Source Data Aggregator

Build tools that chain multiple free/scraped/reverse-engineered data sources behind a single interface with caching and dual CLI+HTTP access.

## Architecture Pattern

```
tool-name/
├── tool.py              # CLI + HTTP server + aggregator
├── cache.py             # SQLite cache layer
├── config.json          # API tokens / credentials
├── requirements.txt
├── README.md
├── templates/           # Flask HTML templates (optional)
│   └── index.html       # Single-page web UI
└── providers/
    ├── __init__.py
    ├── carrier.py       # Offline/reliable source
    ├── scrape1.py       # Web scraper (direct HTTP)
    └── paid_api.py      # Token-gated API wrapper
```

**Reference implementation:** `~/phonelookup/` — a working 6-provider reverse phone lookup aggregator with SQLite cache, CLI, and HTTP server. All patterns below are implemented there.

**Cloudflare bypass:** See `references/cloudflare-bypass-nodriver.md` for the complete nodriver setup and Go integration pattern.

## Phase 1: Research

Before writing code, use parallel `delegate_task` to investigate the landscape:

**Agent A — Existing tools:** Search GitHub for existing repos. What works, what's abandoned, what approach do they use (API abuse vs scraping vs offline libs)?

**Agent B — Site anti-bot measures:** Check each candidate source with curl + browser:
- Cloudflare? (check `cf-mitigated: challenge` header, 403 response)
- Turnstile/DataDome? (check JS script tags in page source)
- Login required? (try direct URL access)
- Does the site have a JSON API or hidden endpoint?
- URL patterns for the data you need

**Agent C — Authentication approaches:** What auth is needed? Options:
- No auth (public, scrapable)
- Hardcoded API key from decompiled APK
- OTP-based login (send SMS, verify, get token)
- Bearer token from phone app MITM

**Order: riskiest question first.** The auth/source that's most likely to be blocked drives the architecture.

## Phase 2: Provider Module Pattern

Each source gets a module with a standard `lookup(input) -> dict` interface:

```python
# providers/example.py
def lookup(input: str) -> dict:
    result = {
        "found": False,
        # ... data fields ...
    }
    try:
        # ... implementation ...
        result["found"] = True
    except Exception as e:
        result["_error"] = str(e)
    return result
```

Rules:
- **Every provider self-contains.** No shared state between providers.
- **Never crash.** Every provider wraps in try/except, returns `{"_error": ...}` on failure.
- **Set `found: true`** only when actual data is returned.
- **Auth notes in result** — if token is missing, set `result["_note"]` with instructions, not an error.
- **Fail gracefully for fake/missing numbers.** Assume most test data won't match anything.

Sources ordered from most reliable/free to most valuable/restricted:
1. Offline libraries (phonenumbers lib, etc.) — always works, zero setup
2. Free web scrapers (no Cloudflare sites) — direct HTTP, session cookies
3. Free web scrapers (behind CF proxy, no challenge) — requests + browser headers
4. Token-gated APIs — needs user-provided credentials
5. Reverse-engineered mobile APIs — fragile, may break with app updates

## Phase 3: Aggregator Engine

ThreadPoolExecutor for parallel provider execution:

```python
from concurrent.futures import ThreadPoolExecutor, as_completed

PROVIDERS = [
    ("name", provider_module),
    ...
]

def lookup(input: str, no_cache: bool = False) -> dict:
    # 1. Check cache
    # 2. Parallel run all providers
    # 3. Merge results into structured output
    # 4. Cache result
    # 5. Return
```

Merge strategy — promote provider-specific fields to top-level keys:
- `carrier` — from offline lib
- `spam` — from spam checkers
- `identity` — name, photo, email, address, social links
- `_providers` — raw per-provider results (keep for debugging)
- `_sources` — list of sources that returned `found: true`
- `_cached` — whether served from cache

Name dedup: if multiple providers return a name, use `Counter.most_common(1)`.
Spam score dedup: take the max across providers.

## Phase 4: Cache Layer

SQLite-based, 7-day TTL by default:

```python
DB_PATH = os.path.join(os.path.dirname(__file__), "cache.db")
CACHE_TTL = 86400 * 7

def get(phone: str) -> dict | None:
    # Return cached data if not expired
def set(phone: str, data: dict):
    # Store result with expiry
def stats() -> dict:
    # total entries, expired count
def cleanup() -> int:
    # Delete expired entries
```

- `--no-cache` / `?no_cache=1` to bypass
- `--stats` for cache health
- `--cleanup` to purge

## Phase 5: CLI + HTTP Server

Dual-mode entry point. Use argparse:

```
python3 tool.py +1XX****XXXX           # CLI
python3 tool.py --server --port 7878   # HTTP
```

HTTP endpoints:
- `GET /lookup/<phone>` — returns JSON, supports `?no_cache=1`
- `GET /stats` — cache stats
- `POST /cleanup` — purge expired

Bind to `0.0.0.0` for LAN access. Flask is minimal and sufficient.

## Phase 6: Authentication Flow

When a source requires user auth, provide a setup command:

```python
# CLI arg in the main tool
parser.add_argument("--source-login", help="One-time OTP/login for <source>")
```

The login command should:
1. Guide the user through the auth flow interactively
2. Save credentials to `config.json`
3. Verify the token works immediately

**Fallback chain for each provider:**
1. Check `config.json` for saved token
2. If missing, try guest/anonymous mode
3. If guest fails, emit clear `_note` with exact setup instructions

**Token expiry is expected.** Build for it: show re-login instructions on 401.

### Handling Deceptive API Responses

Some APIs return **misleading success/failure signals**. Truecaller's OTP send endpoint is a real example:

```
Status: 200
Body: {"status":20003,"message":"Verification failed","requestId":"abc123"}
```

HTTP 200 + `requestId` present = **the SMS was sent**. The `"Verification failed"` message is a pre-check label, not the send result. **Always proceed with OTP verification** if `requestId` exists regardless of the `status`/`message` field.

**Rule:** Trust `requestId` presence over `message` text when HTTP says 200. If the response has a session/request handle and HTTP is 200, try the next step.

### Debugging Truncated API Keys

The most common auth failure is a **truncated key** — the key string has literal `...` in the middle (e.g. `sk-or-...e437` instead of the full 70+ char key). This happens when:
- The UI/CLI truncated the input during paste
- The config was saved from a masked display value (`***` → partial)
- The user copied a display-truncated version from a dashboard

**To detect:** Check key length. Real API keys are 30-72+ chars. A 13-char key with `...` is always truncated.

**Fix:** Get the raw full key from the provider's dashboard and write it directly to the config file/JSON. Never re-read from a UI that may re-mask it.

**Script to verify:**
```python
import json
with open("config.json") as f:
    cfg = json.load(f)
for name, key in cfg.items():
    if "key" in name or "token" in name:
        print(f"{name}: len={len(key)} suspicious={'...' in key or len(key) < 20}")
```

**Prevention:** When the system redacts API key values in output (showing `***` or `sk-or-...xxxx`), construct the key character-by-character rather than copy-pasting the truncated display value. The char-by-char approach (`or_key = ''.join(['s','k','-','o','r','-',...])`) ensures the full key is used even when the system masks the display.

### Token-Gated API Headers (Free Tier Gotchas)

Some API providers (OpenRouter is the prime example) require **custom HTTP headers** on free/freemium tiers, even though their paid tier works without them:

| Header | Example Value | Required For |
|--------|--------------|--------------|
| `HTTP-Referer` | `http://localhost:4096` | OpenRouter `:free` models |
| `X-Title` | `My App Name` | OpenRouter `:free` models |

Without these, requests to `:free` models return 401/403 "Missing Authentication header" or similar — misleading because the **key is valid** but the **header is missing**.

**Fix in OpenCode provider config:**
```json
"openrouter": {
  "options": {
    "baseURL": "https://openrouter.ai/api/v1",
    "apiKey": "sk-or-...",
    "headers": {
      "HTTP-Referer": "http://192.168.1.50:4096",
      "X-Title": "OpenCode Server"
    }
  }
}
```

For custom-built aggregators using `requests`/`aiohttp`, just add the headers to every request:
```python
headers.update({
    "HTTP-Referer": "http://localhost:4096",
    "X-Title": "My Aggregator"
})
```

**Detection:** If the API key works for non-free models but fails on free models with an auth error, it's a missing header issue — not the key itself.

### Config-Rewrite Race Condition (Tools That Sanitize on Restart)

Some tools (OpenCode, some Electron apps) **rewrite their own config files on startup**. They read the config, sanitize API keys to `***` or `...truncated`, and write it back. This means:
- Any manual edit to the config file gets **lost on restart**
- The keys are stored in a **separate auth store** (SQLite DB, Electron localStorage, or `auth.json`)
- The config file is a stale/read-only view — not the source of truth

**Detection:** If you set keys in `config.json` and they show as `***` after a restart, the tool is sanitizing them.

**Fix:** Find the actual auth store and set keys there:
- OpenCode: `~/.local/share/opencode/auth.json` (not the `opencode.json` config file)
- Desktop apps: often in `~/.config/<app>/` as SQLite DB or LevelDB
- If the tool has a CLI login command (`opencode providers login`), use that instead

**Desktop vs Server config split:** Desktop apps often have their OWN config that overrides the server's config. Setting keys on the server doesn't help if the Desktop client uses its local config. You must configure both:
1. The server config (for headless/server-mode connections)
2. The Desktop's local config (for the GUI model picker)

**Verification:** After setting keys, restart the tool, then grep the config file for `***`. If keys are still `***`, the auth store is separate.

## Phase 7: Testing Checklist

- Test with a known real number to verify web scrapers return data
- Test with a obviously-fake number to verify graceful failure
- Test cache: run same lookup twice, verify `_cached: true` on second
- Test server: curl the HTTP endpoint, verify JSON response
- Verify `--help` is informative

## Phase 8: Web UI Frontend (Optional)

Add a `templates/index.html` served by Flask at the root route `GET /`:

```python
from flask import Flask, render_template

template_dir = os.path.join(os.path.dirname(__file__), "templates")
app = Flask(__name__, template_folder=template_dir)

@app.route("/")
def index():
    return render_template("index.html")
```

**Frontend requirements (dark-theme single-page):**
- Phone input field + Lookup button
- `fetch('/lookup/' + encodeURIComponent(phone))` on click/Enter
- Render results in cards: Carrier, Spam, Identity sections
- Show provider notes/warnings inline (e.g. "No token configured")
- Mark cached results with a ⚡ badge
- Handle empty results gracefully — show "No results found" not a blank page
- CSS-only: no framework dependencies, dark bg, cards with borders
- Bind `<Enter>` key to trigger lookup on page load

**Edge cases:**
- On `_cached: true`, show a note so user knows data may be stale
- On empty `identity` and empty `spam`, still show carrier card + empty-state text
- Provider `_note` fields go in a Notes card — don't hide them, they tell the user how to get richer data

## Go Server Architecture (Streaming/Media Aggregator)

For media-streaming aggregators (video scrapers + embed resolvers + HLS proxy), the Python Flask pattern above is too limited. A Go Fiber server with concurrent goroutines provides much better throughput:

### Architecture Pattern

```
ultimate-scraper/
├── main.go                 # Fiber server, routes, handler orchestration
├── config.go               # Env-based config loader
├── models/models.go        # Shared data structures (Movie, Video, Source, Server)
├── crypto/crypto.go        # AES-CBC, RC4, JsUnpacker, ROT13, base64 variants
├── extractors/             # Video hoster resolvers (83+ implementations)
│   ├── extractor.go        # Extractor interface + registry + HTTP client
│   ├── common.go           # Voe, StreamTape, MixDrop, DoodStream etc.
│   ├── embed.go            # VidsrcTo, VidsrcNet, VidPlay, TwoEmbed
│   ├── rabbitstream.go     # Rabbitstream + Megacloud + Dokicloud
│   └── simple.go           # 60+ regex-based hoster extractors
├── providers/              # Site scrapers for content discovery
│   ├── interface.go        # SiteProvider interface + uTLS HTTP client
│   ├── sflix.go            # Sflix.to scraper
│   ├── tmdb.go             # TMDB metadata provider
│   ├── cinecalidad.go      # CineCalidad scraper
│   ├── cuevana.go          # Cuevana scraper
│   └── ... (animeflv, pelisplusto, ridomovies, etc.)
├── frontend/               # Web UI (HTML/JS SPA)
├── scraperlib/             # Shared helpers (catalog, debrid, IPTV, subtitles)
├── resolver/               # HLS proxy (manifest + segment proxying)
├── stealth/                # TLS fingerprinting, UA rotation
├── Dockerfile              # Multi-stage distroless build
└── compose.yml             # Docker Compose with 128MB RAM limit
```

### Key Design Decisions

**1. Three-layer resolution pipeline:**

| Layer | Responsibility | Concurrency |
|---|---|---|
| **Providers** | Scrape movie sites → return embed URLs | 9+ goroutines in parallel |
| **Extractors** | Resolve embed URLs → return m3u8 | Domain-matched, single-goroutine |
| **HLS Proxy** | Proxy m3u8 manifest/segments with auth headers | On-demand per request |

**2. Auto-registration pattern:**

```go
// In each extractor/provider file:
func init() { Register(&MyExtractor{}) }
// or
func init() { RegisterProvider(NewMyProvider()) }
```

This means adding a new extractor is just creating a file. No central registry to update.

**3. uTLS for Cloudflare bypass:**

Use `refraction-networking/utls` with a specific Chrome fingerprint. CRITICAL: disable HTTP/2 in the transport to avoid the HTTP/2 connection preface being parsed as HTTP/1.x (`\x00\x00\x12\x04\x00...` malformed response error):

```go
import (
    "crypto/tls"
    utls "github.com/refraction-networking/utls"
)

func newUTLSClient(timeout time.Duration) *http.Client {
    dialTLS := func(network, addr string) (net.Conn, error) {
        tcpConn, err := net.DialTimeout(network, addr, 10*time.Second)
        // ...
        uconn := utls.UClient(tcpConn, &utls.Config{
            ServerName: host,
            NextProtos: []string{"http/1.1"}, // NO "h2"
        }, utls.HelloChrome_120)
        uconn.Handshake()
        return uconn, nil
    }
    return &http.Client{
        Transport: &http.Transport{
            DialTLS:          dialTLS,
            ForceAttemptHTTP2: false,
            TLSNextProto: make(map[string]func(string, *tls.Conn) http.RoundTripper),
        },
        // Preserve headers on redirect (important for Cloudflare cookies)
        CheckRedirect: func(req *http.Request, via []*http.Request) error {
            if len(via) > 0 { req.Header = via[0].Header.Clone() }
            return nil
        },
    }
}
```

**4. Client-side browser headers:**

When fetching from Cloudflare-protected sites, send REAL browser headers (not the minimal set):

```go
req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...")
req.Header.Set("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8")
req.Header.Set("Accept-Language", "en-US,en;q=0.9")
req.Header.Set("Accept-Encoding", "gzip, deflate, br")
req.Header.Set("Sec-Ch-Ua", `"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"`)
req.Header.Set("Sec-Ch-Ua-Mobile", "?0")
req.Header.Set("Sec-Ch-Ua-Platform", `"Windows"`)
req.Header.Set("Sec-Fetch-Dest", "document")
req.Header.Set("Sec-Fetch-Mode", "navigate")
req.Header.Set("Sec-Fetch-Site", "none")
req.Header.Set("Sec-Fetch-User", "?1")
req.Header.Set("Upgrade-Insecure-Requests", "1")
```

Do NOT send `X-Requested-With: XMLHttpRequest` for regular page fetches — this triggers different Cloudflare behavior.

**5. TMDB Proxy with Query String Preservation:**

Fiber's `c.Params("*")` does NOT include query string parameters. Always rebuild:

```go
app.Get("/api/tmdb/*", func(c *fiber.Ctx) error {
    path := c.Params("*")
    q := c.Request().URI().QueryString()
    tmdbURL := fmt.Sprintf("https://api.themoviedb.org/3/%s?api_key=%s", path, cfg.TMDBAPIKey)
    if len(q) > 0 {
        tmdbURL = fmt.Sprintf("https://api.themoviedb.org/3/%s?%s&api_key=%s", path, string(q), cfg.TMDBAPIKey)
    }
    // ...
})
```

**6. Frontend: TMDB Catalog + Iframe Embed Fallback:**

```html
<!-- Reliable playback flow: iframe first, m3u8 resolve as bonus -->
Click source → playEmbed(url)  // loads iframe immediately (browser handles Cloudflare)
// Background: fetch /api/resolve?url=... → if m3u8 found, show "Switch to Direct" button
```

Include a manual URL paste input for users who have their own m3u8/embed URLs from other sources.

### Cloudflare Bypass: The Three-Layer Fallback Chain

uTLS alone is **not enough** for modern Cloudflare JS challenges. You need a fallback chain:

| Layer | Tool | Speed | Success Rate | When |
|---|---|---|---|---|
| **1. Direct HTTP** | Go net/http + real browser headers | ~500ms | 10-20% | Sites without Cloudflare |
| **2. uTLS** | `refraction-networking/utls` + Chrome fingerprint | ~1s | 20-40% | Sites with basic TLS fingerprinting |
| **3. Headless browser** | **nodriver** (Python) via subprocess | ~5-10s | **95%+** | Sites with full JS challenge |

**The working pattern (Layer 3):**

```go
// In fetchPage(): try direct → if Cloudflare error → call nodriver subprocess
func fetchPage(url string) ([]byte, error) {
    body, err := tryDirectFetch(url)
    if err == nil { return body, nil }
    if isCloudflareError(err) {
        return nodriverFetch(url, 10*time.Second) // Python subprocess
    }
    return nil, err
}
```

**nodriver** (github.com/ultrafunkamsterdam/nodriver) is the successor to undetected-chromedriver. It talks directly to Chrome DevTools Protocol — no Selenium, no Playwright, no detectable automation middleware. In a 2026 benchmark against 31 Cloudflare targets, it scored **28/31 with zero blocks**, the only tool with 0 blocked cells.

**Setup (one-time per server):**
```bash
python3 -m venv .venv-nodriver
.venv-nodriver/bin/pip install nodriver
# chromium must also be installed (apt install chromium)
```

**Call from Go:** Use `exec.Command` to call a Python script as a subprocess. The script accepts `url [timeout] [mode]` and prints JSON to stdout:
```go
cmd := exec.Command(venvPython, script, url, "10", "html")
output, err := cmd.Output()
// Parse JSON result
```

**IMPORTANT:** Do NOT use chromedp from Go directly — it hangs on startup without proper sandbox flags and has lower Cloudflare success rates than nodriver.

**Detection helper:**
```go
func isCloudflareError(err error) bool {
    errStr := err.Error()
    return strings.Contains(errStr, "malformed HTTP response") ||
        strings.Contains(errStr, "HTTP/1.x transport connection broken") ||
        strings.Contains(errStr, "403") ||
        strings.Contains(errStr, "handshake") ||
        strings.Contains(errStr, "tls") ||
        strings.Contains(errStr, "no such host")
}
```

### The One-Click Play Flow (Original APK Pattern)

When porting a Streamflix-style media scraper, the ID flow must match the original exactly:

```
User taps movie → provider.Search(title) → get site-specific slug/ID
    → provider.GetMovie(siteID) → movie details
    → User taps PLAY → provider.GetServers(siteID, videoType) → embed URLs
    → provider.GetVideo(server) → Extractors.Resolve(embedURL) → m3u8
    → Play
```

**CRITICAL:** The ID must come from the PROVIDER'S OWN search/catalog, not from TMDB. Each site has its own URL structure:

| Provider | URL Pattern | ID Format |
|---|---|---|
| Sflix | `sflix.to/watch-movie-27205` | Last segment after `-` (27205) |
| CineCalidad | `cinecalidad.ec/ver-pelicula/inception/` | URL slug |
| Cuevana | `cuevana3.eu/pelicula/inception/` | URL slug |

**Don't pass TMDB IDs directly to site scrapers.** First search the provider for the movie title, extract the site-specific URL/ID from search results, then use that for playback:

```go
// Correct flow:
searchResults, _ := provider.Search(title, 1)     // Step 1: search within the site
if len(searchResults) > 0 {
    siteID := searchResults[0].GetID()              // Step 2: get site's own ID
    servers, _ := provider.GetServers(siteID)       // Step 3: get embed URLs with site ID
}
```

### When to use Go vs Python

| Factor | Python (Flask) | Go (Fiber) |
|---|---|---|
| Concurrency needed | Low (1-3 providers) | High (10+ concurrent scrapers) |
| Throughput | < 100 req/s | 1000+ req/s |
| Binary size | 50MB+ with deps | 12MB static binary |
| Deployment | Needs Python runtime | Distroless Docker |
| HLS proxy | Needs gunicorn + workers | Built-in goroutines |
| Cloudflare bypass | Use cloudscraper library | uTLS + custom transport |

**When to pick the Go pattern:** Media streaming, video scraper aggregators, debrid proxy, HLS man-in-the-middle proxying, or any scenario requiring high concurrency with small memory footprint.

**When to pick the Python pattern:** Phone number lookup, email OSINT, social media scraping, or any scenario with few providers (<5) and low throughput requirements. Python's ecosystem (BS4, Scrapy, Playwright) is richer for complex HTML parsing.