Epictetus
67c002b665
Source-controlled baseline before Phase 5 cron. Excludes oracle.db, logs/, and __pycache__ via .gitignore. Pipeline verified running clean end-to-end (run_log write confirmed before conn.close()).
312 lines
12 KiB
Python
312 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GitHub adapter for AI Research Oracle.
|
|
Fetches repos via GitHub REST API — no scraping, no trending page.
|
|
Definition of "trending": created or pushed in last N days, sorted by stars.
|
|
|
|
Rate limits: 60 req/hr unauthenticated.
|
|
Strategy: cache results, deduplicate, budget calls.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import urllib.parse
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from adapters import SourceAdapter
|
|
|
|
|
|
class GitHubAdapter(SourceAdapter):
|
|
"""GitHub REST API adapter."""
|
|
|
|
BASE = "https://api.github.com"
|
|
|
|
def __init__(self, token: str = None):
|
|
"""Initialize with optional read-only token (5000 req/hr vs 60)."""
|
|
self.token = token or os.environ.get("GITHUB_TOKEN", "")
|
|
self.cache = {}
|
|
|
|
def name(self) -> str:
|
|
return "github"
|
|
|
|
def _headers(self):
|
|
headers = {
|
|
"Accept": "application/vnd.github.v3+json",
|
|
"User-Agent": "ai-oracle/0.1",
|
|
}
|
|
if self.token:
|
|
headers["Authorization"] = f"token {self.token}"
|
|
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
|
|
|
|
def _search_repos(self, query: str, sort: str = "stars", order: str = "desc", per_page: int = 30) -> list:
|
|
"""Search repositories via GitHub API."""
|
|
url = f"{self.BASE}/search/repositories?q={urllib.parse.quote(query)}&sort={sort}&order={order}&per_page={per_page}"
|
|
result = self._request(url)
|
|
if isinstance(result, dict) and "items" in result:
|
|
return result["items"]
|
|
return []
|
|
|
|
def _get_readme(self, repo_url: str) -> str:
|
|
"""Fetch README content for a repo."""
|
|
if repo_url in self.cache:
|
|
return self.cache[repo_url]
|
|
|
|
result = self._request(repo_url, max_retries=1)
|
|
if result and isinstance(result, dict):
|
|
import base64
|
|
content = result.get("content", "")
|
|
encoding = result.get("encoding", "base64")
|
|
if encoding == "base64" and content:
|
|
try:
|
|
decoded = base64.b64decode(content).decode("utf-8", errors="replace")
|
|
# Truncate to 8000 chars to keep context manageable
|
|
self.cache[repo_url] = decoded[:8000]
|
|
return decoded[:8000]
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
def _score(self, repo: dict, age_days: float) -> float:
|
|
"""Score: stars weighted by recency. Higher = better signal."""
|
|
stars = repo.get("stargazers_count", 0)
|
|
# Normalize: log scale on stars, decay by age
|
|
import math
|
|
star_score = min(math.log1p(stars) / 2.0, 10.0) # log(1000) ≈ 6.9 → ~3.5
|
|
# Recency bonus: newer repos get a slight boost
|
|
recency_bonus = max(0, 1.0 - age_days / 30.0) * 1.5 # up to +1.5 for very recent
|
|
return min(star_score + recency_bonus, 10.0)
|
|
|
|
def _tags(self, repo: dict) -> list:
|
|
"""Generate category tags from repo metadata."""
|
|
tags = ["github"]
|
|
lang = repo.get("language", "")
|
|
if lang:
|
|
tags.append(f"lang:{lang.lower()}")
|
|
topics = repo.get("topics", [])
|
|
for t in topics[:5]:
|
|
tags.append(f"topic:{t}")
|
|
# Check description for AI/ML signals
|
|
desc = (repo.get("description") or "").lower()
|
|
if any(kw in desc for kw in ["agent", "agents"]):
|
|
tags.append("agents")
|
|
if any(kw in desc for kw in ["llm", "large language"]):
|
|
tags.append("llm")
|
|
if any(kw in desc for kw in ["rag", "retrieval"]):
|
|
tags.append("rag")
|
|
return tags
|
|
|
|
def fetch(self, query: str = "", limit: int = 20) -> list[dict]:
|
|
"""
|
|
Fetch repos from GitHub.
|
|
|
|
If query is empty, fetch "trending" = recent high-star AI repos.
|
|
If query is provided, search for it.
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
|
|
if query:
|
|
print(f" Searching: '{query}'")
|
|
repos = self._search_repos(query, sort="stars", per_page=min(limit * 2, 100))
|
|
else:
|
|
# "Trending" = AI repos created in last 14 days, sorted by stars
|
|
cutoff = (now - timedelta(days=14)).strftime("%Y-%m-%d")
|
|
# Two queries to get breadth
|
|
repos = []
|
|
for q in ["ai agents created:>=2026-06-01", "llm inference created:>=2026-06-01"]:
|
|
batch = self._search_repos(q, sort="stars", per_page=50)
|
|
repos.extend(batch)
|
|
time.sleep(1) # polite spacing
|
|
|
|
# Deduplicate by full_name
|
|
seen = set()
|
|
unique = []
|
|
for r in repos:
|
|
fn = r.get("full_name", "")
|
|
if fn not in seen:
|
|
seen.add(fn)
|
|
unique.append(r)
|
|
repos = unique
|
|
|
|
# Sort by stars descending, take top limit
|
|
repos.sort(key=lambda r: r.get("stargazers_count", 0), reverse=True)
|
|
repos = repos[:limit]
|
|
|
|
entries = []
|
|
readme_budget = min(10, limit) # Only fetch README for top 10 to stay under rate limit
|
|
for idx, repo in enumerate(repos):
|
|
# Calculate age
|
|
created = repo.get("created_at")
|
|
if created:
|
|
try:
|
|
created_dt = datetime.fromisoformat(created.replace("Z", "+00:00"))
|
|
age_days = (now - created_dt).days
|
|
except (ValueError, TypeError):
|
|
age_days = 0
|
|
else:
|
|
age_days = 0
|
|
|
|
score = self._score(repo, age_days)
|
|
tags = self._tags(repo)
|
|
|
|
# Source ID: repo full_name
|
|
source_id = repo.get("full_name", "").replace("/", "__")
|
|
|
|
# URL
|
|
url = repo.get("html_url", "")
|
|
|
|
# Title: repo name with description
|
|
name = repo.get("name", "")
|
|
desc = repo.get("description", "") or ""
|
|
if desc:
|
|
title = f"{name}: {desc[:100]}"
|
|
else:
|
|
title = name
|
|
|
|
# README as extracted_text
|
|
# Search API doesn't return readme_url — construct it from owner/repo
|
|
owner = repo.get("owner", {}).get("login", "")
|
|
repo_name = repo.get("name", "")
|
|
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
|
|
|
|
# Structured metadata
|
|
raw_meta = {
|
|
"full_name": repo.get("full_name", ""),
|
|
"owner": repo.get("owner", {}).get("login", ""),
|
|
"stars": repo.get("stargazers_count", 0),
|
|
"forks": repo.get("forks_count", 0),
|
|
"open_issues": repo.get("open_issues_count", 0),
|
|
"language": repo.get("language", ""),
|
|
"topics": repo.get("topics", []),
|
|
"created_at": repo.get("created_at", ""),
|
|
"updated_at": repo.get("updated_at", ""),
|
|
"pushed_at": repo.get("pushed_at", ""),
|
|
"age_days": age_days,
|
|
"readme_length": len(readme_text),
|
|
"score_type": "actual", # based on real star counts
|
|
}
|
|
|
|
now_str = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
entries.append({
|
|
"source": "github",
|
|
"source_id": source_id,
|
|
"url": url,
|
|
"title": title,
|
|
"extracted_text": readme_text,
|
|
"summary": None, # LLM later
|
|
"category_tags": json.dumps(tags),
|
|
"signal_score": round(score, 2),
|
|
"raw_metadata": json.dumps(raw_meta),
|
|
"first_seen": now_str,
|
|
"last_updated": now_str,
|
|
})
|
|
|
|
return entries
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
import sqlite3
|
|
import os
|
|
|
|
parser = argparse.ArgumentParser(description="GitHub adapter for AI Research Oracle")
|
|
parser.add_argument("--query", default="", help="Search query (empty = trending AI)")
|
|
parser.add_argument("--limit", type=int, default=20, help="Max entries")
|
|
parser.add_argument("--db", default=os.path.join(os.path.dirname(__file__), "..", "oracle.db"), help="SQLite DB")
|
|
parser.add_argument("--schema", default=os.path.join(os.path.dirname(__file__), "..", "schema.sql"), help="Schema file")
|
|
parser.add_argument("--dry-run", action="store_true", help="Don't store in DB")
|
|
args = parser.parse_args()
|
|
|
|
print(f"=== GitHub Adapter ===")
|
|
print(f" Query: {args.query or '(trending AI)'}")
|
|
print(f" Limit: {args.limit}")
|
|
print()
|
|
|
|
adapter = GitHubAdapter()
|
|
entries = adapter.fetch(query=args.query, limit=args.limit)
|
|
|
|
print(f" Fetched {len(entries)} entries")
|
|
|
|
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" Stored {stored} entries")
|
|
|
|
# Print top 5
|
|
print(f"\n Top entries:")
|
|
for i, e in enumerate(entries[:5]):
|
|
meta = json.loads(e["raw_metadata"]) if isinstance(e["raw_metadata"], str) else e["raw_metadata"]
|
|
print(f" [{i+1}] score={e['signal_score']:.2f} stars={meta.get('stars', '?')}")
|
|
print(f" {e['title'][:90]}")
|
|
print(f" {e['url']}")
|
|
print(f" text={len(e.get('extracted_text', ''))}ch")
|
|
|
|
print(f"\n Done.")
|