Generalize Zenodo publish script for future releases

scripts/publish_zenodo_release.py

@@ -13,6 +13,7 @@
 import argparse
 import json
 import os
+import re
 import sys
 import tempfile
 import urllib.error
@@ -22,129 +23,165 @@
 
 ZENODO_API = "https://zenodo.org/api"
 GITHUB_REPO = "brunomartinsmv/eto-methods-comparison"
-CHANGELOG_ANCHOR = "200---2026-07-12"
+DEFAULT_TIMEOUT = 120
+DEFAULT_ZENODO_JSON = Path(".zenodo.json")
+DEFAULT_CHANGELOG = Path("CHANGELOG.md")
 
 
 class ZenodoError(RuntimeError):
     pass
 
 
+def _github_heading_anchor(heading: str) -> str:
+    slug = re.sub(r"[^\w\s-]", "", heading.lower())
+    return re.sub(r"\s+", "-", slug.strip())
+
+
+def _changelog_entry_for_tag(tag: str, changelog_path: Path = DEFAULT_CHANGELOG) -> tuple[str, str]:
+    version = tag.removeprefix("v")
+    pattern = re.compile(
+        rf"^## \[{re.escape(version)}\] - (\d{{4}}-\d{{2}}-\d{{2}})\s*$",
+        re.MULTILINE,
+    )
+    match = pattern.search(changelog_path.read_text(encoding="utf-8"))
+    if match is None:
+        raise ZenodoError(
+            f"No CHANGELOG entry found for {tag!r} in {changelog_path}. "
+            f"Expected a heading like '## [{version}] - YYYY-MM-DD'."
+        )
+    heading = match.group(0).lstrip("#").strip()
+    return match.group(1), _github_heading_anchor(heading)
+
+
+def _build_metadata(
+    tag: str,
+    *,
+    zenodo_path: Path = DEFAULT_ZENODO_JSON,
+    changelog_path: Path = DEFAULT_CHANGELOG,
+) -> dict:
+    base = json.loads(zenodo_path.read_text(encoding="utf-8"))
+    publication_date, anchor = _changelog_entry_for_tag(tag, changelog_path)
+    changelog_url = (
+        f"https://github.com/{GITHUB_REPO}/blob/{tag}/CHANGELOG.md#{anchor}"
+    )
+    description = base["description"]
+    if not description.startswith("<p>"):
+        description = f"<p>{description}</p>"
+    base["description"] = (
+        f"{description}"
+        f'<p><strong>Changelog:</strong> <a href="{changelog_url}">{changelog_url}</a></p>'
+    )
+    base["publication_date"] = publication_date
+    base["version"] = tag
+    base["related_identifiers"] = [
+        {
+            "relation": "isSupplementTo",
+            "identifier": f"https://github.com/{GITHUB_REPO}/tree/{tag}",
+            "resource_type": "software",
+            "scheme": "url",
+        }
+    ]
+    return {"metadata": base}
+
+
 def _request(
     method: str,
     url: str,
     *,
     token: str,
     data: bytes | None = None,
     headers: dict[str, str] | None = None,
+    timeout: float = DEFAULT_TIMEOUT,
 ) -> dict:
     request_headers = {"Authorization": f"Bearer {token}"}
     if headers:
         request_headers.update(headers)
     request = urllib.request.Request(url, data=data, headers=request_headers, method=method)
     try:
-        with urllib.request.urlopen(request) as response:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
             body = response.read().decode("utf-8")
             return json.loads(body) if body else {}
     except urllib.error.HTTPError as exc:
         detail = exc.read().decode("utf-8", errors="replace")
         raise ZenodoError(f"{method} {url} failed ({exc.code}): {detail}") from exc
+    except urllib.error.URLError as exc:
+        raise ZenodoError(f"{method} {url} failed: {exc.reason}") from exc
 
 
-def _download(url: str, destination: Path) -> None:
-    with urllib.request.urlopen(url) as response, destination.open("wb") as handle:
-        handle.write(response.read())
+def _download(url: str, destination: Path, *, timeout: float = DEFAULT_TIMEOUT) -> None:
+    try:
+        with urllib.request.urlopen(url, timeout=timeout) as response, destination.open("wb") as handle:
+            handle.write(response.read())
+    except urllib.error.HTTPError as exc:
+        detail = exc.read().decode("utf-8", errors="replace")
+        raise ZenodoError(f"GET {url} failed ({exc.code}): {detail}") from exc
+    except urllib.error.URLError as exc:
+        raise ZenodoError(f"GET {url} failed: {exc.reason}") from exc
 
 
-def _delete_inherited_files(deposition: dict, token: str) -> None:
+def _delete_inherited_files(deposition: dict, token: str, *, timeout: float) -> None:
     for file_info in deposition.get("files", []):
