Claude Codeでresearchmap APIから業績を取り、公募との照合表を作る

"""researchmap API から業績を取り、公募キーワードとの照合表を CSV に出す。"""
import csv, json, os, sys, time
import urllib.parse, urllib.request, urllib.error
from datetime import date
 
import jwt  # PyJWT(RS256 署名に cryptography が必要)
 
BASE = "https://api.researchmap.jp"
TOKEN_URL = BASE + "/oauth2/token"
TYPES = ["research_interests", "published_papers", "research_projects"]
WEIGHT = {"research_interests": 3, "research_projects": 2, "published_papers": 1}
YEARS = int(os.environ.get("RMAP_YEARS", "5"))  # 何年前までの業績を見るか
 
 
def get_token():
    now = int(time.time())
    claim = {"iss": os.environ["RMAP_CLIENT_ID"], "aud": TOKEN_URL,
             "sub": "0", "iat": now - 10, "exp": now + 90}
    with open(os.environ["RMAP_KEY_FILE"], "rb") as f:  # 鍵はリポジトリの外に置く
        assertion = jwt.encode(claim, f.read(), algorithm="RS256")
    body = urllib.parse.urlencode({
        "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
        "assertion": assertion, "scope": "public_only achievements", "version": "2",
    }).encode()
    with urllib.request.urlopen(urllib.request.Request(TOKEN_URL, data=body)) as r:
        return json.load(r)["access_token"]
 
 
def fetch_list(token, permalink, kind, from_year):
    """業績リストを最後のページまで取る。失敗は (None, 理由) で返す。"""
    url = f"{BASE}/{permalink}/{kind}?limit=1000&format=json"
    if kind != "research_interests":
        url += f"&from_date={from_year}"
    items = []
    while url:
        req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
        try:
            with urllib.request.urlopen(req) as r:
                data = json.load(r)
        except urllib.error.HTTPError as e:
            return None, f"HTTP {e.code}"
        items += data.get("items", [])
        nxt = data.get("_links", {}).get("next")
        url = nxt["href"] if nxt else None
    return items, ""
 
 
def text_of(kind, it):
    def ja_en(d):
        return " ".join((d or {}).get(k, "") for k in ("ja", "en"))
    if kind == "research_interests":
        return ja_en(it.get("keyword"))
    if kind == "published_papers":
        return ja_en(it.get("paper_title"))
    return " ".join([ja_en(it.get("research_project_title")),
                     ja_en(it.get("system_name")), ja_en(it.get("description"))])
 
 
def score(blobs, keywords):
    """blobs: {kind: [text,...]}。キーワードごとに最初に当たった種別だけ数える。"""
    total, hits = 0, []
    for kw in keywords:
        for kind in TYPES:
            hit = next((t for t in blobs.get(kind, []) if kw.lower() in t.lower()), None)
            if hit:
                total += WEIGHT[kind]
                hits.append(f"{kw}←{kind}:{hit[:30]}")
                break
    return total, hits
 
 
def main(researchers_csv, calls_csv, out_csv):
    token = get_token()
    from_year = date.today().year - YEARS
    people = []
    for r in csv.DictReader(open(researchers_csv, encoding="utf-8-sig")):
        blobs, flags, papers = {}, [], []
        for kind in TYPES:
            items, err = fetch_list(token, r["permalink"], kind, from_year)
            if items is None:
                flags.append(f"{kind}取得失敗({err})")
                continue
            blobs[kind] = [text_of(kind, i) for i in items]
            if kind == "published_papers":
                papers = items
        if not any(blobs.values()):
            flags.append("公開業績なし(permalink 違い・非公開・自機関外の可能性)")
        elif papers and not any(r["family_ja"] in json.dumps(p.get("authors", {}), ensure_ascii=False)
                                for p in papers):
            flags.append("著者欄に姓が出ない(別人の permalink か表記違い)")
        if not blobs.get("published_papers"):
            flags.append(f"直近{YEARS}年の論文なし(古い業績のみの可能性)")
        people.append((r, blobs, flags))
    rows = []
    for c in csv.DictReader(open(calls_csv, encoding="utf-8-sig")):
        kws = [k.strip() for k in c["keywords"].split(";") if k.strip()]
        for r, blobs, flags in people:
            s, hits = score(blobs, kws)
            if s:
                rows.append({"call_id": c["call_id"], "公募名": c["title"], "締切": c["deadline"],
                             "研究者": r["name_ja"], "permalink": r["permalink"], "点数": s,
                             "根拠": " / ".join(hits), "要確認": " ; ".join(flags), "人の確認": ""})
    rows.sort(key=lambda x: (x["call_id"], -x["点数"]))
    with open(out_csv, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0]) if rows else ["call_id"])
        w.writeheader(); w.writerows(rows)
    print(f"{len(rows)} 行を {out_csv} に書きました", file=sys.stderr)
 
 
if __name__ == "__main__":
    main(*sys.argv[1:4])