sources.py

#!/usr/bin/env python
"""Source adapters for the Web Research Agent (#13) — different areas need different sources.

Every adapter is keyless and returns the same shape [{title,url,snippet,source}] so the agent can
fetch + cite uniformly. Areas map to a blend of adapters:

  web         → general web            (opportunity/competitive/client baseline)
  opportunity → web + fresh news
  competitive → web + fresh news
  technical   → web + arXiv + GitHub
  client      → web
"""
import urllib.parse

import requests
import feedparser

from search import search as web_search, search_news

_UA = "Mozilla/5.0 (compatible; GritAI-Research/1.0)"


def arxiv(query: str, n: int = 4) -> list:
    """arXiv abstracts (keyless Atom API). Abstract pages are readable/fetchable."""
    url = ("http://export.arxiv.org/api/query?search_query=all:"
           f"{urllib.parse.quote(query)}&start=0&max_results={int(n)}&sortBy=relevance")
    d = feedparser.parse(url)
    out = []
    for e in d.entries[:n]:
        link = e.get("link", "")
        out.append({"title": e.get("title", "").replace("\n", " ").strip(),
                    "url": link, "snippet": (e.get("summary", "") or "")[:300], "source": "arxiv"})
    return [r for r in out if r["url"]]


def github(query: str, n: int = 4) -> list:
    """GitHub repository search (keyless public API; low-volume friendly)."""
    try:
        r = requests.get("https://api.github.com/search/repositories",
                         params={"q": query, "sort": "stars", "per_page": n},
                         headers={"User-Agent": _UA, "Accept": "application/vnd.github+json"}, timeout=15)
        items = r.json().get("items", [])
        return [{"title": f"{it['full_name']} (★{it.get('stargazers_count', 0)})",
                 "url": it["html_url"], "snippet": (it.get("description") or "")[:300], "source": "github"}
                for it in items[:n]]
    except Exception:
        return []


AREAS = {
    "web":         [("web", 1.0)],
    "opportunity": [("web", 0.6), ("news", 0.4)],
    "competitive": [("web", 0.6), ("news", 0.4)],
    "technical":   [("web", 0.5), ("arxiv", 0.25), ("github", 0.25)],
    "client":      [("web", 1.0)],
}

_ADAPTERS = {"web": web_search, "news": search_news, "arxiv": arxiv, "github": github}


def search_area(query: str, area: str = "web", n: int = 6) -> list:
    """Blend results from the adapters mapped to `area`, deduped by URL."""
    plan = AREAS.get(area, AREAS["web"])
    seen, out = set(), []
    for name, weight in plan:
        want = max(1, round(n * weight))
        try:
            hits = _ADAPTERS[name](query, want + 1)
        except Exception:
            hits = []
        for h in hits[:want]:
            key = (h["url"] or "").split("#")[0].rstrip("/")
            if key and key not in seen:
                seen.add(key)
                out.append(h)
    return out[:n]


if __name__ == "__main__":
    import sys, json
    area = sys.argv[1] if len(sys.argv) > 1 else "technical"
    q = " ".join(sys.argv[2:]) or "retrieval augmented generation agents"
    print(f"area={area} query={q}")
    for r in search_area(q, area, 8):
        print(f"  [{r['source']}] {r['title'][:64]} — {r['url'][:70]}")