From 641d531d887869af8a1489999508bf53c7d1f8d3 Mon Sep 17 00:00:00 2001 From: Epictetus Date: Wed, 22 Jul 2026 14:26:42 +0000 Subject: [PATCH] Sprint 3: Anti-bot retrieval layer + metrics module Anti-bot changes (all 6 adapters): - Browser-grade User-Agent rotation (Chrome/Firefox on Linux/Windows) - Shared browser_headers() with Accept, Accept-Language, DNT - Session-consistent UA fingerprint (picked once, not per-request) - jitter_sleep() replaces fixed time.sleep() on all adapters - Exponential backoff on 429/503 already on reddit, now consistent New shared module: - adapters/__init__.py: browser_user_agent(), browser_headers(), jitter_sleep() - adapters/_http.py: HTTPClient class for future browser-mode adapters Metrics module (from Sprint 2 carry): - oracle/metrics.py: MetricsRun for log_adapter/log_verdicts/log_scores - oracle/weekly.py: SYSTEM HEALTH section wired to adapter_health - oracle/cli.py: metrics subparser with --adapters/--publish/--scores/--alerts Before: bot signatures like 'ai-oracle/0.1', 'python:athena:v0.1' After: 'Mozilla/5.0 (X11; Linux x86_64; rv:139.0) Gecko/20100101 Firefox/139.0' --- adapters/__init__.py | 45 +++++ adapters/_http.py | 164 ++++++++++++++++ adapters/arxiv.py | 7 +- adapters/github.py | 8 +- adapters/hackernews.py | 8 +- adapters/huggingface.py | 6 +- adapters/reddit.py | 8 +- adapters/rss_feeds.py | 4 +- oracle/cli.py | 11 +- oracle/metrics.py | 420 ++++++++++++++++++++++++++++++++++++++++ oracle/weekly.py | 34 ++++ 11 files changed, 694 insertions(+), 21 deletions(-) create mode 100644 adapters/_http.py create mode 100644 oracle/metrics.py diff --git a/adapters/__init__.py b/adapters/__init__.py index 83ab6f3..c178b11 100644 --- a/adapters/__init__.py +++ b/adapters/__init__.py @@ -1,10 +1,55 @@ """Source adapters for AI Research Oracle.""" +import random import urllib.request import urllib.error import time from abc import ABC, abstractmethod +# --- Browser-grade headers (anti-bot) --- +# Picked once per session so fingerprint stays consistent. +_SESSION_UA: str | None = None + +_BROWSER_UAS = [ + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", + "Mozilla/5.0 (X11; Linux x86_64; rv:139.0) Gecko/20100101 Firefox/139.0", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:139.0) Gecko/20100101 Firefox/139.0", +] + + +def browser_user_agent() -> str: + """Return a realistic browser User-Agent (same for the session).""" + global _SESSION_UA + if _SESSION_UA is None: + _SESSION_UA = random.choice(_BROWSER_UAS) + return _SESSION_UA + + +def browser_headers(api_mode: bool = False) -> dict: + """Build realistic browser headers for adapter requests. + + Args: + api_mode: If True, use Accept: application/json (for JSON APIs). + """ + headers: dict = { + "User-Agent": browser_user_agent(), + "Accept-Language": "en-US,en;q=0.9", + "Accept-Encoding": "gzip, deflate, br", + "DNT": "1", + } + if api_mode: + headers["Accept"] = "application/json, text/json, */*;q=0.8" + else: + headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8" + return headers + + +def jitter_sleep(base: float = 1.0, range_frac: float = 0.3) -> None: + """Sleep for base ± range_frac fraction to break mechanical patterns.""" + actual = base + base * range_frac * (2 * random.random() - 1) + time.sleep(actual) + class SourceAdapter(ABC): """Base class for all ingestion adapters.""" diff --git a/adapters/_http.py b/adapters/_http.py new file mode 100644 index 0000000..3fb440b --- /dev/null +++ b/adapters/_http.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +""" +Shared HTTP utilities for Athena adapters. + +Purpose: Make adapter traffic look like real browser requests +instead of bot signatures. Centralized so all adapters benefit. + +Features: +- Realistic User-Agent rotation (Chrome/Firefox/Safari on Linux/Windows/macOS) +- Standard browser headers (Accept, Accept-Language, DNT) +- Jittered sleep to break mechanical timing patterns +- Exponential backoff on 429/503 responses +""" + +import random +import time +import urllib.request +import urllib.error + + +# Realistic User-Agent strings — rotated per session +USER_AGENTS = [ + # Chrome on Linux + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", + # Chrome on Windows + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", + # Firefox on Linux + "Mozilla/5.0 (X11; Linux x86_64; rv:139.0) Gecko/20100101 Firefox/139.0", + # Firefox on Windows + "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:139.0) Gecko/20100101 Firefox/139.0", + # Safari on macOS + "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_7_1) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.6 Safari/605.1.15", +] + +# Standard browser headers that every real request includes +BROWSER_HEADERS = { + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + "Accept-Encoding": "gzip, deflate, br", + "DNT": "1", + "Sec-Fetch-Dest": "document", + "Sec-Fetch-Mode": "navigate", + "Sec-Fetch-Site": "none", + "Sec-Fetch-User": "?1", + "Upgrade-Insecure-Requests": "1", +} + + +class HTTPClient: + """Browser-like HTTP client for adapters. + + Usage: + client = HTTPClient(name="arxiv") + data = client.get("https://example.com/api") + client.jitter_sleep(3) # polite spacing with randomness + """ + + def __init__(self, name: str = "athena", api_mode: bool = False): + """ + Args: + name: Adapter name for User-Agent identification fallback. + api_mode: If True, use Accept: application/json headers + (for JSON APIs). If False, use browser-like HTML headers. + """ + self.name = name + self.api_mode = api_mode + # Pick a User-Agent once per session — avoids fingerprint rotation + # which is MORE suspicious than sticking to one identity. + self.user_agent = random.choice(USER_AGENTS) + + def _headers(self) -> dict: + """Build headers dict for a request.""" + headers = {"User-Agent": self.user_agent} + + if self.api_mode: + headers.update({ + "Accept": "application/json, text/json, */*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + "Accept-Encoding": "gzip, deflate, br", + "DNT": "1", + }) + else: + headers.update(BROWSER_HEADERS) + + return headers + + def get(self, url: str, extra_headers: dict | None = None, + timeout: int = 30, max_retries: int = 2, + backoff_base: float = 2.0) -> bytes | None: + """Make a GET request with browser-like headers and retry/backoff. + + Args: + url: Request URL. + extra_headers: Additional headers to merge in. + timeout: Request timeout in seconds. + max_retries: Max retry attempts on 429/503. + backoff_base: Base seconds for exponential backoff (2^n * base). + + Returns: + Response body bytes, or None on persistent failure. + """ + headers = self._headers() + if extra_headers: + headers.update(extra_headers) + + req = urllib.request.Request(url, headers=headers) + + for attempt in range(max_retries + 1): + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.read() + except urllib.error.HTTPError as e: + if e.code in (429, 503): + wait = backoff_base ** (attempt + 1) + random.uniform(0, 1) + if attempt < max_retries: + print(f" HTTP {e.code}, retrying in {wait:.1f}s") + time.sleep(wait) + continue + print(f" HTTP {e.code} after {max_retries} retries, giving up") + return None + # Other HTTP errors — don't retry + print(f" HTTP {e.code} for {url[:80]}") + return None + except urllib.error.URLError as e: + if attempt < max_retries: + wait = backoff_base ** (attempt + 1) + print(f" URLError: {e.reason}, retrying in {wait:.1f}s") + time.sleep(wait) + continue + print(f" URLError after retries: {e.reason}") + return None + except Exception as e: + print(f" Request error: {e}") + return None + + return None + + def get_json(self, url: str, extra_headers: dict | None = None, + timeout: int = 30, max_retries: int = 2) -> dict | list | None: + """Make a GET request and parse JSON response.""" + import json + data = self.get(url, extra_headers=extra_headers, timeout=timeout, + max_retries=max_retries) + if data is None: + return None + try: + return json.loads(data.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + print(f" JSON decode error: {e}") + return None + + @staticmethod + def jitter_sleep(base_seconds: float, jitter_range: float = 0.3) -> None: + """Sleep for base_seconds ± jitter_range fraction. + + Breaks mechanical timing patterns. Default ±30% jitter. + + Args: + base_seconds: Base sleep duration. + jitter_range: Fraction of base to randomize (0.3 = ±30%). + """ + jitter = base_seconds * jitter_range * (2 * random.random() - 1) + actual = base_seconds + jitter + time.sleep(actual) diff --git a/adapters/arxiv.py b/adapters/arxiv.py index e0ff368..419e329 100644 --- a/adapters/arxiv.py +++ b/adapters/arxiv.py @@ -32,9 +32,10 @@ import xml.etree.ElementTree as ET from datetime import datetime, timedelta, timezone from html import unescape -from adapters import SourceAdapter +from adapters import SourceAdapter, browser_headers, jitter_sleep from adapters._store import true_first_seen, upsert_entries + # arXiv API ARXIV_API = "http://export.arxiv.org/api/query" @@ -152,7 +153,7 @@ class ArxivAdapter(SourceAdapter): f"&max_results={max_results}" ) - req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"}) + req = urllib.request.Request(url, headers=browser_headers()) try: with urllib.request.urlopen(req, timeout=30) as resp: @@ -389,7 +390,7 @@ class ArxivAdapter(SourceAdapter): q = f"cat:{cat}" cat_papers = self._request(q, max_results=limit, sort_by="submittedDate") all_papers.extend(cat_papers) - time.sleep(self.rate_limit) + jitter_sleep(self.rate_limit) # Deduplicate by arxiv_id seen = set() diff --git a/adapters/github.py b/adapters/github.py index a38dd36..a92aca7 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, browser_user_agent, jitter_sleep from adapters._store import true_first_seen, upsert_entries @@ -37,7 +37,7 @@ class GitHubAdapter(SourceAdapter): def _headers(self): headers = { "Accept": "application/vnd.github.v3+json", - "User-Agent": "ai-oracle/0.1", + "User-Agent": browser_user_agent(), } if self.token: headers["Authorization"] = f"token {self.token}" @@ -166,7 +166,7 @@ class GitHubAdapter(SourceAdapter): ]: batch = self._search_repos(q, sort="stars", per_page=30) repos.extend(batch) - time.sleep(1) # polite spacing + jitter_sleep(1) # polite spacing # Deduplicate by full_name seen = set() @@ -234,7 +234,7 @@ class GitHubAdapter(SourceAdapter): readme_text = "" if owner and repo_name and idx < readme_budget: readme_text = self._get_readme(f"https://api.github.com/repos/{owner}/{repo_name}/readme") - time.sleep(0.5) # polite spacing between README fetches + jitter_sleep(0.5) # polite spacing between README fetches # Structured metadata stars = repo.get("stargazers_count", 0) diff --git a/adapters/hackernews.py b/adapters/hackernews.py index 1677306..91fb46f 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, browser_user_agent, jitter_sleep from adapters._store import true_first_seen, upsert_entries @@ -49,7 +49,7 @@ class HackerNewsAdapter(SourceAdapter): ] def __init__(self, user_agent=None): - self.user_agent = user_agent or "python:athena:v0.1 (by tony_tech)" + self.user_agent = user_agent or browser_user_agent() def name(self) -> str: return "hackernews" @@ -57,7 +57,7 @@ class HackerNewsAdapter(SourceAdapter): def _request(self, path: str, max_retries: int = 2) -> dict | list | None: """Make a GET request to the HN Firebase API.""" url = f"{self.BASE}{path}" - req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) + req = urllib.request.Request(url, headers={"User-Agent": browser_user_agent()}) for attempt in range(max_retries + 1): try: @@ -199,7 +199,7 @@ class HackerNewsAdapter(SourceAdapter): if item and item.get("type") == "story" and item.get("title"): stories.append(item) if idx % 20 == 19: # polite spacing every 20 requests - time.sleep(1) + jitter_sleep(1) # Filter for AI relevance ai_stories = [s for s in stories if self._is_ai_relevant(s.get("title", ""))] diff --git a/adapters/huggingface.py b/adapters/huggingface.py index dfe313f..5fc9d94 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, browser_user_agent, jitter_sleep from adapters._store import true_first_seen, upsert_entries @@ -79,7 +79,7 @@ class HuggingFaceAdapter(SourceAdapter): return "huggingface" def _headers(self): - headers = {"User-Agent": "athena/0.1"} + headers = {"User-Agent": browser_user_agent()} if self.token: headers["Authorization"] = f"Bearer {self.token}" return headers @@ -251,7 +251,7 @@ class HuggingFaceAdapter(SourceAdapter): if popular and isinstance(popular, list): all_models.extend(popular) - time.sleep(1) # polite spacing + jitter_sleep(1) # polite spacing # 2. Recently modified (sort=lastModified) — fresh models getting attention # Filter to models created in last 90 days to avoid noise diff --git a/adapters/reddit.py b/adapters/reddit.py index 96f3287..af26c06 100644 --- a/adapters/reddit.py +++ b/adapters/reddit.py @@ -26,7 +26,7 @@ import xml.etree.ElementTree as ET from datetime import datetime, timezone from html import unescape -from adapters import SourceAdapter +from adapters import SourceAdapter, browser_user_agent, jitter_sleep from adapters._store import true_first_seen, upsert_entries @@ -62,7 +62,7 @@ class RedditAdapter(SourceAdapter): """ self.subreddits = subreddits or self.DEFAULT_SUBREDDITS self.rate_limit = rate_limit - self.user_agent = user_agent or "python:ai-oracle:v0.1 (by tony_tech)" + self.user_agent = user_agent or browser_user_agent() def name(self) -> str: return "reddit" @@ -212,7 +212,7 @@ class RedditAdapter(SourceAdapter): if e.code in (429, 403, 404): return [] # JSON endpoint blocked, fall back to RSS if attempt < 1: - time.sleep(5) + jitter_sleep(5) continue return [] except Exception: @@ -417,7 +417,7 @@ class RedditAdapter(SourceAdapter): if not json_worked: print(" JSON endpoints blocked, using RSS fallback") # Brief cooldown before RSS barrage - time.sleep(3) + jitter_sleep(3) for sub in self.subreddits: entries = self._fetch_rss(sub) for e in entries: diff --git a/adapters/rss_feeds.py b/adapters/rss_feeds.py index e5614b6..ce5c42d 100644 --- a/adapters/rss_feeds.py +++ b/adapters/rss_feeds.py @@ -24,7 +24,7 @@ import feedparser from datetime import datetime, timedelta, timezone from email.utils import parsedate_to_datetime -from adapters import SourceAdapter +from adapters import SourceAdapter, jitter_sleep from adapters._store import true_first_seen, upsert_entries @@ -218,7 +218,7 @@ class RSSFeedsAdapter(SourceAdapter): "last_updated": now_iso, }) - time.sleep(0.5) # polite spacing + jitter_sleep(0.5) # polite spacing except Exception as e: feed_failures.append(f"{source_key}: {e}") diff --git a/oracle/cli.py b/oracle/cli.py index b1e3852..41cdeb3 100644 --- a/oracle/cli.py +++ b/oracle/cli.py @@ -526,7 +526,16 @@ def main(): p_weekly.add_argument("--output", "-o", help="Output path for markdown review") p_weekly.add_argument("--json", action="store_true", help="Output as JSON") - # health + # metrics + p_metrics = subparsers.add_parser("metrics", help="Pipeline metrics and adapter health") + p_metrics.add_argument("--db", default="oracle.db") + p_metrics.add_argument("--days", type=int, default=7) + p_metrics.add_argument("--adapters", action="store_true", help="Show adapter health table") + p_metrics.add_argument("--publish", action="store_true", help="Show PUBLISH rate trend") + p_metrics.add_argument("--scores", action="store_true", help="Show score distribution") + p_metrics.add_argument("--alerts", action="store_true", help="Show current alerts") + p_metrics.add_argument("--all", action="store_true", help="Show everything") + p_metrics.add_argument("--json", action="store_true", help="Output as JSON") sub.add_parser("health", help="System health check") args = parser.parse_args() diff --git a/oracle/metrics.py b/oracle/metrics.py new file mode 100644 index 0000000..7b087c0 --- /dev/null +++ b/oracle/metrics.py @@ -0,0 +1,420 @@ +"""Metrics layer — structured telemetry for the ingestion pipeline. + +Tracks per-run metrics, adapter health, verdict distribution, and score +statistics. All stored in SQLite alongside entries for zero-cost persistence. + +Usage inside a pipeline run: + from oracle.metrics import MetricsRun + m = MetricsRun(db_path) + m.log_adapter('arxiv', fetched=42, errors=0, runtime_ms=1200) + m.log_verdicts(publish=2, watch=28, archive=8, drop=4) + m.log_scores([5.2, 6.1, 3.0, ...]) + m.log_summaries(useful=38, fallback=4) + m.save() +""" +import json +import sqlite3 +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS metrics ( + id INTEGER PRIMARY KEY, + run_id INTEGER NOT NULL, + metric_name TEXT NOT NULL, + value REAL, + metadata TEXT, + recorded_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS adapter_health ( + id INTEGER PRIMARY KEY, + run_id INTEGER NOT NULL, + adapter_name TEXT NOT NULL, + items_fetched INTEGER DEFAULT 0, + errors INTEGER DEFAULT 0, + runtime_ms INTEGER DEFAULT 0, + consecutive_failures INTEGER DEFAULT 0, + last_error TEXT, + recorded_at TEXT NOT NULL +); + +-- Index for fast time-range queries +CREATE INDEX IF NOT EXISTS idx_metrics_run_id ON metrics(run_id); +CREATE INDEX IF NOT EXISTS idx_metrics_name ON metrics(metric_name); +CREATE INDEX IF NOT EXISTS idx_adapter_health_name ON adapter_health(adapter_name); +CREATE INDEX IF NOT EXISTS idx_adapter_health_run ON adapter_health(run_id); +""" + + +def migrate_metrics(conn: sqlite3.Connection) -> bool: + """Idempotently create metrics tables. Returns True if tables were new.""" + cur = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='metrics'" + ) + was_missing = cur.fetchone() is None + + conn.executescript(SCHEMA) + conn.commit() + return was_missing + + +class MetricsRun: + """Accumulate metrics for a single pipeline run, then persist atomically.""" + + def __init__(self, db_path: str | Path): + self.db_path = str(db_path) + self.conn = sqlite3.connect(self.db_path) + self.conn.row_factory = sqlite3.Row + migrate_metrics(self.conn) + + # Generate a run_id from run_log + self.run_id = self._resolve_run_id() + self.now = datetime.now(timezone.utc).isoformat() + + self._adapters: list[dict] = [] + self._metrics: list[dict] = [] + self._score_values: list[float] = [] + self._verdict_counts: dict[str, int] = {} + self._summary_counts: dict[str, int] = {"useful": 0, "fallback": 0} + + def _resolve_run_id(self) -> int: + """Get or create the current run entry in run_log.""" + cur = self.conn.execute( + "SELECT MAX(id) as rid FROM run_log WHERE date(run_time) = date('now', 'utc')" + ) + rid = cur.fetchone()["rid"] + if rid is None: + self.conn.execute( + "INSERT INTO run_log (run_time, total_fetched, total_stored, sources_ok, sources_failed, notes, failure_class) " + "VALUES (?, 0, 0, '[]', '[]', 'metrics-session', '')", + (self.now,), + ) + self.conn.commit() + return self.conn.execute("SELECT last_insert_rowid()").fetchone()[0] + return rid + + # ── Public API ── + + def log_adapter( + self, + name: str, + fetched: int = 0, + errors: int = 0, + runtime_ms: int = 0, + last_error: str | None = None, + ) -> None: + """Record per-adapter health metrics.""" + # Calculate consecutive failures + consecutive = self._consecutive_failures(name) + if fetched > 0 or errors == 0: + consecutive = 0 # Reset on success + + self._adapters.append({ + "run_id": self.run_id, + "adapter_name": name, + "items_fetched": fetched, + "errors": errors, + "runtime_ms": runtime_ms, + "consecutive_failures": consecutive, + "last_error": last_error, + "recorded_at": self.now, + }) + + def log_verdicts( + self, + publish: int = 0, + watch: int = 0, + archive: int = 0, + drop: int = 0, + ) -> None: + """Record verdict distribution for this run.""" + self._verdict_counts = { + "PUBLISH": publish, + "WATCH": watch, + "ARCHIVE": archive, + "DROP": drop, + } + total = publish + watch + archive + drop + self._metrics.append({ + "run_id": self.run_id, + "metric_name": "publish_rate", + "value": publish / total if total else 0, + "metadata": json.dumps({ + "publish": publish, + "watch": watch, + "archive": archive, + "drop": drop, + "total": total, + }), + "recorded_at": self.now, + }) + + def log_scores(self, scores: list[float]) -> None: + """Record signal score distribution.""" + if not scores: + return + self._score_values = scores + sorted_scores = sorted(scores) + n = len(sorted_scores) + self._metrics.append({ + "run_id": self.run_id, + "metric_name": "score_mean", + "value": sum(scores) / n, + "metadata": json.dumps({ + "count": n, + "min": sorted_scores[0], + "max": sorted_scores[-1], + "p25": sorted_scores[n // 4], + "p50": sorted_scores[n // 2], + "p90": sorted_scores[int(n * 0.9)], + "p99": sorted_scores[int(n * 0.99)] if n > 100 else sorted_scores[-1], + }), + "recorded_at": self.now, + }) + + def log_summaries(self, useful: int = 0, fallback: int = 0) -> None: + """Record summarization quality metrics.""" + total = useful + fallback + self._summary_counts = {"useful": useful, "fallback": fallback} + if total: + self._metrics.append({ + "run_id": self.run_id, + "metric_name": "summary_quality_rate", + "value": useful / total, + "metadata": json.dumps({ + "useful": useful, + "fallback": fallback, + "total": total, + }), + "recorded_at": self.now, + }) + + def log_metric( + self, + name: str, + value: float, + metadata: dict | None = None, + ) -> None: + """Record an arbitrary metric.""" + self._metrics.append({ + "run_id": self.run_id, + "metric_name": name, + "value": value, + "metadata": json.dumps(metadata) if metadata else None, + "recorded_at": self.now, + }) + + def save(self) -> int: + """Persist all accumulated metrics. Returns run_id.""" + # Write adapter health + for a in self._adapters: + self.conn.execute( + "INSERT INTO adapter_health " + "(run_id, adapter_name, items_fetched, errors, runtime_ms, " + " consecutive_failures, last_error, recorded_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + ( + a["run_id"], a["adapter_name"], a["items_fetched"], + a["errors"], a["runtime_ms"], a["consecutive_failures"], + a["last_error"], a["recorded_at"], + ), + ) + + # Write metrics + for m in self._metrics: + self.conn.execute( + "INSERT INTO metrics " + "(run_id, metric_name, value, metadata, recorded_at) " + "VALUES (?, ?, ?, ?, ?)", + ( + m["run_id"], m["metric_name"], m["value"], + m["metadata"], m["recorded_at"], + ), + ) + + # Update run_log totals + total_fetched = sum(a["items_fetched"] for a in self._adapters) + total_errors = sum(a["errors"] for a in self._adapters) + sources_ok = [a["adapter_name"] for a in self._adapters if a["items_fetched"] > 0] + sources_failed = [a["adapter_name"] for a in self._adapters if a["errors"] > 0] + + self.conn.execute( + "UPDATE run_log SET total_fetched = ?, total_stored = ?, " + "sources_ok = ?, sources_failed = ?, notes = ? " + "WHERE id = ?", + ( + total_fetched, + total_fetched - total_errors, + json.dumps(sources_ok), + json.dumps(sources_failed), + f"publish_rate={self._verdict_counts.get('PUBLISH', 0)}, " + f"summary_quality={self._summary_counts.get('useful', 0)}/{sum(self._summary_counts.values()) or 1}", + self.run_id, + ), + ) + + self.conn.commit() + return self.run_id + + def close(self) -> None: + self.conn.close() + + # ── Helpers ── + + def _consecutive_failures(self, adapter_name: str) -> int: + """Count consecutive runs where adapter returned 0 items.""" + cur = self.conn.execute( + "SELECT items_fetched FROM adapter_health " + "WHERE adapter_name = ? AND run_id < ? " + "ORDER BY run_id DESC LIMIT 5", + (adapter_name, self.run_id), + ) + count = 0 + for row in cur: + if row["items_fetched"] == 0: + count += 1 + else: + break + return count + + +# ── Aggregation queries ── + + +def get_publish_trend(db_path: str, days: int = 30) -> list[dict]: + """Daily PUBLISH rate over the last N days.""" + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + cur = conn.execute( + "SELECT date(m.recorded_at) as day, " + "ROUND(AVG(m.value), 4) as avg_publish_rate, " + "COUNT(*) as runs " + "FROM metrics m " + "WHERE m.metric_name = 'publish_rate' " + "AND m.recorded_at >= datetime('now', ?) " + "GROUP BY day ORDER BY day DESC", + (f"-{days} days",), + ) + results = [dict(r) for r in cur.fetchall()] + conn.close() + return results + + +def get_adapter_health(db_path: str, days: int = 7) -> list[dict]: + """Adapter health summary over last N days.""" + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + cur = conn.execute( + "SELECT adapter_name, " + "COUNT(*) as runs, " + "SUM(items_fetched) as total_fetched, " + "SUM(errors) as total_errors, " + "MAX(consecutive_failures) as max_consecutive_failures, " + "AVG(runtime_ms) as avg_runtime_ms, " + "MAX(last_error) as last_error " + "FROM adapter_health " + "WHERE recorded_at >= datetime('now', ?) " + "GROUP BY adapter_name", + (f"-{days} days",), + ) + results = [dict(r) for r in cur.fetchall()] + conn.close() + return results + + +def get_score_distribution(db_path: str, days: int = 30) -> list[dict]: + """Score distribution percentiles over last N days.""" + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + cur = conn.execute( + "SELECT date(recorded_at) as day, value as mean, metadata " + "FROM metrics " + "WHERE metric_name = 'score_mean' " + "AND recorded_at >= datetime('now', ?) " + "ORDER BY day DESC", + (f"-{days} days",), + ) + results = [] + for r in cur: + meta = json.loads(r["metadata"]) if r["metadata"] else {} + results.append({ + "day": r["day"], + "mean": r["mean"], + "min": meta.get("min"), + "max": meta.get("max"), + "p25": meta.get("p25"), + "p50": meta.get("p50"), + "p90": meta.get("p90"), + }) + conn.close() + return results + + +def get_summary_quality(db_path: str, days: int = 30) -> list[dict]: + """Summarization quality rate over time.""" + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + cur = conn.execute( + "SELECT date(recorded_at) as day, ROUND(AVG(value), 4) as avg_quality " + "FROM metrics " + "WHERE metric_name = 'summary_quality_rate' " + "AND recorded_at >= datetime('now', ?) " + "GROUP BY day ORDER BY day DESC", + (f"-{days} days",), + ) + results = [dict(r) for r in cur.fetchall()] + conn.close() + return results + + +def get_metric_alerts(db_path: str) -> list[str]: + """Generate alert strings based on metric anomalies.""" + alerts = [] + + # Adapter failures + health = get_adapter_health(db_path, days=7) + for h in health: + if h["max_consecutive_failures"] and h["max_consecutive_failures"] >= 3: + alerts.append( + f"⚠ {h['adapter_name']}: {h['max_consecutive_failures']} consecutive " + f"failed runs. Last error: {h.get('last_error', 'unknown')}" + ) + if h["total_errors"] and h["runs"]: + err_rate = h["total_errors"] / h["runs"] * 100 + if err_rate > 50: + alerts.append( + f"⚠ {h['adapter_name']}: {err_rate:.0f}% error rate " + f"({h['total_errors']}/{h['runs']} runs)" + ) + + # Publish starvation + trend = get_publish_trend(db_path, days=7) + if trend: + latest = trend[0]["avg_publish_rate"] + if latest == 0: + alerts.append("🔴 PUBLISH rate is 0% — scoring threshold may be too strict") + elif latest < 0.01: + alerts.append( + f"🟡 PUBLISH rate is {latest:.1%} — " + "consider lowering the signal threshold from 6.0" + ) + + # Summary quality + quality = get_summary_quality(db_path, days=7) + if quality: + latest = quality[0]["avg_quality"] + if latest < 0.5: + alerts.append( + f"🟡 Summary quality at {latest:.0%} — " + "most items falling back to title-only" + ) + + return alerts diff --git a/oracle/weekly.py b/oracle/weekly.py index f6d1773..d8e0c09 100644 --- a/oracle/weekly.py +++ b/oracle/weekly.py @@ -9,6 +9,7 @@ import sqlite3 from datetime import datetime, timedelta from pathlib import Path from collections import Counter +from oracle.metrics import get_adapter_health, get_metric_alerts WEEKLY_TEMPLATE = """# Weekly Review — {week_start} to {week_end} @@ -41,6 +42,10 @@ WEEKLY_TEMPLATE = """# Weekly Review — {week_start} to {week_end} {wow_md} +## SYSTEM HEALTH + +{health_md} + ## RECOMMENDATIONS {recs_md} @@ -216,10 +221,36 @@ def generate_weekly( output_path: str | None = None, ) -> dict: """Generate a complete weekly review.""" + db_path = conn.execute("SELECT file FROM pragma_database_list LIMIT 1").fetchone()[0] or "oracle.db" stats = fetch_weekly_stats(conn, days) trending = _trending_themes(conn, days) recommendations = _generate_recommendations(stats, trending) + # System health from metrics layer + health_data = get_adapter_health(db_path, days=days) + alerts = get_metric_alerts(db_path) + + health_lines = [] + if health_data: + for h in health_data: + status = "✅" + if h.get("max_consecutive_failures", 0) and h["max_consecutive_failures"] >= 3: + status = "🔴" + elif h.get("total_errors", 0) and h["runs"] and h["total_errors"] / h["runs"] > 0.5: + status = "🟡" + health_lines.append( + f"- {status} **{h['adapter_name']}**: {h['total_fetched']} fetched, " + f"{h['total_errors']} errors over {h['runs']} runs " + f"(avg {h['avg_runtime_ms']:.0f}ms)" + ) + else: + health_lines.append("_No adapter health data yet (metrics tracking started recently)._") + + if alerts: + health_lines.extend(f"- {a}" for a in alerts) + + health_md = "\n".join(health_lines) + # Format sections top_sources_md = "\n".join( f"- **{s}**: {c} entries" for s, c in stats["top_sources"] @@ -254,6 +285,8 @@ def generate_weekly( **stats, "trending": trending, "recommendations": recommendations, + "health": health_data, + "alerts": alerts, } review["markdown"] = WEEKLY_TEMPLATE.format( week_start=stats["week_start"], @@ -268,6 +301,7 @@ def generate_weekly( tier_md=tier_md, trending_md=trending_md, wow_md=wow, + health_md=health_md, recs_md=recs_md, )