Claude CodeでKAKEN APIの採択課題を集計する — 部局別・種目別の基礎表

## KAKEN採択課題の集計
- 取得は `scripts/kaken_summary.py` だけで行う。appidは環境変数 KAKEN_APPID から読み、出力しない
- 件数は開始年度（o1=1）で数える。実施期間の一部を含む条件（o1=3）で取った行は足し合わせない
- 集計表を出したら、必ず `scripts/check_totals.py` で行数と合計の一致を確かめる
- 部局名・配分額の不整合は直さずに issues.csv へ出す。表の値を手で補正しない
- 配分額は採択時の額として扱い、表の注記にそう書く

{
  "permissions": {
    "allow": [
      "Bash(python3 scripts/kaken_summary.py *)",
      "Bash(python3 scripts/check_totals.py *)"
    ],
    "deny": ["Read(./.env)", "Read(./.env.*)"]
  }
}

"""KAKEN OpenSearch(研究課題をさがす, format=xml)から機関の採択課題を集め、集計表と点検表を出す。"""
import csv, os, sys, time, urllib.parse, urllib.request
import xml.etree.ElementTree as ET
from collections import defaultdict
 
BASE = os.environ.get("KAKEN_BASE", "https://kaken.nii.ac.jp/opensearch/")
XML_LANG = "{http://www.w3.org/XML/1998/namespace}lang"
PAGE = 500  # rw に指定できる最大値
 
 
def local(tag):
    return tag.rsplit("}", 1)[-1]
 
 
def kids(el, name):
    return [c for c in el if local(c.tag) == name]
 
 
def first(el, name):
    k = kids(el, name)
    return k[0] if k else None
 
 
def text(el):
    return " ".join("".join(el.itertext()).split()) if el is not None else ""
 
 
def fetch(inst, y1, y2, st):
    q = {"appid": os.environ["KAKEN_APPID"], "qe": inst, "s1": y1, "s2": y2,
         "o1": 1, "c6": "project", "od": 3, "rw": PAGE, "st": st, "format": "xml"}
    url = BASE + "?" + urllib.parse.urlencode(q)
    with urllib.request.urlopen(url, timeout=60) as r:
        return ET.fromstring(r.read())
 
 
def pick_summary(award):
    sums = kids(award, "summary")
    for s in sums:
        if s.get(XML_LANG) == "ja":
            return s
    return sums[0] if sums else None
 
 
def amount(summary):
    for oa in kids(summary, "overallAwardAmount"):
        if oa.get("planned") == "true":
            continue
        tc = first(oa, "totalCost")
        if tc is None:
            continue
        nv = first(tc, "normalizedValue")
        raw = text(nv) if nv is not None else text(tc)
        digits = "".join(ch for ch in raw if ch.isdigit())
        return int(digits) if digits else None
    return None
 
 
def parse(award):
    s = pick_summary(award)
    if s is None:
        return None
    pi = next((m for m in kids(s, "member") if m.get("role") == "principal_investigator"), None)
    dept, dept_code, pi_inst = "", "", ""
    if pi is not None:
        aff = first(pi, "affiliation")
        if aff is not None:
            d, i = first(aff, "department"), first(aff, "institution")
            dept, dept_code = text(d), (d.get("niiCode") or "") if d is not None else ""
            pi_inst = text(i)
    pa = first(s, "periodOfAward")
    sy = first(pa, "startFiscalYear") if pa is not None else None
    ps = first(s, "projectStatus")
    cats = kids(s, "category")
    return {
        "id": award.get("id", ""),
        "number": award.get("awardNumber", ""),
        "title": text(first(s, "title")),
        "category": text(cats[0]) if cats else "",
        "start_year": text(sy),
        "status": (ps.get("statusCode") if ps is not None else "") or "",
        "amount": amount(s),
        "dept": dept, "dept_code": dept_code, "pi_inst": pi_inst,
        "institutions": " / ".join(text(i) for i in kids(s, "institution")),
    }
 
 
