Ever watched a tech blog’s RSS feed and wondered why you keep seeing the same headlines over and over, while the truly fresh ideas slip through the cracks? I hit that wall last quarter when my MLOps dashboards needed “hot” engineering topics in seconds, not minutes. The culprit was a naïve scraper that re‑processed every article on each poll, inflating storage and drowning the signal with noise. In this post I’ll show you, step by step, how I turned the GitHub Engineering RSS feed into a reliable, idempotent trend detector that surfaces emerging keywords in real time—perfect for feeding downstream alerts or dashboards.
Key Takeaways
- Idempotency is best achieved by hashing the immutable parts of an RSS entry and persisting the hash in SQLite.
- Asynchronous `aiohttp` + `feedparser` reduces poll latency to sub‑second levels without blocking the event loop.
- spaCy’s linguistic pipeline lets you extract high‑quality noun‑phrases that serve as robust trend signals.
- A time‑based rolling window (e.g., 1 hour) combined with a Counter gives you a live “keyword heatmap”.
- Graceful error handling (network failures, DB lock errors) keeps the pipeline alive in production.
The Problem
Data engineers often need fresh, deduplicated signals from semi‑structured sources like RSS. The challenges are threefold: (1) the feed updates irregularly, (2) entries can reappear on subsequent polls, and (3) raw text must be transformed into actionable signals without manual curation. My goal was to build a self‑contained Python service that solves all three while staying under a 200 ms per‑poll budget.
Data and Sources
We pull the live feed from GitHub’s Engineering blog: https://github.blog/engineering/feed/. For NLP we rely on spaCy’s small English model (en_core_web_sm).
Data accessed on 2026-10-12.
Step 1 — Resilient Asynchronous Feed Ingestion with feedparser
The first sub‑problem is fetching the XML without blocking the rest of the pipeline. Using aiohttp lets us poll the feed every 30 seconds while the event loop stays free for processing.
import aiohttp
import asyncio
import feedparser
async def fetch_feed(session, url):
try:
async with session.get(url, timeout=10) as resp:
resp.raise_for_status()
text = await resp.text()
return feedparser.parse(text)
except aiohttp.ClientError as e:
print(f"Network error: {e}")
return None
Step 2 — Ensuring Idempotency with Content Hashing and SQLite
RSS entries are identified by a combination of title, link, and published timestamp. By hashing these fields we get a deterministic fingerprint. Storing the hash as a primary key in SQLite guarantees we never process the same article twice, even across restarts.
import hashlib
import sqlite3
def entry_hash(entry):
payload = f"{entry.get('title','')}{entry.get('link','')}{entry.get('published','')}"
return hashlib.sha256(payload.encode('utf-8')).hexdigest()
def init_db(db_path="rss_trends.db"):
conn = sqlite3.connect(db_path, timeout=30, isolation_level=None)
conn.execute(
"CREATE TABLE IF NOT EXISTS processed (hash TEXT PRIMARY KEY, title TEXT, published TEXT)"
)
return conn
def is_new_entry(conn, h):
cur = conn.execute("SELECT 1 FROM processed WHERE hash=?", (h,))
return cur.fetchone() is None
def mark_processed(conn, h, title, published):
conn.execute(
"INSERT OR IGNORE INTO processed (hash, title, published) VALUES (?,?,?)",
(h, title, published)
)
Step 3 — Extracting Semantic Features with spaCy for Keyword Analysis
We need meaningful signals, not just raw words. spaCy lets us pull noun chunks and filter out stop words, giving us concise technical phrases like “agent‑scale development” or “CSS‑in‑JS”.
import spacy
from collections import Counter
nlp = spacy.load("en_core_web_sm")
def extract_keywords(text, top_n=5):
doc = nlp(text)
phrases = [chunk.text.lower() for chunk in doc.noun_chunks if not chunk.is_stop]
return [p for p, _ in Counter(phrases).most_common(top_n)]
Step 4 — Simulating Real‑time Trend Tracking with a Rolling Window
Each new entry contributes its keywords to a time‑bucketed Counter. By pruning buckets older than our window (e.g., 1 hour) we keep the trend surface fresh.
import datetime
from collections import defaultdict, deque
WINDOW_SECONDS = 3600 # 1 hour
class TrendTracker:
def __init__(self):
self.buckets = deque() # each element: (timestamp, Counter)
def add(self, keywords):
ts = datetime.datetime.utcnow()
self.buckets.append((ts, Counter(keywords)))
self._prune()
def _prune(self):
cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=WINDOW_SECONDS)
while self.buckets and self.buckets[0][0] < cutoff:
self.buckets.popleft()
def aggregate(self):
total = Counter()
for _, cnt in self.buckets:
total.update(cnt)
return total
Putting It Together
We now wire the async fetcher, idempotency layer, keyword extractor, and trend tracker. Errors in network or DB operations are caught so the loop never dies.
import matplotlib.pyplot as plt
import os
async def monitor_loop(feed_url, db_conn, tracker):
async with aiohttp.ClientSession() as session:
while True:
parsed = await fetch_feed(session, feed_url)
if parsed is None:
await asyncio.sleep(30)
continue
for entry in parsed.entries:
h = entry_hash(entry)
if not is_new_entry(db_conn, h):
continue # duplicate, skip
title = entry.get('title', '')
summary = entry.get('summary', '')
content = f"{title}. {summary}"
keywords = extract_keywords(content)
if keywords:
tracker.add(keywords)
mark_processed(db_conn, h, title, entry.get('published', ''))
# Plot top‑5 trends after each poll
agg = tracker.aggregate()
if agg:
top = agg.most_common(5)
labels, values = zip(*top)
plt.figure(figsize=(6,4))
plt.bar(labels, values, color="#4A90E2")
plt.title("Top Technical Keywords (last hour)")
plt.ylabel("Mentions")
plt.tight_layout()
plt.savefig("output.png")
plt.close()
print("Trend plot updated → output.png")
await asyncio.sleep(30) # poll interval
Complete Script
The full runnable script combining all steps. Save it as trend_detector.py and run python trend_detector.py.
#!/usr/bin/env python3
"""
Idempotent real‑time trend detector for GitHub Engineering RSS.
"""
import asyncio
import aiohttp
import feedparser
import hashlib
import sqlite3
import spacy
import datetime
from collections import Counter, deque
import matplotlib.pyplot as plt
import os
import sys
# ---------- Configuration ----------
FEED_URL = "https://github.blog/engineering/feed/"
DB_PATH = "rss_trends.db"
WINDOW_SECONDS = 3600 # 1 hour
POLL_INTERVAL = 30 # seconds
TOP_N_KEYWORDS = 5
OUTPUT_PLOT = "output.png"
# ---------- Idempotency ----------
def entry_hash(entry):
payload = f"{entry.get('title','')}{entry.get('link','')}{entry.get('published','')}"
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def init_db(path=DB_PATH):
try:
conn = sqlite3.connect(path, timeout=30, isolation_level=None)
conn.execute(
"CREATE TABLE IF NOT EXISTS processed (hash TEXT PRIMARY KEY, title TEXT, published TEXT)"
)
return conn
except sqlite3.Error as e:
print(f"SQLite init error: {e}")
sys.exit(1)
def is_new_entry(conn, h):
cur = conn.execute("SELECT 1 FROM processed WHERE hash=?", (h,))
return cur.fetchone() is None
def mark_processed(conn, h, title, published):
try:
conn.execute(
"INSERT OR IGNORE INTO processed (hash, title, published) VALUES (?,?,?)",
(h, title, published)
)
except sqlite3.Error as e:
print(f"SQLite write error: {e}")
# ---------- Async Feed ----------
async def fetch_feed(session, url):
try:
async with session.get(url, timeout=10) as resp:
resp.raise_for_status()
text = await resp.text()
return feedparser.parse(text)
except aiohttp.ClientError as e:
print(f"Network error while fetching feed: {e}")
return None
# ---------- NLP ----------
nlp = spacy.load("en_core_web_sm")
def extract_keywords(text, top_n=TOP_N_KEYWORDS):
doc = nlp(text)
phrases = [
chunk.text.lower()
for chunk in doc.noun_chunks
if not any(tok.is_stop for tok in chunk)
]
return [p for p, _ in Counter(phrases).most_common(top_n)]
# ---------- Trend Tracker ----------
class TrendTracker:
def __init__(self, window_seconds=WINDOW_SECONDS):
self.window = datetime.timedelta(seconds=window_seconds)
self.buckets = deque() # (timestamp, Counter)
def add(self, keywords):
ts = datetime.datetime.utcnow()
self.buckets.append((ts, Counter(keywords)))
self._prune()
def _prune(self):
cutoff = datetime.datetime.utcnow() - self.window
while self.buckets and self.buckets[0][0] < cutoff:
self.buckets.popleft()
def aggregate(self):
total = Counter()
for _, cnt in self.buckets:
total.update(cnt)
return total
# ---------- Plotting ----------
def plot_trends(counter):
if not counter:
return
top = counter.most_common(5)
labels, values = zip(*top)
plt.figure(figsize=(6, 4))
plt.bar(labels, values, color="#4A90E2")
plt.title("Top Technical Keywords (last hour)")
plt.ylabel("Mentions")
plt.tight_layout()
plt.savefig(OUTPUT_PLOT)
plt.close()
print(f"Saved trend plot → {OUTPUT_PLOT}")
# ---------- Main Loop ----------
async def monitor_loop():
db_conn = init_db()
tracker = TrendTracker()
async with aiohttp.ClientSession() as session:
while True:
parsed = await fetch_feed(session, FEED_URL)
if parsed is None:
await asyncio.sleep(POLL_INTERVAL)
continue
for entry in parsed.entries:
h = entry_hash(entry)
if not is_new_entry(db_conn, h):
continue # duplicate
title = entry.get("title", "")
summary = entry.get("summary", "")
content = f"{title}. {summary}"
keywords = extract_keywords(content)
if keywords:
tracker.add(keywords)
mark_processed(db_conn, h, title, entry.get("published", ""))
agg = tracker.aggregate()
plot_trends(agg)
await asyncio.sleep(POLL_INTERVAL)
def main():
try:
asyncio.run(monitor_loop())
except KeyboardInterrupt:
print("\nInterrupted – shutting down.")
except Exception as e:
print(f"Unexpected error: {e}")
if __name__ == "__main__":
main()
Expected Output
Every 30 seconds you’ll see a line like Saved trend plot → output.png. Opening output.png shows a bar chart of the five most‑mentioned technical phrases from the last hour, e.g., “agent‑scale development”, “CSS‑in‑JS”, “diff surface”. The console also prints any network or SQLite warnings, but the loop never crashes.
Limitations and Tradeoffs
This design assumes the RSS feed stays under a few hundred items per poll; a massive feed would require pagination or a streaming parser. SQLite works for a single‑process daemon but becomes a bottleneck under multi‑worker scaling—swap it for PostgreSQL or a KV store in that case. spaCy’s small model is fast but may miss domain‑specific jargon; a custom NER model would improve recall at the cost of latency.
Frequently Asked Questions
What if the feed adds new XML fields or changes its structure?
Because we only rely on title, link, and published, most structural changes are harmless. Add defensive entry.get(..., '') checks if you need more fields.
Can I monitor multiple RSS feeds with the same script?
Yes. Instantiate a separate TrendTracker per feed or tag each entry with a feed identifier before hashing. For dozens of feeds consider a task queue (Celery/RQ) and a shared DB.
How do I reduce the memory footprint of the rolling window?
Store only the aggregated Counter per minute instead of per entry, or persist buckets to disk after each poll.
Is SHA‑256 overkill for deduplication?
It’s cheap (< 0.1 ms per entry) and eliminates collision risk even at millions of items. If you need ultra‑lightweight hashing, xxhash works too.
Why not use a streaming service like Kafka?
Kafka adds operational overhead. For a single‑source, low‑throughput use case the async‑SQLite combo is simpler and still provides durability.
What I'd Change
If I were to ship this to a production team, I’d replace the SQLite layer with a PostgreSQL table that supports UPSERT and connection pooling, and I’d move the NLP step into a separate worker process to keep the event loop truly non‑blocking. The current script is a solid prototype, but scaling to dozens of feeds and sub‑second latency demands a more decoupled architecture.
Next Steps
Try extending the pipeline to push the top‑5 keywords into a Slack channel or a Grafana dashboard via a webhook. Then experiment with a custom spaCy pipeline that flags emerging proper nouns (e.g., new project names) and see how quickly you can surface brand‑new engineering initiatives.