From 8017ded3bafb36d5c668cf6b3d1c9d75fb816576 Mon Sep 17 00:00:00 2001 From: Epictetus Date: Fri, 10 Jul 2026 16:34:05 +0000 Subject: [PATCH] fix(adapters): shared retry helper + run_log failure_class + enable RSS (issues #1 #2 #9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - adapters/__init__.py: add http_get() unified retry (429/5xx only, max 2 attempts, capped exp backoff) + AdapterHTTPError carrying failure_class; SourceAdapter.last_failure_class set on failure for pipeline capture. - arxiv/github/huggingface/hackernews/reddit: route HTTP through http_get. Preserves GitHub 403 rate-limit retry and Reddit 403/429 fast-bail. - schema.sql + pipeline.py: add run_log.failure_class column; rollup most- severe class across sources (5xx>4xx>429>error>zero_fetch>ok). - pipeline.py: ENABLE RSS in ENABLED_SOURCES (was registered, disabled). - RSS smoke test surfaced 3 broken feeds (anthropic 404, googleai 404, metaai 301) — left as-is, captured in feed_failures; URL fix is separate discovery task, not guessed. Verified: full dry-run fetches all 6 sources; github live fetch OK; Reddit 429 fast-bail preserved; no import/syntax errors. --- adapters/__init__.py | 85 ++++++++++++++++++++++++++++++++++++++ adapters/arxiv.py | 14 +++---- adapters/github.py | 57 ++++++++++---------------- adapters/hackernews.py | 31 ++++++-------- adapters/huggingface.py | 31 ++++++-------- adapters/reddit.py | 90 ++++++++++++++++++----------------------- pipeline.py | 35 ++++++++++++---- schema.sql | 3 ++ 8 files changed, 209 insertions(+), 137 deletions(-) diff --git a/adapters/__init__.py b/adapters/__init__.py index 317cdec..83ab6f3 100644 --- a/adapters/__init__.py +++ b/adapters/__init__.py @@ -1,11 +1,18 @@ """Source adapters for AI Research Oracle.""" +import urllib.request +import urllib.error +import time from abc import ABC, abstractmethod class SourceAdapter(ABC): """Base class for all ingestion adapters.""" + def __init__(self): + # Set by http_get when a request fails (issue #2 classification) + self.last_failure_class = None + @abstractmethod def name(self) -> str: """Source name: 'github', 'arxiv', 'reddit'.""" @@ -15,3 +22,81 @@ class SourceAdapter(ABC): def fetch(self, query: str = "", limit: int = 20) -> list[dict]: """Return entries matching DB schema fields.""" pass + + +class AdapterHTTPError(Exception): + """Raised by http_get on non-retryable or exhausted HTTP failures. + + failure_class is one of: '429', '5xx', '4xx', 'error'. + Consumed by pipeline.py to populate run_log.failure_class (issue #2). + """ + + def __init__(self, status, failure_class, message=""): + super().__init__(message) + self.status = status + self.failure_class = failure_class + + +def classify_http_status(code): + """Map an HTTP status code to a run_log failure_class.""" + if code == 429: + return "429" + if 500 <= code < 600: + return "5xx" + return "4xx" # 401/403/404 etc = client/config problem, not retried + + +def http_get(url, headers=None, timeout=15, max_retries=2, + backoff_base=2, retry_403_ratelimit=False, return_headers=False, + owner=None): + """GET with a unified retry policy shared by all adapters (issue #1). + + Retries ONLY on 429 and 5xx (transient). 4xx other than 429 are NOT + retried (config/client problems). Optional GitHub-style 403 rate-limit + handling via retry_403_ratelimit (waits for X-RateLimit-Reset header). + + Returns response body bytes (or (body, headers) tuple if return_headers). + Raises AdapterHTTPError on non-retryable or exhausted failures, carrying + .failure_class for run_log classification. If owner is provided, sets + owner.last_failure_class so the pipeline can record it (issue #2). + """ + def _fail(code, fclass, msg): + if owner is not None: + owner.last_failure_class = fclass + raise AdapterHTTPError(code, fclass, msg) + + req = urllib.request.Request(url, headers=headers or {}) + last_err = None + for attempt in range(max_retries + 1): + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + body = resp.read() + if return_headers: + return body, resp.headers + return body + except urllib.error.HTTPError as e: + last_err = e + code = e.code + if code == 429 or (code == 403 and retry_403_ratelimit): + if attempt < max_retries: + if code == 403: + reset = int(e.headers.get("X-RateLimit-Reset", 0)) + wait = (max(reset - int(time.time()), 0) + 1 + if reset else backoff_base * (attempt + 1) * 15) + time.sleep(min(wait, 300)) + else: + time.sleep(min(backoff_base * (attempt + 1), 30)) + continue + fclass = "429" if code == 429 else "4xx" + _fail(code, fclass, f"HTTP {code} for {url} (exhausted)") + if 500 <= code < 600: + if attempt < max_retries: + time.sleep(min(backoff_base * (attempt + 1), 30)) + continue + _fail(code, "5xx", f"HTTP {code} for {url} (exhausted)") + # other 4xx (401/403 w/o flag/404) — not retried + _fail(code, classify_http_status(code), f"HTTP {code} for {url}") + except Exception as e: # non-HTTP (timeout, DNS, conn reset) + _fail(None, "error", f"Request error: {e}") + # Defensive: loop should always raise or return + _fail(getattr(last_err, "code", None), "error", str(last_err)) diff --git a/adapters/arxiv.py b/adapters/arxiv.py index 2002d7a..1bca9d7 100644 --- a/adapters/arxiv.py +++ b/adapters/arxiv.py @@ -32,7 +32,7 @@ import xml.etree.ElementTree as ET from datetime import datetime, timedelta, timezone from html import unescape -from adapters import SourceAdapter +from adapters import SourceAdapter, http_get, AdapterHTTPError # arXiv API ARXIV_API = "http://export.arxiv.org/api/query" @@ -151,14 +151,12 @@ class ArxivAdapter(SourceAdapter): f"&max_results={max_results}" ) - req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"}) - try: - with urllib.request.urlopen(req, timeout=30) as resp: - xml_data = resp.read().decode("utf-8") - return self._parse_atom(xml_data) - except urllib.error.HTTPError as e: - print(f" HTTP {e.code} for arXiv query") + raw = http_get(url, headers={"User-Agent": "ai-oracle/0.1"}, + timeout=30, max_retries=2, owner=self) + return self._parse_atom(raw.decode("utf-8")) + except AdapterHTTPError as e: + print(f" {e.failure_class}: arXiv query ({e})") return [] except Exception as e: print(f" arXiv request error: {e}") diff --git a/adapters/github.py b/adapters/github.py index dda90a5..8d13f4e 100644 --- a/adapters/github.py +++ b/adapters/github.py @@ -17,7 +17,7 @@ import urllib.error import urllib.parse from datetime import datetime, timedelta, timezone -from adapters import SourceAdapter +from adapters import SourceAdapter, http_get, AdapterHTTPError class GitHubAdapter(SourceAdapter): @@ -43,40 +43,27 @@ class GitHubAdapter(SourceAdapter): return headers def _request(self, url: str, max_retries: int = 2) -> dict | list | None: - """Make a GET request with retry on 403 (rate limit).""" - req = urllib.request.Request(url, headers=self._headers()) - - for attempt in range(max_retries + 1): - try: - with urllib.request.urlopen(req, timeout=15) as resp: - data = json.loads(resp.read().decode("utf-8")) - - # Check rate limit headers - remaining = int(resp.headers.get("X-RateLimit-Remaining", 0)) - if remaining <= 5: - print(f" ⚠ Rate limit low ({remaining} remaining), stopping") - break - - return data - except urllib.error.HTTPError as e: - if e.code == 403: - # Rate limited — reset time is in headers - reset = int(e.headers.get("X-RateLimit-Reset", 0)) - if reset: - wait = max(reset - int(time.time()), 0) + 1 - print(f" ⚠ Rate limited, wait {wait}s") - else: - wait = 30 * (attempt + 1) - print(f" 403 on attempt {attempt + 1}, retry in {wait}s") - time.sleep(min(wait, 300)) # cap at 5 min - continue - print(f" HTTP {e.code} for {url}") - return None - except Exception as e: - print(f" Request error: {e}") - return None - - return None + """GET via shared retry helper; 403 rate-limit handled as transient.""" + try: + raw, headers = http_get( + url, headers=self._headers(), timeout=15, + max_retries=max_retries, retry_403_ratelimit=True, + return_headers=True, owner=self) + except AdapterHTTPError as e: + print(f" {e.failure_class}: GitHub {url}") + return None + # Informational: flag if we're close to the unauth rate ceiling + try: + remaining = int(headers.get("X-RateLimit-Remaining", 0)) + if remaining <= 5: + print(f" ⚠ Rate limit low ({remaining} remaining)") + except Exception: + pass + try: + return json.loads(raw.decode("utf-8")) + except Exception as e: + print(f" GitHub decode error: {e}") + return None def _search_repos(self, query: str, sort: str = "stars", order: str = "desc", per_page: int = 30) -> list: """Search repositories via GitHub API.""" diff --git a/adapters/hackernews.py b/adapters/hackernews.py index 6680c6b..f5935a1 100644 --- a/adapters/hackernews.py +++ b/adapters/hackernews.py @@ -20,7 +20,7 @@ import urllib.request import urllib.error from datetime import datetime, timezone -from adapters import SourceAdapter +from adapters import SourceAdapter, http_get, AdapterHTTPError class HackerNewsAdapter(SourceAdapter): @@ -54,24 +54,19 @@ class HackerNewsAdapter(SourceAdapter): return "hackernews" def _request(self, path: str, max_retries: int = 2) -> dict | list | None: - """Make a GET request to the HN Firebase API.""" + """GET via shared retry helper (retries 429/5xx).""" url = f"{self.BASE}{path}" - req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) - - for attempt in range(max_retries + 1): - try: - with urllib.request.urlopen(req, timeout=15) as resp: - return json.loads(resp.read().decode("utf-8")) - except (urllib.error.HTTPError, urllib.error.URLError) as e: - if attempt < max_retries: - time.sleep(3 * (attempt + 1)) - continue - print(f" HTTP error: {e}") - return None - except Exception as e: - print(f" Request error: {e}") - return None - return None + try: + raw = http_get(url, headers={"User-Agent": self.user_agent}, + timeout=15, max_retries=max_retries, owner=self) + except AdapterHTTPError as e: + print(f" {e.failure_class}: HN {path}") + return None + try: + return json.loads(raw.decode("utf-8")) + except Exception as e: + print(f" HN decode error: {e}") + return None def _is_ai_relevant(self, title: str) -> bool: """Check if a story title is AI/ML relevant. diff --git a/adapters/huggingface.py b/adapters/huggingface.py index 76c4746..b2251ed 100644 --- a/adapters/huggingface.py +++ b/adapters/huggingface.py @@ -34,7 +34,7 @@ import urllib.request import urllib.error from datetime import datetime, timedelta, timezone -from adapters import SourceAdapter +from adapters import SourceAdapter, http_get, AdapterHTTPError class HuggingFaceAdapter(SourceAdapter): @@ -84,24 +84,19 @@ class HuggingFaceAdapter(SourceAdapter): return headers def _request(self, path: str, max_retries: int = 2) -> list | dict | None: - """Make a GET request to the HF API.""" + """GET via shared retry helper (retries 429/5xx).""" url = f"{self.BASE}{path}" - req = urllib.request.Request(url, headers=self._headers()) - - for attempt in range(max_retries + 1): - try: - with urllib.request.urlopen(req, timeout=20) as resp: - return json.loads(resp.read().decode("utf-8")) - except (urllib.error.HTTPError, urllib.error.URLError) as e: - if attempt < max_retries: - time.sleep(3 * (attempt + 1)) - continue - print(f" HF API error: {e}") - return None - except Exception as e: - print(f" Request error: {e}") - return None - return None + try: + raw = http_get(url, headers=self._headers(), timeout=20, + max_retries=max_retries, owner=self) + except AdapterHTTPError as e: + print(f" {e.failure_class}: HF {path}") + return None + try: + return json.loads(raw.decode("utf-8")) + except Exception as e: + print(f" HF decode error: {e}") + return None def _is_ai_relevant(self, model: dict) -> bool: """Check if a model/dataset is AI/ML relevant. diff --git a/adapters/reddit.py b/adapters/reddit.py index 2f3815d..3faa8f7 100644 --- a/adapters/reddit.py +++ b/adapters/reddit.py @@ -24,7 +24,7 @@ import xml.etree.ElementTree as ET from datetime import datetime, timezone from html import unescape -from adapters import SourceAdapter +from adapters import SourceAdapter, http_get, AdapterHTTPError class RedditAdapter(SourceAdapter): @@ -101,33 +101,27 @@ class RedditAdapter(SourceAdapter): return False def _fetch_rss(self, subreddit: str) -> list[dict]: - """Fetch RSS feed for a subreddit.""" - url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" - req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) + """Fetch RSS feed for a subreddit (shared retry helper). - for attempt in range(2): # max 2 attempts, fail fast - try: - with urllib.request.urlopen(req, timeout=10) as resp: - xml_data = resp.read().decode("utf-8") - break - except urllib.error.HTTPError as e: - if e.code in (403,): - print(f" RSS blocked (HTTP {e.code}) for r/{subreddit}") - return [] - if e.code == 429: - if attempt == 0: - time.sleep(2) # single retry with short backoff - continue - print(f" RSS rate-limited for r/{subreddit}, skip") - return [] - print(f" RSS HTTP {e.code} for r/{subreddit}") - return [] - except Exception as e: - print(f" RSS error r/{subreddit}: {e}") - return [] - else: - print(f" r/{subreddit}: still rate limited, skip") + Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s + retry then []; 5xx -> helper retry then []. Not slower than before. + """ + url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" + try: + raw = http_get(url, headers={"User-Agent": self.user_agent}, + timeout=10, max_retries=1, backoff_base=2, owner=self) + except AdapterHTTPError as e: + if e.status == 403: + print(f" RSS blocked (HTTP 403) for r/{subreddit}") + elif e.status == 429: + print(f" RSS rate-limited for r/{subreddit}, skip") + else: + print(f" RSS {e.failure_class} for r/{subreddit}") return [] + except Exception as e: + print(f" RSS error r/{subreddit}: {e}") + return [] + xml_data = raw.decode("utf-8") # Parse Atom XML entries = [] @@ -303,33 +297,27 @@ class RedditAdapter(SourceAdapter): return tags def _fetch_rss(self, subreddit: str) -> list[dict]: - """Fetch RSS feed for a subreddit.""" - url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" - req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) + """Fetch RSS feed for a subreddit (shared retry helper). - for attempt in range(2): # max 2 attempts, fail fast - try: - with urllib.request.urlopen(req, timeout=10) as resp: - xml_data = resp.read().decode("utf-8") - break - except urllib.error.HTTPError as e: - if e.code in (403,): - print(f" RSS blocked (HTTP {e.code}) for r/{subreddit}") - return [] - if e.code == 429: - if attempt == 0: - time.sleep(2) # single retry with short backoff - continue - print(f" RSS rate-limited for r/{subreddit}, skip") - return [] - print(f" RSS HTTP {e.code} for r/{subreddit}") - return [] - except Exception as e: - print(f" RSS error r/{subreddit}: {e}") - return [] - else: - print(f" r/{subreddit}: still rate limited, skip") + Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s + retry then []; 5xx -> helper retry then []. Not slower than before. + """ + url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" + try: + raw = http_get(url, headers={"User-Agent": self.user_agent}, + timeout=10, max_retries=1, backoff_base=2, owner=self) + except AdapterHTTPError as e: + if e.status == 403: + print(f" RSS blocked (HTTP 403) for r/{subreddit}") + elif e.status == 429: + print(f" RSS rate-limited for r/{subreddit}, skip") + else: + print(f" RSS {e.failure_class} for r/{subreddit}") return [] + except Exception as e: + print(f" RSS error r/{subreddit}: {e}") + return [] + xml_data = raw.decode("utf-8") # Parse Atom XML entries = [] diff --git a/pipeline.py b/pipeline.py index c7bdd5f..f0fe8fd 100644 --- a/pipeline.py +++ b/pipeline.py @@ -37,7 +37,7 @@ ADAPTERS = { } # 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: @@ -252,9 +252,14 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo entries = adapter.fetch(limit=limit) except Exception as e: print(f" ✗ {source_name} failed: {e}") - source_stats[source_name] = {"fetched": 0, "stored": 0, "error": str(e)} + source_stats[source_name] = {"fetched": 0, "stored": 0, + "error": str(e), + "failure_class": "error"} continue + # Capture classification from the adapter (set by http_get on failure) + fc = getattr(adapter, "last_failure_class", None) + # Add adapter_version to metadata for entry in entries: meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"] @@ -262,7 +267,8 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo entry["raw_metadata"] = json.dumps(meta) all_entries.extend(entries) - source_stats[source_name] = {"fetched": len(entries), "stored": 0} + source_stats[source_name] = {"fetched": len(entries), "stored": 0, + "failure_class": fc or "ok"} print(f" Fetched: {len(entries)} entries") # Small spacing between sources @@ -294,16 +300,31 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo # Zero-fetch (e.g. Reddit fully rate-limited) raises no exception but # is still a degraded run — record it so run_log can tell # "intermittent vs consistently-broken" apart over time. - zero = [s for s, st in source_stats.items() if st.get("fetched", 0) == 0 and not st.get("error")] + zero = [s for s, st in source_stats.items() + if st.get("fetched", 0) == 0 and not st.get("error")] notes_parts = [f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")] if zero: notes_parts.append(f"no-fetch (degraded): {', '.join(zero)}") notes = "; ".join(notes_parts) or "all sources ok" + + # Rollup failure_class (issue #2): most severe across sources. + # Priority: 5xx > 4xx > 429 > error > zero_fetch > ok + rank = {"5xx": 5, "4xx": 4, "429": 3, "error": 2, "zero_fetch": 1, "ok": 0} + classes = [st.get("failure_class", "ok") for st in source_stats.values()] + if any(c in ("5xx", "4xx", "429", "error") for c in classes): + run_fc = max((c for c in classes if c in rank), + key=lambda c: rank[c]) + elif zero: + run_fc = "zero_fetch" + else: + run_fc = "ok" try: conn.execute(""" - INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes) - VALUES (?, ?, ?, ?, ?) - """, (len(all_entries), stored, json.dumps(ok), json.dumps(failed), notes)) + INSERT INTO run_log (total_fetched, total_stored, sources_ok, + sources_failed, failure_class, notes) + VALUES (?, ?, ?, ?, ?, ?) + """, (len(all_entries), stored, json.dumps(ok), json.dumps(failed), + run_fc, notes)) conn.commit() except Exception as e: print(f" ⚠ run_log write failed: {e}") diff --git a/schema.sql b/schema.sql index 623ec96..700db4d 100644 --- a/schema.sql +++ b/schema.sql @@ -22,6 +22,8 @@ CREATE INDEX IF NOT EXISTS idx_entries_category ON entries(category_tags); -- Partial failures (e.g. Reddit rate-limited) are detectable here, not hidden -- as a "complete" run. Also enables future pruning decisions (entries older -- than N days with no re-fetch can be archived). +-- failure_class (issue #2): one of 4xx / 5xx / 429 / zero_fetch / ok, derived +-- from the real HTTP response via adapters.http_get, not guessed after the fact. CREATE TABLE IF NOT EXISTS run_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_time TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), @@ -29,6 +31,7 @@ CREATE TABLE IF NOT EXISTS run_log ( total_stored INTEGER DEFAULT 0, sources_ok TEXT, -- JSON list of sources that succeeded sources_failed TEXT, -- JSON list of sources that errored/skipped + failure_class TEXT, -- 4xx / 5xx / 429 / zero_fetch / ok notes TEXT );