-        _request("DELETE", file_info["links"]["self"], token=token)
+        _request("DELETE", file_info["links"]["self"], token=token, timeout=timeout)
 
 
-def _build_metadata(tag: str) -> dict:
-    changelog_url = (
-        f"https://github.com/{GITHUB_REPO}/blob/{tag}/CHANGELOG.md#{CHANGELOG_ANCHOR}"
-    )
-    return {
-        "metadata": {
-            "title": f"{GITHUB_REPO}: {tag}",
-            "upload_type": "software",
-            "description": (
-                "<p>Major release with full ET0 computation pipeline, UX commands, "
-                "uncertainty and sensitivity diagnostics, safety-net CI, and LaTeX "
-                f"equation documentation. <strong>Full changelog:</strong> "
-                f'<a href="{changelog_url}">{changelog_url}</a></p>'
-            ),
-            "publication_date": "2026-07-12",
-            "version": tag,
-            "access_right": "open",
-            "license": "mit",
-            "creators": [
-                {
-                    "name": "Vieira, Bruno Martins M.",
-                    "affiliation": "Universidade Federal do Mato Grosso",
-                }
-            ],
-            "keywords": [
-                "evapotranspiration",
-                "reference evapotranspiration",
-                "Penman-Monteith",
-                "FAO-56",
-                "reproducible research",
-                "scientific computing",
-            ],
-            "related_identifiers": [
-                {
-                    "relation": "isSupplementTo",
-                    "identifier": f"https://github.com/{GITHUB_REPO}/tree/{tag}",
-                    "resource_type": "software",
-                    "scheme": "url",
-                }
-            ],
-        }
-    }
-
-
-def publish_release(*, tag: str, deposition_id: str, token: str) -> dict:
+def publish_release(
+    *,
+    tag: str,
+    deposition_id: str,
+    token: str,
+    dry_run: bool = False,
+    timeout: float = DEFAULT_TIMEOUT,
+    zenodo_path: Path = DEFAULT_ZENODO_JSON,
+    changelog_path: Path = DEFAULT_CHANGELOG,
+) -> dict:
     new_version = _request(
         "POST",
         f"{ZENODO_API}/deposit/depositions/{deposition_id}/actions/newversion",
         token=token,
+        timeout=timeout,
     )
     draft_url = new_version["links"]["latest_draft"]
-    draft = _request("GET", draft_url, token=token)
+    draft = _request("GET", draft_url, token=token, timeout=timeout)
     draft_id = str(draft["id"])
     bucket_url = draft["links"]["bucket"]
 
-    _delete_inherited_files(draft, token)
+    _delete_inherited_files(draft, token, timeout=timeout)
 
     archive_name = f"{GITHUB_REPO.replace('/', '-')}-{tag}.zip"
     archive_url = f"https://github.com/{GITHUB_REPO}/archive/refs/tags/{tag}.zip"
     with tempfile.TemporaryDirectory() as tmp_dir:
         archive_path = Path(tmp_dir) / archive_name
-        _download(archive_url, archive_path)
+        _download(archive_url, archive_path, timeout=timeout)
         with archive_path.open("rb") as handle:
             _request(
                 "PUT",
                 f"{bucket_url}/{urllib.parse.quote(archive_name)}",
                 token=token,
                 data=handle.read(),
                 headers={"Content-Type": "application/octet-stream"},
+                timeout=timeout,
             )
 
-    _request(
+    updated = _request(
         "PUT",
         f"{ZENODO_API}/deposit/depositions/{draft_id}",
         token=token,
-        data=json.dumps(_build_metadata(tag)).encode("utf-8"),
+        data=json.dumps(_build_metadata(tag, zenodo_path=zenodo_path, changelog_path=changelog_path)).encode(
+            "utf-8"
+        ),
         headers={"Content-Type": "application/json"},
+        timeout=timeout,
     )
 
-    published = _request(
+    if dry_run:
+        return updated
+
+    return _request(
         "POST",
         f"{ZENODO_API}/deposit/depositions/{draft_id}/actions/publish",
         token=token,
+        timeout=timeout,
     )
-    return published
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -160,25 +197,39 @@ def main(argv: list[str] | None = None) -> int:
         default=os.environ.get("ZENODO_ACCESS_TOKEN", ""),
         help="Zenodo personal access token (or set ZENODO_ACCESS_TOKEN)",
     )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Create and update a draft deposition without publishing",
+    )
+    parser.add_argument(
+        "--timeout",
+        type=float,
+        default=DEFAULT_TIMEOUT,
+        help=f"HTTP timeout in seconds (default: {DEFAULT_TIMEOUT})",
+    )
     args = parser.parse_args(argv)
 
     if not args.token:
         print("Missing Zenodo token. Set ZENODO_ACCESS_TOKEN or pass --token.", file=sys.stderr)
         return 1
 
