Building a Multi-Engine Search & Project Mention Aggregator

As developers, musicians, and creators, keeping track of our digital footprint across multiple projects can be a daunting task. Between web engineering portfolios, scientific data sonification research, and album releases across various band configurations, mentions, reviews, and catalog indices are scattered all over the web.

To solve this, we built a focused, lightweight command-line aggregator in Python that queries multiple search engines simultaneously, parses and consolidates the results, tracks previously seen items in a local cache, and formats them into clean, actionable Markdown reports.

Here is a technical walkthrough of how we integrated multiple search APIs (Google/Serper and Brave Search), designed a robust, key-free scraper fallback for DuckDuckGo, and implemented delta-tracking to highlight new web mentions.

---

The Architecture: Consolidated Search Pipelines

Rather than manually querying search engines or relying on complex, bloated enterprise monitoring suites, our aggregator relies on a direct, multi-threaded search pipeline.

The tool organizes search targets into three core pillars:

1. 💻 Web Development: Monitoring studio projects under the Kennebec and mydevelopment names, interactive audio galleries, and Web Audio API applications.
2. 🔬 Science & Data Sonification: Tracking academic papers, soundscape research in partnership with Cities & Memory, and genealogical lineage publications.
3. 🎵 Music & Bands: Aggregating album reviews, blog features, and Bandcamp/Discogs releases across several active musical projects (Karhide, Ann Arbor, Half Seas Over, Constant Pool, Synnax, etc.).

graph TD
    A[CLI Input / Cron] --> B[Target Configurations]
    B --> C{Active Search Engines}
    C -->|Serper API| D[Google Search Results]
    C -->|Brave API| E[Brave Search Results]
    C -->|DuckDuckGo Scraping| F[DDG HTML Parser]
    D & E & F --> G[Merge & Deduplicate Links]
    G --> H[Compare with seen_items_cache.json]
    H -->|Identify Unseen Links| I[Tag 🔥 NEW Items]
    I --> J[Generate Markdown Report & Update Cache]

To optimize execution speed, queries across these different targets are run concurrently using Python’s concurrent.futures.ThreadPoolExecutor, cutting down API response latency dramatically.

---

Deep Dive: Integrating the Search APIs

Each search engine has its own authentication mechanisms, query structure, and return formats. Our aggregator normalizes these variations into a uniform dictionary model containing the document’s title, link, snippet, and index date.

1. Google Search via Serper.dev

Serper is a fast, low-cost API gateway to Google Search results. It accepts JSON payloads and returns structured search results, supporting timeframe queries (e.g. past day, week, month) via Google’s tbs search parameter.

import urllib.request
import json

def query_serper_api(query, api_key, timeframe="m"):
    url = "https://google.serper.dev/search"
    payload_dict = {"q": query, "num": 5}
    
    # Configure timeframe constraints (qdr:d, qdr:w, qdr:m, qdr:y)
    if timeframe and timeframe.lower() != "all":
        payload_dict["tbs"] = f"qdr:{timeframe.lower()}"

    headers = {
        'X-API-KEY': api_key,
        'Content-Type': 'application/json'
    }
    
    req = urllib.request.Request(url, data=json.dumps(payload_dict).encode('utf-8'), headers=headers)
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read().decode('utf-8'))
        return [
            {
                'title': item.get('title'),
                'link': item.get('link'),
                'snippet': item.get('snippet'),
                'date': item.get('date', 'N/A')
            }
            for item in data.get('organic', [])
        ]

2. Brave Search API

The Brave Search API provides independent web search indexes. Brave handles timeframe constraints via its freshness query parameter (mapping to pd (past day), pw (past week), etc.) and requires key authentication through the X-Subscription-Token header.

def query_brave_api(query, api_key, timeframe="m"):
    url = "https://api.search.brave.com/res/v1/web/search"
    params = {"q": query, "count": 5}
    
    # Freshness mapping
    tf = timeframe.lower()
    freshness_map = {'d': 'pd', 'w': 'pw', 'm': 'pm', 'y': 'py'}
    if tf in freshness_map:
        params["freshness"] = freshness_map[tf]

    query_string = urllib.parse.urlencode(params)
    headers = {
        'Accept': 'application/json',
        'X-Subscription-Token': api_key
    }
    
    req = urllib.request.Request(f"{url}?{query_string}", headers=headers)
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read().decode('utf-8'))
        results = data.get('web', {}).get('results', [])
        return [
            {
                'title': item.get('title'),
                'link': item.get('url'),
                'snippet': item.get('description'),
                'date': item.get('age') or item.get('page_age') or 'N/A'
            }
            for item in results
        ]

