diff --git a/adapters/rss_feeds.py b/adapters/rss_feeds.py new file mode 100644 index 0000000..bc57703 --- /dev/null +++ b/adapters/rss_feeds.py @@ -0,0 +1,282 @@ +#!/usr/bin/env python3 +""" +RSS Feeds adapter for Athena. +Ingests commercial AI news from curated RSS feeds — no auth, no scraping, +reliable structured data. + +Fills the "commercial AI news" gap: corporate moves, funding, product launches, +partnerships, policy. Genuinely different signal from GitHub/arXiv/Reddit/HN/HF. + +Feeds selected for AI relevance, reliability, and freshness. Techmeme skipped +(requires JS rendering). Axios/The Decoder kept despite 3xx status codes +(feedparser follows redirects fine). + +Usage: + python3 adapters/rss_feeds.py +""" + +import json +import os +import re +import time +import sqlite3 +import feedparser +from datetime import datetime, timedelta, timezone +from email.utils import parsedate_to_datetime + +from adapters import SourceAdapter + + +# Curated feed list — AI-focused, reliable, diverse publishers. +# Format: (source_key, label, url) +# Each source_key becomes the `source` field in the DB. +FEEDS = [ + # Industry news + ("rss:techcrunch", "TechCrunch AI", + "https://techcrunch.com/category/artificial-intelligence/feed/"), + ("rss:venturebeat", "VentureBeat AI", + "https://venturebeat.com/category/ai/feed/"), + ("rss:theverge", "The Verge AI", + "https://www.theverge.com/rss/ai-artificial-intelligence/index.xml"), + ("rss:ainews", "AI News", + "https://www.artificialintelligence-news.com/feed/"), + # Analysis + ("rss:decoder", "The Decoder", + "https://www.the-decoder.com/feed/"), + ("rss:mittr", "MIT Tech Review AI", + "https://www.technologyreview.com/topic/artificial-intelligence/feed/"), + # Company blogs (primary signals for launches) + ("rss:openai", "OpenAI Blog", + "https://openai.com/blog/rss.xml"), + ("rss:anthropic", "Anthropic News", + "https://www.anthropic.com/rss/news.xml"), + ("rss:googleai", "Google AI Blog", + "https://blog.google/technology/rss.xml"), + ("rss:metaai", "Meta AI Blog", + "https://ai.meta.com/blog/rss.xml"), +] + +# AI relevance keywords for filtering — word-boundary matching +# to avoid false positives ("Great Britain" != "AI") +AI_KEYWORDS = [ + # Core AI terms + r"\bai\b", r"\bmachine learning\b", r"\bdeep learning\b", r"\bneural\b", + r"\bgenerative ai\b", r"\bgenerative\b", r"\bllm\b", r"\blarge language\b", + r"\bfoundation model\b", r"\btransformer\b", r"\baugmented\b", + # Agents & automation + r"\bagent\b", r"\bautonomous\b", r"\bmcp\b", r"\bfunction call\b", + r"\btool use\b", r"\brai\b", r"\bretrieval\b", + # Models & inference + r"\binference\b", r"\bmodel\b", r"\bembedding\b", r"\btoken\b", + r"\bfine-tune\b", r"\btraining\b", r"\bquantiz\b", + # Specific models (current landscape) + r"\bclaude\b", r"\bgpt\b", r"\bgemini\b", r"\bllama\b", r"\bdeepseek\b", + r"\bqwen\b", r"\bmistral\b", r"\bcodex\b", r"\bchatgpt\b", + r"\bsora\b", r"\bkimi\b", r"\bglm\b", r"\bminimax\b", + # Companies + r"\bopenai\b", r"\banthropic\b", r"\bgoogle\b", r"\bmeta\b", + r"\bmicrosoft\b", r"\bnvidia\b", r"\bamazon\b", r"\baws\b", + r"\bsamba(?=\w*nova\b)", r"\bcohere\b", r"\baider\b", +] + + +class RSSFeedsAdapter(SourceAdapter): + """RSS feed aggregator for commercial AI news.""" + + def name(self) -> str: + return "rss" + + def _is_ai_relevant(self, title: str, summary: str, tags: list) -> bool: + """Check if an entry is AI-relevant using word-boundary keyword matching.""" + text = (title + " " + summary).lower() + tag_text = " ".join(tags).lower() + combined = text + " " + tag_text + + for pattern in AI_KEYWORDS: + if re.search(pattern, combined): + return True + return False + + def _score(self, entry: dict, source_label: str, age_hours: float) -> float: + """Score: source authority + recency. Estimated (no native popularity metric). + + Source authority weights (based on AI signal quality): + - OpenAI/Anthropic/Google/Meta blogs: 1.5 (primary signals) + - TechCrunch/VentureBeat: 1.3 (industry news) + - The Verge/AI News/The Decoder/MIT TR: 1.0 (general coverage) + Recency decay: exponential half-life of 48 hours. + """ + # Source authority + authority = 1.0 + primary_blogs = ("openai", "anthropic", "googleai", "metaai") + industry_news = ("techcrunch", "venturebeat") + source_key = entry.get("source_key", "") + + if source_key in primary_blogs: + authority = 1.5 + elif source_key in industry_news: + authority = 1.3 + + # Recency decay: half-life = 48 hours + import math + recency = 2 ** (-age_hours / 48.0) # 1.0 at t=0, 0.5 at 48h, 0.25 at 96h + + # Tag bonus: more tags = more signal + tag_count = len(entry.get("tags", [])) + tag_bonus = min(tag_count * 0.1, 0.5) # max +0.5 + + score = (authority * recency * 4.0) + tag_bonus + return min(round(score, 2), 10.0) + + def _tags(self, entry: dict) -> list: + """Generate category tags from feed entry metadata.""" + tags = ["rss", entry.get("source_key", "rss:unknown")] + # Add feed-provided tags + for t in entry.get("tags", []): + clean = re.sub(r"[^a-z0-9-]", "-", t.lower()) + tags.append(f"tag:{clean}") + return tags + + def fetch(self, query: str = "", limit: int = 20) -> list[dict]: + """Fetch entries from all configured RSS feeds. + + Returns the top `limit` entries across all feeds, sorted by score. + Only AI-relevant entries are included. + """ + now = datetime.now(timezone.utc) + all_entries = [] + feed_failures = [] + + for source_key, label, url in FEEDS: + try: + d = feedparser.parse(url) + if d.status not in (200, 301, 302, 307, 308) or not d.entries: + feed_failures.append(f"{source_key}: status={d.status}, entries=0") + continue + + for entry in d.entries: + if not self._is_ai_relevant( + entry.get("title", ""), + entry.get("summary", ""), + [t.get("term", "") for t in entry.get("tags", [])], + ): + continue + + # Parse publish time — handle RFC 2822 (most RSS) and ISO 8601 + published = entry.get("published", entry.get("updated", "")) + age_hours = 999 # default: old + if published: + try: + pub_dt = parsedate_to_datetime(published) + except (ValueError, TypeError): + try: + pub_dt = datetime.fromisoformat(published.replace("Z", "+00:00")) + except (ValueError, TypeError): + pass + else: + age_hours = max((now - pub_dt).total_seconds() / 3600, 0.1) + + # Skip entries older than 14 days + if age_hours > 14 * 24: + continue + + tags = [t.get("term", "") for t in entry.get("tags", [])] + entry_dict = { + "source_key": source_key, + "source_label": label, + "age_hours": age_hours, + "tags": tags, + } + score = self._score(entry_dict, label, age_hours) + entry_dict["score"] = score + + all_entries.append({ + "source": "rss", + "source_id": f"{source_key}:{entry.get('id', entry.get('link', ''))[-30:]}", + "url": entry.get("link", ""), + "title": entry.get("title", ""), + "extracted_text": entry.get("summary", ""), + "summary": None, # LLM later + "category_tags": json.dumps(self._tags(entry_dict)), + "signal_score": score, + "raw_metadata": json.dumps({ + "source_key": source_key, + "source_label": label, + "author": entry.get("author", ""), + "published": published, + "age_hours": round(age_hours, 1), + "tags": tags, + "score_type": "estimated", + }), + "first_seen": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + "last_updated": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + }) + + time.sleep(0.5) # polite spacing + + except Exception as e: + feed_failures.append(f"{source_key}: {e}") + + if feed_failures: + print(f" ⚠ Feed failures: {', '.join(feed_failures)}") + + # Sort by score descending, take top N + all_entries.sort(key=lambda e: e["signal_score"], reverse=True) + return all_entries[:limit] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="RSS feeds adapter for Athena") + parser.add_argument("--limit", type=int, default=20, help="Max entries") + parser.add_argument("--dry-run", action="store_true", help="Don't store in DB") + parser.add_argument("--db", default=os.path.join(os.path.dirname(__file__), "..", "oracle.db")) + parser.add_argument("--schema", default=os.path.join(os.path.dirname(__file__), "..", "schema.sql")) + args = parser.parse_args() + + print(f"=== RSS Feeds Adapter ({len(FEEDS)} feeds) ===") + adapter = RSSFeedsAdapter() + entries = adapter.fetch(limit=args.limit * 2) # fetch extra for relevance filtering + entries = entries[:args.limit] + + print(f"\n Fetched {len(entries)} AI-relevant entries from {len(FEEDS)} feeds") + print() + + if entries: + print(f" {'Rank':<4} {'Score':<7} {'Age':<8} {'Source':<16} {'Title'}") + print(" " + "-" * 90) + for i, e in enumerate(entries[:15], 1): + meta = json.loads(e["raw_metadata"]) + print(f" {i:<4} {e['signal_score']:<7.2f} {meta['age_hours']:<8.1f}h {meta['source_label']:<16} {e['title'][:70]}") + + if not args.dry_run: + conn = sqlite3.connect(args.db) + if os.path.exists(args.schema): + with open(args.schema) as f: + conn.executescript(f.read()) + conn.commit() + + cur = conn.cursor() + stored = 0 + for entry in entries: + try: + cur.execute(""" + INSERT OR REPLACE INTO entries + (source, source_id, url, title, extracted_text, summary, + category_tags, signal_score, raw_metadata, first_seen, last_updated) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, ( + entry["source"], entry["source_id"], entry["url"], entry["title"], + entry["extracted_text"], entry["summary"], + entry["category_tags"], entry["signal_score"], + entry["raw_metadata"], entry["first_seen"], entry["last_updated"], + )) + stored += 1 + except Exception as e: + print(f" DB error: {e}") + conn.commit() + conn.close() + print(f"\n Stored {stored} entries") + + print(f"\n Done.") diff --git a/pipeline.py b/pipeline.py index da8d17d..7b1b0ec 100644 --- a/pipeline.py +++ b/pipeline.py @@ -33,10 +33,11 @@ ADAPTERS = { "reddit": lambda: __import__("adapters.reddit", fromlist=["RedditAdapter"]).RedditAdapter(), "hackernews": lambda: __import__("adapters.hackernews", fromlist=["HackerNewsAdapter"]).HackerNewsAdapter(), "huggingface": lambda: __import__("adapters.huggingface", fromlist=["HuggingFaceAdapter"]).HuggingFaceAdapter(), + "rss": lambda: __import__("adapters.rss_feeds", fromlist=["RSSFeedsAdapter"]).RSSFeedsAdapter(), } # Default enabled sources -ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface"] +ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface", "rss"] def init_db(db_path: str, schema_path: str) -> sqlite3.Connection: @@ -203,6 +204,21 @@ def verify_entries(conn: sqlite3.Connection, source: str, sample_size: int = 3): except Exception as e: print(f" ? [{eid}] {title[:60]}... verify failed: {e}") + elif src == "rss": + # For RSS, verify the article URL is reachable + if url: + try: + req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"}) + with urllib.request.urlopen(req, timeout=10) as resp: + status = resp.status + if status in (200, 301, 302): + print(f" ✓ [{eid}] {title[:60]}... URL reachable (HTTP {status})") + passed += 1 + else: + print(f" ? [{eid}] {title[:60]}... HTTP {status}") + except Exception as e: + print(f" ? [{eid}] {title[:60]}... verify failed: {e}") + time.sleep(0.5) # polite spacing return passed > 0