東北大学デジタルアーカイブの公開 API 調査 — OAI-PMH 経由で setSpec ごとに Excel 出力する

baseURL                : https://touda.tohoku.ac.jp/collection/
repositoryName         : Tohoku University Digital Archives Collection Database
earliestDatestamp      : 2013-01-01
deletedRecord          : persistent
granularity            : YYYY-MM-DDThh:mm:ssZ

pip install openpyxl

"""
東北大学 デジタルアーカイブ (touda.tohoku.ac.jp/collection) の
OAI-PMH エンドポイントから setSpec ごとに Excel ファイルを生成する。

  pip install openpyxl
  python oai_to_xlsx.py                 # 全セット
  python oai_to_xlsx.py libkano libsose # 指定セットのみ
"""

from __future__ import annotations

import re
import sys
import time
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path

from openpyxl import Workbook

BASE = "https://touda.tohoku.ac.jp/collection/oai"
PREFIX = "dcndl_simple"
OUT_DIR = Path("xlsx")
SLEEP_SEC = 0.3
TIMEOUT = 60

NS = {"oai": "http://www.openarchives.org/OAI/2.0/"}
XML_LANG = "{http://www.w3.org/XML/1998/namespace}lang"

HARVESTABLE_SETS = [
    "maibun", "tokiwa", "libkano", "libwasa", "libsose", "libakit",
    "libhonk", "libhonkandoc", "libshin", "arcreki", "arckoji",
    "arckank", "arcshas", "arcstdm", "mlmaterials",
]

NS_SHORT = {
    "http://purl.org/dc/elements/1.1/": "dc",
    "http://purl.org/dc/terms/": "dcterms",
    "http://ndl.go.jp/dcndl/terms/": "dcndl",
    "http://ndl.go.jp/dcndl/dcndl_simple/": "dcndl_simple",
    "http://xmlns.com/foaf/0.1/": "foaf",
    "http://www.w3.org/2002/07/owl#": "owl",
    "http://www.w3.org/1999/02/22-rdf-syntax-ns#": "rdf",
    "http://www.w3.org/2000/01/rdf-schema#": "rdfs",
    "https://irdb.nii.ac.jp/schema/jpcoar/1.0/": "jpcoar",
    "http://datacite.org/schema/kernel-4": "datacite",
}


def short_name(qname: str) -> str:
    m = re.match(r"\{([^}]+)\}(.+)", qname)
    if not m:
        return qname
    ns, local = m.group(1), m.group(2)
    return f"{NS_SHORT.get(ns, ns)}:{local}"


def fetch(url: str) -> bytes:
    req = urllib.request.Request(url, headers={"User-Agent": "oai-harvest/1.0"})
    with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
        return r.read()


def harvest_set(set_spec: str) -> list[dict]:
    rows: list[dict] = []
    params = {"verb": "ListRecords", "metadataPrefix": PREFIX, "set": set_spec}
    url = f"{BASE}?{urllib.parse.urlencode(params)}"
    page = 0
    while url:
        page += 1
        body = fetch(url)
        root = ET.fromstring(body)

        err = root.find("oai:error", NS)
        if err is not None:
            print(f"  [{set_spec}] error: {err.get('code')} {err.text}", file=sys.stderr)
            return rows

        for rec in root.findall("oai:ListRecords/oai:record", NS):
            rows.append(record_to_dict(rec))

        token_el = root.find("oai:ListRecords/oai:resumptionToken", NS)
        total = token_el.get("completeListSize") if token_el is not None else None
        print(f"  [{set_spec}] page {page} acquired={len(rows)} total={total}")

        if token_el is not None and (token_el.text or "").strip():
            params = {"verb": "ListRecords", "resumptionToken": token_el.text.strip()}
            url = f"{BASE}?{urllib.parse.urlencode(params)}"
            time.sleep(SLEEP_SEC)
        else:
            url = None
    return rows


def record_to_dict(rec: ET.Element) -> dict:
    out: dict[str, list[str]] = {}

    header = rec.find("oai:header", NS)
    if header is not None:
        out["_status"] = [header.get("status") or ""]
        ident = header.find("oai:identifier", NS)
        out["_identifier"] = [ident.text or ""] if ident is not None else [""]
        ds = header.find("oai:datestamp", NS)
        out["_datestamp"] = [ds.text or ""] if ds is not None else [""]
        sets = [s.text or "" for s in header.findall("oai:setSpec", NS)]
        out["_setSpec"] = sets

    metadata = rec.find("oai:metadata", NS)
    if metadata is not None:
        for wrapper in list(metadata):
            for child in list(wrapper):
                key = short_name(child.tag)
                lang = child.get(XML_LANG)
                if lang:
                    key = f"{key}@{lang}"
                value = (child.text or "").strip()
                if not value and len(list(child)) > 0:
                    value = " ".join(
                        (g.text or "").strip() for g in child.iter() if (g.text or "").strip()
                    )
                if value:
                    out.setdefault(key, []).append(value)
    return {k: "\n".join(v) for k, v in out.items()}


def write_xlsx(set_spec: str, rows: list[dict]) -> Path:
    OUT_DIR.mkdir(exist_ok=True)
    path = OUT_DIR / f"{set_spec}.xlsx"

    columns: list[str] = []
    seen: set[str] = set()
    for r in rows:
        for k in r:
            if k not in seen:
                seen.add(k)
                columns.append(k)
    header_cols = [c for c in columns if c.startswith("_")]
    body_cols = [c for c in columns if not c.startswith("_")]
    columns = header_cols + body_cols

    wb = Workbook(write_only=True)
    ws = wb.create_sheet(set_spec[:31] or "data")
    ws.append(columns)
    for r in rows:
        ws.append([r.get(c, "") for c in columns])
    wb.save(path)
    return path


def main(argv: list[str]) -> int:
    targets = argv[1:] if len(argv) > 1 else HARVESTABLE_SETS
    for s in targets:
        if s not in HARVESTABLE_SETS:
            print(f"warn: '{s}' is not in the known harvestable list; trying anyway")
        print(f"==> {s}")
        rows = harvest_set(s)
        if not rows:
            print(f"  [{s}] no records, skipping")
            continue
        path = write_xlsx(s, rows)
        print(f"  [{s}] wrote {len(rows)} rows -> {path}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

_identifier  _datestamp  _setSpec  _status
dc:identifier
dcndl:materialType
dcterms:language
dc:title@ja
dcndl:alternative@ja
dcterms:publisher@ja
dcndl:publicationPlace@ja
dcterms:date
dcterms:extent          ← "B5判\n168p" のように 1 セルに複数値
dcterms:spatial@ja
dcterms:spatial@en
dcterms:description@ja  ← 注記が複数あれば改行で連結
dcterms:rights

# 全 15 セット
python oai_to_xlsx.py

# 指定セットのみ
python oai_to_xlsx.py libsose libhonkandoc