-    published = publish_release(tag=args.tag, deposition_id=args.deposition_id, token=args.token)
-    doi = published.get("doi", "")
-    record_id = published.get("id", "")
-    print(
-        json.dumps(
-            {
-                "doi": doi,
-                "record_id": record_id,
-                "record_url": published.get("links", {}).get("record_html"),
-            },
-            indent=2,
-        )
+    result = publish_release(
+        tag=args.tag,
+        deposition_id=args.deposition_id,
+        token=args.token,
+        dry_run=args.dry_run,
+        timeout=args.timeout,
     )
+    output = {
+        "doi": result.get("doi", ""),
+        "record_id": result.get("id", ""),
+        "record_url": result.get("links", {}).get("record_html"),
+        "state": result.get("state", ""),
+    }
+    if args.dry_run:
+        output["draft_url"] = result.get("links", {}).get("self")
+    print(json.dumps(output, indent=2))
     return 0
 
 

tests/test_publish_zenodo_release.py

@@ -0,0 +1,112 @@
+"""Tests for Zenodo release publishing helpers."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from scripts import publish_zenodo_release as zenodo
+
+
+def test_build_metadata_v200_uses_zenodo_json_and_changelog() -> None:
+    metadata = zenodo._build_metadata("v2.0.0")["metadata"]
+
+    assert metadata["version"] == "v2.0.0"
+    assert metadata["publication_date"] == "2026-07-12"
+    assert "Reproducible Python pipeline" in metadata["description"]
+    assert "CHANGELOG.md#200---2026-07-12" in metadata["description"]
+    assert metadata["related_identifiers"][0]["identifier"].endswith("/tree/v2.0.0")
+    assert metadata["license"] == "mit"
+
+
+def test_changelog_entry_for_tag_unknown_version_raises(tmp_path: Path) -> None:
+    changelog = tmp_path / "CHANGELOG.md"
+    changelog.write_text("## [Unreleased]\n\n## [2.0.0] - 2026-07-12\n", encoding="utf-8")
+
+    with pytest.raises(zenodo.ZenodoError, match="No CHANGELOG entry found"):
+        zenodo._changelog_entry_for_tag("v9.9.9", changelog)
+
+
+def test_github_heading_anchor_matches_github_slug() -> None:
+    assert zenodo._github_heading_anchor("[2.0.0] - 2026-07-12") == "200---2026-07-12"
+
+
+def test_download_http_error_raises_zenodo_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    import urllib.error
+
+    def fake_urlopen(*_args, **_kwargs):
+        raise urllib.error.HTTPError(
+            "https://example.com/archive.zip",
+            404,
+            "Not Found",
+            hdrs=None,
+            fp=MagicMock(read=lambda: b"release not found"),
+        )
+
+    monkeypatch.setattr(zenodo.urllib.request, "urlopen", fake_urlopen)
+
+    with pytest.raises(zenodo.ZenodoError, match="GET .* failed \\(404\\)"):
+        zenodo._download("https://example.com/archive.zip", tmp_path / "archive.zip")
+
+
+def test_publish_release_dry_run_skips_publish(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    calls: list[tuple[str, str]] = []
+
+    def fake_request(method: str, url: str, **_kwargs) -> dict:
+        calls.append((method, url))
+        if method == "POST" and url.endswith("/actions/newversion"):
+            return {"links": {"latest_draft": "https://zenodo.org/api/deposit/depositions/99"}}
+        if method == "GET":
+            return {
+                "id": 99,
+                "links": {"bucket": "https://zenodo.org/api/files/bucket/abc"},
+                "files": [{"links": {"self": "https://zenodo.org/api/files/1"}}],
+            }
+        if method == "DELETE":
+            return {}
+        if method == "PUT" and "/depositions/" in url:
+            return {
+                "id": 99,
+                "state": "draft",
+                "links": {
+                    "self": "https://zenodo.org/api/deposit/depositions/99",
+                    "record_html": "https://zenodo.org/deposit/99",
+                },
+            }
+        if method == "PUT":
+            return {}
+        raise AssertionError(f"Unexpected request: {method} {url}")
+
+    def fake_download(url: str, destination: Path, **_kwargs) -> None:
+        destination.write_bytes(b"archive")
+
+    monkeypatch.setattr(zenodo, "_request", fake_request)
+    monkeypatch.setattr(zenodo, "_download", fake_download)
+
+    zenodo_json = tmp_path / ".zenodo.json"
+    changelog = tmp_path / "CHANGELOG.md"
+    zenodo_json.write_text(
+        json.dumps({"description": "Test description", "license": "mit"}),
+        encoding="utf-8",
+    )
+    changelog.write_text("## [1.0.0] - 2026-02-11\n", encoding="utf-8")
+
+    result = zenodo.publish_release(
+        tag="v1.0.0",
+        deposition_id="18615164",
+        token="token",
+        dry_run=True,
+        zenodo_path=zenodo_json,
+        changelog_path=changelog,
+    )
+
+    assert result["state"] == "draft"
+    assert not any(method == "POST" and url.endswith("/actions/publish") for method, url in calls)
+
+
+def test_main_missing_token_returns_error(capsys: pytest.CaptureFixture[str]) -> None:
+    assert zenodo.main(["--tag", "v2.0.0", "--token", ""]) == 1
+    assert "Missing Zenodo token" in capsys.readouterr().err