stack/dev/scripts/osf_metadata.py

"""Set tags, subjects and license on the LEAD review's OSF records.

The OSF registration (8wq3m) and its project (bwc9n) were created bare:
no tags, no subjects, no license. Registration metadata stays editable
after OSF Projects go read-only (2027-02-19); the project is editable
until then. Dry run by default (prints the requests); --apply writes.

Needs a personal access token with osf.full_write in OSF_TOKEN (.env).

    uv run python dev/scripts/osf_metadata.py            # show the plan
    uv run python dev/scripts/osf_metadata.py --apply    # write it
"""

from __future__ import annotations

import argparse
import json
import os
import sys

import httpx

API = "https://api.osf.io/v2"
UA = {"User-Agent": "stack-osf-metadata/1.0"}

TARGETS = [("registrations", "8wq3m"), ("nodes", "bwc9n")]

TAGS = [
    "LEAD model",
    "ACO",
    "accountable care",
    "Medicare",
    "CMS Innovation Center",
    "living review",
    "scoping review",
    "systematic review",
    "prognostic factors",
    "health policy",
    "PRISMA-ScR",
    "performance year 2027",
]

LICENSE_ID = "563c1cf88c5e4a3877f9e96a"  # CC-By Attribution 4.0 International

# Leaf subjects (bepress taxonomy on the OSF registration provider);
# their ancestors are added at run time, as OSF expects full paths.
LEAF_SUBJECTS = {
    "584240da54be81056cecacac": "Health Services Research",
    "584240da54be81056cecab1f": "Health Policy",
    "584240d954be81056ceca846": "Health Economics",
}


def subject_path(client: httpx.Client, sid: str) -> list[str]:
    """``sid`` and its ancestors, root first."""
    path: list[str] = []
    while sid:
        path.insert(0, sid)
        d = client.get(f"{API}/subjects/{sid}/").raise_for_status().json()["data"]
        sid = ((d["relationships"].get("parent") or {}).get("data") or {}).get("id", "")
    return path


def plan(client: httpx.Client) -> dict:
    subjects: list[str] = []
    for sid in LEAF_SUBJECTS:
        for s in subject_path(client, sid):
            if s not in subjects:
                subjects.append(s)
    return {"tags": TAGS, "license": LICENSE_ID, "subjects": subjects}


def apply(client: httpx.Client, kind: str, guid: str, p: dict) -> None:
    body = {
        "data": {
            "type": kind,
            "id": guid,
            "attributes": {"tags": p["tags"]},
            "relationships": {
                "license": {"data": {"type": "licenses", "id": p["license"]}}
            },
        }
    }
    client.patch(f"{API}/{kind}/{guid}/", json=body).raise_for_status()
    rel = {"data": [{"type": "subjects", "id": s} for s in p["subjects"]]}
    client.patch(
        f"{API}/{kind}/{guid}/relationships/subjects/", json=rel
    ).raise_for_status()


def verify(client: httpx.Client, kind: str, guid: str) -> dict:
    d = (
        client.get(f"{API}/{kind}/{guid}/?embed=license")
        .raise_for_status()
        .json()["data"]
    )
    subs = (
        client.get(f"{API}/{kind}/{guid}/subjects/").raise_for_status().json()["data"]
    )
    lic = (d.get("embeds", {}).get("license", {}).get("data") or {}).get(
        "attributes", {}
    )
    return {
        "tags": d["attributes"].get("tags"),
        "license": lic.get("name"),
        "subjects": sorted(s["attributes"]["text"] for s in subs),
    }


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--apply", action="store_true")
    args = ap.parse_args()
    token = os.environ.get("OSF_TOKEN", "")
    headers = dict(UA)
    if token:
        headers["Authorization"] = f"Bearer {token}"
    with httpx.Client(headers=headers, timeout=60, follow_redirects=True) as client:
        p = plan(client)
        print(json.dumps(p, indent=1))
        if not args.apply:
            print("dry run — pass --apply to write")
            return 0
        if not token:
            print("OSF_TOKEN is not set", file=sys.stderr)
            return 2
        for kind, guid in TARGETS:
            apply(client, kind, guid, p)
            print(kind, guid, verify(client, kind, guid))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())