def main(inst, y1, y2, out):
    rows, st = [], 1
    while True:
        root = fetch(inst, y1, y2, st)
        batch = [parse(a) for a in root.iter() if local(a.tag) == "grantAward"]
        batch = [b for b in batch if b]
        rows += batch
        if len(batch) < PAGE:
            break
        st += PAGE
        time.sleep(2)  # 短時間の大量アクセスを避ける
    os.makedirs(out, exist_ok=True)
    with open(f"{out}/projects.csv", "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0]) if rows else ["id"])
        w.writeheader(); w.writerows(rows)
 
    def pivot(key, name):
        cnt, amt = defaultdict(int), defaultdict(int)
        for r in rows:
            k = (r[key] or "(未設定)", r["start_year"])
            cnt[k] += 1; amt[k] += r["amount"] or 0
        years = sorted({k[1] for k in cnt})
        with open(f"{out}/{name}.csv", "w", newline="", encoding="utf-8-sig") as f:
            w = csv.writer(f)
            w.writerow([key] + [f"{y}件" for y in years] + [f"{y}円" for y in years])
            for g in sorted({k[0] for k in cnt}):
                w.writerow([g] + [cnt[(g, y)] for y in years] + [amt[(g, y)] for y in years])
 
    pivot("dept", "by_dept"); pivot("category", "by_category")
 
    issues = []
    ids = defaultdict(list)
    for r in rows:
        ids[r["id"]].append(r)
    issues += [("id重複", i, "") for i, v in ids.items() if len(v) > 1]
    by_code = defaultdict(set)
    for r in rows:
        if r["dept_code"]:
            by_code[r["dept_code"]].add(r["dept"])
    issues += [("部局表記ゆれ", c, " | ".join(sorted(v))) for c, v in by_code.items() if len(v) > 1]
    issues += [("部局コードなし", r["id"], r["dept"]) for r in rows if not r["dept_code"]]
    issues += [("配分額なし", r["id"], r["title"]) for r in rows if r["amount"] is None]
    issues += [("終了系ステータス", r["id"], r["status"]) for r in rows
               if r["status"] in ("declined", "discontinued")]
    issues += [("代表者所属が他機関", r["id"], r["pi_inst"]) for r in rows
               if r["pi_inst"] and inst not in r["pi_inst"]]
    with open(f"{out}/issues.csv", "w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f); w.writerow(["種別", "課題ID/コード", "内容"]); w.writerows(issues)
    print(f"{len(rows)}件取得 / 点検 {len(issues)}件 -> {out}/")
 
 
if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4])

"""by_dept / by_category の件数・配分額の合計が projects.csv と一致するか確かめる。"""
import csv, sys
 
out = sys.argv[1]
rows = list(csv.DictReader(open(f"{out}/projects.csv", encoding="utf-8-sig")))
n, total = len(rows), sum(int(r["amount"] or 0) for r in rows)
for name in ("by_dept", "by_category"):
    t = list(csv.reader(open(f"{out}/{name}.csv", encoding="utf-8-sig")))
    half = (len(t[0]) - 1) // 2
    cnt = sum(int(v) for r in t[1:] for v in r[1:1 + half])
    amt = sum(int(v) for r in t[1:] for v in r[1 + half:])
    ok = "OK" if (cnt, amt) == (n, total) else "NG"
    print(f"{name}: {cnt}件 {amt}円 / projects.csv {n}件 {total}円 {ok}")

rw=20 の1ページ分のXMLを out/sample.xml に保存して。
grantAward ごとに、種目・開始年度・配分額・研究代表者の所属部局が
kaken_summary.py の parse() で取り出せているか、5件で示して。
取り出せない項目があれば、実際の要素名を挙げて parse() を直す案だけ出して。

out/2026/issues.csv の「部局表記ゆれ」と「部局コードなし」を読んで、
scripts/dept_map.csv（列: 元の表記, 集計用の部局名）の案を作って。
コードが同じ表記は1つにまとめ、コードのない行は推測で割り当てず、
「要確認」として別の列に残して。

出典：KAKEN：科学研究費助成事業データベース（国立情報学研究所）
（https://kaken.nii.ac.jp/）をもとに〇〇大学作成