"""Publish this repository as a new Zenodo deposition version."""
from __future__ import annotations
import os
import subprocess
import sys
from http import HTTPStatus
from pathlib import Path
import requests
import yaml
REPO_ROOT = Path(__file__).resolve().parents[3]
CITATION_PATH = REPO_ROOT / "CITATION.cff"
ZENODO_ID_PATH = REPO_ROOT / ".zenodo_id"
ZENODO_API_BASE = "https://zenodo.org/api"
# Seconds to wait on Zenodo API calls; the archive upload gets a longer budget than the metadata calls.
REQUEST_TIMEOUT = 60
UPLOAD_TIMEOUT = 600
[docs]
def load_citation(citation_path: Path) -> dict:
return yaml.safe_load(citation_path.read_text())
[docs]
def require_token() -> str:
token = os.environ.get("ZENODO_TOKEN", "").strip()
if not token:
msg = "ZENODO_TOKEN environment variable is not set"
raise SystemExit(msg)
return token
[docs]
def read_deposition_id() -> str | None:
if not ZENODO_ID_PATH.is_file():
return None
content = ZENODO_ID_PATH.read_text().strip()
return content or None
[docs]
def raise_for_zenodo_error(response: requests.Response) -> None:
if response.status_code >= HTTPStatus.BAD_REQUEST:
print(response.text, file=sys.stderr) # noqa: T201
raise SystemExit(1)
[docs]
def _auth_params(token: str) -> dict:
return {"access_token": token}
[docs]
def create_deposition(token: str) -> dict:
response = requests.post(
f"{ZENODO_API_BASE}/deposit/depositions",
params=_auth_params(token),
json={},
timeout=REQUEST_TIMEOUT,
)
raise_for_zenodo_error(response)
return response.json()
[docs]
def create_new_version(token: str, deposition_id: str) -> dict:
response = requests.post(
f"{ZENODO_API_BASE}/deposit/depositions/{deposition_id}/actions/newversion",
params=_auth_params(token),
timeout=REQUEST_TIMEOUT,
)
raise_for_zenodo_error(response)
draft_url = response.json()["links"]["latest_draft"]
draft_response = requests.get(draft_url, params=_auth_params(token), timeout=REQUEST_TIMEOUT)
raise_for_zenodo_error(draft_response)
return draft_response.json()
[docs]
def build_archive(version: str) -> Path:
archive_path = REPO_ROOT / f"uulabscheduler-{version}.zip"
subprocess.run( # noqa: S603 -- fixed argv, no untrusted input
["git", "archive", "--format=zip", "-o", str(archive_path), "HEAD"], # noqa: S607 -- git resolved from PATH
cwd=REPO_ROOT,
check=True,
)
return archive_path
[docs]
def remove_existing_files(token: str, draft: dict) -> None:
for file_entry in draft.get("files", []):
file_id = file_entry["id"]
deposition_id = draft["id"]
response = requests.delete(
f"{ZENODO_API_BASE}/deposit/depositions/{deposition_id}/files/{file_id}",
params=_auth_params(token),
timeout=REQUEST_TIMEOUT,
)
raise_for_zenodo_error(response)
[docs]
def upload_file(token: str, draft: dict, file_path: Path) -> None:
bucket_url = draft["links"]["bucket"]
with open(file_path, "rb") as fh:
response = requests.put(
f"{bucket_url}/{file_path.name}",
params=_auth_params(token),
data=fh,
timeout=UPLOAD_TIMEOUT,
)
raise_for_zenodo_error(response)
[docs]
def publish(token: str, draft: dict) -> dict:
response = requests.post(
draft["links"]["publish"],
params=_auth_params(token),
timeout=REQUEST_TIMEOUT,
)
raise_for_zenodo_error(response)
return response.json()
[docs]
def main() -> None:
token = require_token()
citation = load_citation(CITATION_PATH)
metadata = build_zenodo_metadata(citation)
version = citation["version"]
existing_id = read_deposition_id()
if existing_id is None:
draft = create_deposition(token)
else:
draft = create_new_version(token, existing_id)
remove_existing_files(token, draft)
archive_path = build_archive(version)
upload_file(token, draft, archive_path)
update_metadata(token, draft, metadata)
record = publish(token, draft)
if existing_id is None:
ZENODO_ID_PATH.write_text(f"{record['id']}\n")
print(f"Published Zenodo record: {record['id']}") # noqa: T201
print(f"DOI: {record['doi']}") # noqa: T201
if __name__ == "__main__":
main()