3. DuckDuckGo Fallback Scraper (Key-Free)

To ensure the script functions even without paid API subscriptions, we implemented a fallback crawler targeting the non-JS html.duckduckgo.com page. Using Python’s standard html.parser.HTMLParser, we extract result blocks, titles, URLs, and snippets.

To prevent IP bans or bot-detection flags, a global lock controls DuckDuckGo calls, injecting a randomized sleep delay between queries:

import time
import random
import threading
from html.parser import HTMLParser

class DDGHTMLParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.results = []
        self.current_result = None
        self.in_title = False
        self.in_snippet = False
        self.temp_title = []
        self.temp_snippet = []

    def handle_starttag(self, tag, attrs):
        attrs_dict = dict(attrs)
        cls = attrs_dict.get('class', '')
        if tag == 'div' and 'web-result' in cls:
            if self.current_result:
                self.results.append(self.current_result)
            self.current_result = {'title': '', 'link': '', 'snippet': '', 'date': 'N/A'}
        
        if self.current_result is not None:
            if tag == 'a' and 'result__a' in cls:
                self.in_title = True
                self.temp_title = []
                href = attrs_dict.get('href', '')
                self.current_result['link'] = self.parse_ddg_url(href)
            elif tag == 'a' and 'result__snippet' in cls:
                self.in_snippet = True
                self.temp_snippet = []

    def handle_data(self, data):
        if self.in_title:
            self.temp_title.append(data)
        elif self.in_snippet:
            self.temp_snippet.append(data)

    def handle_endtag(self, tag):
        if self.in_title and tag == 'a':
            self.in_title = False
            self.current_result['title'] = ''.join(self.temp_title).strip()
        elif self.in_snippet and tag == 'a':
            self.in_snippet = False
            self.current_result['snippet'] = ''.join(self.temp_snippet).strip()
            self.results.append(self.current_result)
            self.current_result = None

    def parse_ddg_url(self, url):
        # Extracts actual destination URL from redirect query parameter 'uddg'
        if 'uddg=' in url:
            parsed = urllib.parse.urlparse(url)
            qs = urllib.parse.parse_qs(parsed.query)
            if 'uddg' in qs:
                return qs['uddg'][0]
        return url

# Thread-safe query handler with natural request pacing
ddg_lock = threading.Lock()

def query_duckduckgo(query, timeframe="m"):
    with ddg_lock:
        time.sleep(random.uniform(1.5, 3.5)) # Pacing delay
        url = "https://html.duckduckgo.com/html/?"
        params = {"q": query}
        if timeframe in ('d', 'w', 'm', 'y'):
            params["df"] = timeframe
            
        req_url = url + urllib.parse.urlencode(params)
        headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel...'}
        
        req = urllib.request.Request(req_url, headers=headers)
        with urllib.request.urlopen(req, timeout=10) as resp:
            parser = DDGHTMLParser()
            parser.feed(resp.read().decode('utf-8'))
            parser.close()
            return parser.results

---

Delta Tracking: Highlighting Unseen Mentions

An aggregator is only useful if it makes it easy to find new information. To avoid showing the same historical pages on every run, the script stores a simple JSON ledger of previously seen URLs in a file called seen_items_cache.json.

During processing, we compare incoming links against the cache:

• If the cache is empty (first run), all links are loaded as a baseline.
• If a cache already exists, any link not present in the historical set is marked with an is_new flag.
• The output Markdown report parses this flag and prepends a 🔥 [NEW] badge to the title.
• The new URLs are appended to the JSON cache for future runs.

This keeps reports clean and lets us monitor new indexing activity or reviews in real time.

---

Rebuilding & Reviewing

Using native standard library calls and strict schema separations, the aggregator provides a highly efficient command-line workflow:

# Aggregates web mentions across all projects from the past week
python3 tim_waterfield_aggregator.py --focus all --timeframe w --engine all

# Run specific searches on bands only, exporting to a custom markdown file
python3 tim_waterfield_aggregator.py --focus music --output digests/music_august.md

By decoupling search extraction from complex databases and web-servers, the project proves that targeted information-gathering pipelines don’t need heavy infrastructures—just clean, asynchronous Python scripts routing to reliable, modern Search APIs.