gitlink-cli/skills/gitlink-spark/scripts/spark.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""gitlink-spark data fusion: arXiv x GitLink x GitHub -> JSON on stdout. Stdlib only."""
import argparse, json, os, sys, time, subprocess, urllib.request, urllib.parse, re
from xml.etree import ElementTree as ET

ARXIV_ENDPOINT = "https://export.arxiv.org/api/query"
GITHUB_ENDPOINT = "https://api.github.com/search/repositories"

_NS = {"a": "http://www.w3.org/2005/Atom"}

def parse_arxiv_atom(xml_text):
    """Parse arXiv Atom feed -> list of {arxiv_id, title, abstract, published}."""
    root = ET.fromstring(xml_text)
    papers = []
    for e in root.findall("a:entry", _NS):
        aid = (e.find("a:id", _NS).text or "").strip().split("/")[-1]
        title = re.sub(r"\s+", " ", (e.find("a:title", _NS).text or "").strip())
        summary = re.sub(r"\s+", " ", (e.find("a:summary", _NS).text or "").strip())
        pub = (e.find("a:published", _NS).text or "")[:10]
        papers.append({"arxiv_id": aid, "title": title, "abstract": summary, "published": pub})
    return papers

def parse_github_search(json_text):
    """Parse GitHub search JSON -> {total_count, top:[{full_name, stars}]}."""
    d = json.loads(json_text)
    return {
        "total_count": d.get("total_count", 0),
        "top": [{"full_name": r.get("full_name"), "stars": r.get("stargazers_count")}
                for r in (d.get("items") or [])[:3]],
    }

def extract_method_keywords(title, abstract, max_k=5):
    """Crude keyword extraction for GitHub/arXiv query."""
    text = (title + " " + abstract).lower()
    stop = {"the", "a", "an", "of", "for", "and", "to", "in", "on", "with", "via",
            "based", "using", "by", "from", "as", "is", "are", "we", "our", "this",
            "that", "propose", "proposed", "paper", "method", "approach", "novel", "new"}
    tokens = re.findall(r"[a-z][a-z0-9-]+", text)
    seen = set(); out = []
    for t in tokens:
        if t in stop or len(t) < 3 or t in seen:
            continue
        seen.add(t); out.append(t)
        if len(out) >= max_k:
            break
    return out

def fetch_arxiv(field, max_papers=10):
    """Search arXiv (HTTPS) for recent papers in field. Returns list of paper dicts."""
    q = urllib.parse.quote(f'abs:"{field}"')
    url = (f"{ARXIV_ENDPOINT}?search_query={q}&max_results={max_papers}"
           f"&sortBy=submittedDate&sortOrder=descending")
    with urllib.request.urlopen(url, timeout=30) as r:
        papers = parse_arxiv_atom(r.read().decode("utf-8", "replace"))
    for p in papers:
        p["method_keywords"] = extract_method_keywords(p["title"], p["abstract"])
    return papers

def _gitlink(*args):
    """Run gitlink-cli with json output; return parsed dict (UTF-8 safe)."""
    r = subprocess.run(["gitlink-cli"] + list(args) + ["--format", "json"],
                       capture_output=True, text=True, encoding="utf-8",
                       errors="replace", timeout=60)
    raw = r.stdout
    i = raw.find("{")
    return json.loads(raw[i:]) if i >= 0 else {}

def fetch_gitlink_repos(field):
    """gitlink-cli search +repos -> list of {owner, repo(identifier), name, desc, topics}."""
    d = _gitlink("search", "+repos", "-k", field)
    projs = d.get("data", {}).get("projects", []) or []
    out = []
    for p in projs:
        out.append({
            "owner": (p.get("author") or {}).get("login"),
            "repo": p.get("identifier"),
            "name": p.get("name"),
            "desc": p.get("description"),
            "topics": [t.get("name") if isinstance(t, dict) else t for t in (p.get("topics") or [])],
        })
    return out

def fetch_gitlink_issues(repos, max_per_repo=10):
    """Per-repo issue +list (open) -> list of {repo, number, subject, status, participants}.
    Works around search +issues returning HTML."""
    out = []
    for r in repos:
        if not (r.get("owner") and r.get("repo")):
            continue
        d = _gitlink("issue", "+list", "--owner", r["owner"], "--repo", r["repo"], "--state", "open")
        data = d.get("data", {}) or {}
        issues = data.get("issues") or []
        for it in issues[:max_per_repo]:
            st = (it.get("status") or {})
            if st.get("name") == "关闭":
                continue
            out.append({
                "repo": f'{r["owner"]}/{r["repo"]}',
                "number": it.get("project_issues_index") or it.get("number"),
                "subject": it.get("subject"),
                "status": st.get("name"),
                "participants": it.get("participants_count") or 0,
            })
    return out

_GH_CACHE = {}

def fetch_github_count(query, token=None, throttle=True):
    """GitHub search total_count + top3 for a query. Caches + throttles (10/min unauth)."""
    if query in _GH_CACHE:
        return _GH_CACHE[query]
    url = f"{GITHUB_ENDPOINT}?q={urllib.parse.quote(query)}&per_page=3&sort=stars"
    req = urllib.request.Request(url, headers={"Accept": "application/vnd.github+json",
                                               "User-Agent": "gitlink-spark/1.0"})
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    try:
        with urllib.request.urlopen(req, timeout=25) as r:
            res = parse_github_search(r.read().decode("utf-8", "replace"))
    except Exception as e:
        res = {"total_count": None, "top": [], "error": str(e)[:80]}
    if throttle and not token:
        time.sleep(7)  # unauthenticated = 10 req/min
    _GH_CACHE[query] = res
    return res

def main():
    ap = argparse.ArgumentParser(description="gitlink-spark data fusion")
    ap.add_argument("--field", required=True)
    ap.add_argument("--max-papers", type=int, default=10)
    ap.add_argument("--gap-type", default="both", choices=["both", "theory", "demand"])
    ap.add_argument("--github-token", default=os.environ.get("GITHUB_TOKEN"))
    args = ap.parse_args()

    papers = fetch_arxiv(args.field, args.max_papers)
    grepos = fetch_gitlink_repos(args.field)
    gissues = fetch_gitlink_issues(grepos) if args.gap_type in ("both", "demand") else []
    gh_counts = {}
    if args.gap_type in ("both", "theory"):
        for p in papers:
            mk = p.get("method_keywords") or []
            q = " ".join(mk[:3]) if mk else p["title"][:40]  # method keywords = implementation prevalence (NOT exact-title)
            gh_counts[q] = fetch_github_count(q, args.github_token)

    out = {
        "field": args.field,
        "papers": papers,
        "gitlink_repos": grepos,
        "gitlink_issues": gissues,
        "github_counts": gh_counts,
    }
    json.dump(out, sys.stdout, ensure_ascii=False, indent=2)
    sys.stdout.write("\n")

if __name__ == "__main__":
    main()