worldmodel/plans/PRISM/.research/fetch_projects.py

#!/usr/bin/env python3
"""
fetch_projects.py — 抓取同档比较项目的元数据(arXiv + Semantic Scholar + GitHub)
输出 JSON 到 stdout 与 projects.json,供后续 comparison.md 编写引用。
"""
from __future__ import annotations
import json
import pathlib
import sys
import time
from typing import Any
import urllib.parse
import urllib.request

OUT = pathlib.Path(__file__).parent / "projects.json"

# 待研究项目列表
PROJECTS = [
    {
        "key": "concept-graphs",
        "name": "ConceptGraphs",
        "arxiv": "2309.16650",
        "gh": "concept-graphs/concept-graphs",
        "homepage": "https://concept-graphs.github.io/",
    },
    {
        "key": "hov-sg",
        "name": "HOV-SG",
        "arxiv": "2403.17846",
        "gh": "hovsg/HOV-SG",
        "homepage": "https://hovsg.github.io/",
    },
    {
        "key": "openscene",
        "name": "OpenScene",
        "arxiv": "2211.15654",
        "gh": "pengsongyou/openscene",
        "homepage": "https://pengsongyou.github.io/openscene",
    },
    {
        "key": "3d-llm",
        "name": "3D-LLM",
        "arxiv": "2307.12981",
        "gh": "UMass-Foundation-Model/3D-LLM",
        "homepage": "https://vis-www.cs.umass.edu/3dllm/",
    },
    {
        "key": "ok-robot",
        "name": "OK-Robot",
        "arxiv": "2401.12202",
        "gh": "ok-robot/ok-robot",
        "homepage": "https://ok-robot.github.io/",
    },
    {
        "key": "openmask3d",
        "name": "OpenMask3D",
        "arxiv": "2306.13631",
        "gh": "OpenMask3D/openmask3d",
        "homepage": "https://openmask3d.github.io/",
    },
    {
        "key": "conceptfusion",
        "name": "ConceptFusion",
        "arxiv": "2302.07241",
        "gh": "concept-fusion/concept-fusion",
        "homepage": "https://concept-fusion.github.io/",
    },
    {
        "key": "clio",
        "name": "Clio (MIT-SPARK)",
        "arxiv": "2404.13696",
        "gh": "MIT-SPARK/Clio",
        "homepage": "https://clio-cmu.github.io/",
    },
]

UA = "Mozilla/5.0 (research) PRISM-comparison/1.0"


def _get(url: str, headers: dict[str, str] | None = None,
         timeout: int = 20) -> str:
    h = {"User-Agent": UA, "Accept": "application/json"}
    if headers:
        h.update(headers)
    req = urllib.request.Request(url, headers=h)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.read().decode("utf-8", errors="replace")
    except Exception as e:
        return json.dumps({"_error": str(e)})


def fetch_arxiv(arxiv_id: str) -> dict[str, Any]:
    """Semantic Scholar 一次查到 abstract、引用数、年份"""
    url = (
        "https://api.semanticscholar.org/graph/v1/paper/"
        f"arXiv:{arxiv_id}?fields=title,abstract,year,citationCount,"
        "authors.name,venue,publicationVenue,referenceCount,influentialCitationCount,"
        "openAccessPdf,fieldsOfStudy"
    )
    raw = _get(url)
    try:
        return json.loads(raw)
    except Exception as e:
        return {"_error": f"json: {e}", "_raw": raw[:300]}


def fetch_github(repo: str) -> dict[str, Any]:
    url = f"https://api.github.com/repos/{repo}"
    raw = _get(url, headers={"Accept": "application/vnd.github+json"})
    try:
        d = json.loads(raw)
        # 只保留关心字段
        if "stargazers_count" in d:
            return {
                "stars": d.get("stargazers_count"),
                "forks": d.get("forks_count"),
                "subscribers": d.get("subscribers_count"),
                "language": d.get("language"),
                "license": (d.get("license") or {}).get("spdx_id"),
                "open_issues": d.get("open_issues_count"),
                "created_at": d.get("created_at"),
                "updated_at": d.get("updated_at"),
                "pushed_at": d.get("pushed_at"),
                "topics": d.get("topics"),
                "description": d.get("description"),
                "default_branch": d.get("default_branch"),
                "archived": d.get("archived"),
                "homepage": d.get("homepage"),
            }
        return d
    except Exception as e:
        return {"_error": f"json: {e}", "_raw": raw[:300]}


def main() -> None:
    out: dict[str, Any] = {}
    for p in PROJECTS:
        sys.stderr.write(f"[fetch] {p['name']} ({p['arxiv']}) ...\n")
        out[p["key"]] = {
            "meta": p,
            "arxiv_via_s2": fetch_arxiv(p["arxiv"]),
        }
        time.sleep(1.2)  # 礼貌:S2 rate limit
        out[p["key"]]["github"] = fetch_github(p["gh"])
        time.sleep(0.8)

    OUT.write_text(json.dumps(out, ensure_ascii=False, indent=2),
                   encoding="utf-8")
    sys.stderr.write(f"[ok] wrote {OUT}\n")
    print(f"saved {OUT}")


if __name__ == "__main__":
    main()