Add list-owned: enumerate every concept/DOI the account owns

README.md

@@ -41,6 +41,8 @@ you can omit `--concept`/`--repo` entirely:
 
 ```bash
 zenodo-maint verify-token
+zenodo-maint list-owned                     # every concept/DOI this account owns + source repo
+zenodo-maint list-owned --repo-only --json  # machine-readable; only GitHub-linked records
 zenodo-maint list-versions                 # concept from CITATION.cff doi:
 zenodo-maint check-drift                    # repo from CITATION.cff / $GITHUB_REPOSITORY
 

tests/test_repo_from_related.py

@@ -0,0 +1,52 @@
+"""Unit tests for api.repo_from_related — the GitHub owner/repo extractor used by
+`list-owned` to map an owned Zenodo concept back to its source repository.
+
+Stdlib `unittest` only. Run with:  python -m unittest discover -s tests
+"""
+import unittest
+
+from zenodo_maint.api import repo_from_related
+
+
+class RepoFromRelated(unittest.TestCase):
+    def _md(self, *identifiers):
+        return {"related_identifiers": [
+            {"relation": "isSupplementTo", "identifier": i, "scheme": "url"}
+            for i in identifiers
+        ]}
+
+    def test_tree_tag_url(self) -> None:
+        md = self._md("https://github.com/virtualcell/vcell/tree/v8.0")
+        self.assertEqual(repo_from_related(md), "virtualcell/vcell")
+
+    def test_bare_repo_url(self) -> None:
+        self.assertEqual(repo_from_related(self._md("https://github.com/org/repo")), "org/repo")
+
+    def test_git_suffix_stripped(self) -> None:
+        self.assertEqual(repo_from_related(self._md("https://github.com/org/repo.git")), "org/repo")
+
+    def test_only_owner_repo_segments(self) -> None:
+        # deeper paths must not leak into the repo id
+        md = self._md("https://github.com/biosimulators/Biosimulators_utils/releases/tag/0.2.3")
+        self.assertEqual(repo_from_related(md), "biosimulators/Biosimulators_utils")
+
+    def test_picks_github_among_others(self) -> None:
+        md = {"related_identifiers": [
+            {"relation": "continues", "identifier": "10.5281/zenodo.123", "scheme": "doi"},
+            {"relation": "isSupplementTo",
+             "identifier": "https://github.com/cam-center/SpringSaLaD/tree/2.4.6", "scheme": "url"},
+        ]}
+        self.assertEqual(repo_from_related(md), "cam-center/SpringSaLaD")
+
+    def test_none_when_no_github(self) -> None:
+        md = {"related_identifiers": [
+            {"relation": "continues", "identifier": "10.5281/zenodo.123", "scheme": "doi"}]}
+        self.assertIsNone(repo_from_related(md))
+
+    def test_none_on_empty(self) -> None:
+        self.assertIsNone(repo_from_related({}))
+        self.assertIsNone(repo_from_related({"related_identifiers": None}))
+
+
+if __name__ == "__main__":
+    unittest.main()

zenodo_maint/api.py

@@ -103,6 +103,24 @@ def owned_records(self, size: int = 50) -> list[dict[str, Any]]:
         st, d = self._call("GET", f"/deposit/depositions?size={size}&sort=mostrecent")
         return list(self._expect((200,), st, d, "list owned depositions"))
 
+    def owned_records_all(self, page_size: int = 100) -> list[dict[str, Any]]:
+        """Every deposition the token's account owns, across all pages. Unlike
+        owned_records(), this walks the full result set so accounts with many
+        records aren't silently truncated. Callers dedupe by concept as needed."""
+        out: list[dict[str, Any]] = []
+        page = 1
+        while True:
+            st, d = self._call(
+                "GET", f"/deposit/depositions?size={page_size}&page={page}&sort=mostrecent")
+            self._expect((200,), st, d, "list owned depositions")
+            if not d:
+                break
+            out.extend(d)
+            if len(d) < page_size:
+                break
+            page += 1
+        return out
+
     def concept_versions(self, concept_recid: str) -> list[dict[str, Any]]:
         """All published depositions in a concept, oldest->newest by created.
 
@@ -250,6 +268,24 @@ def concept_ids_in_related(metadata: dict[str, Any]) -> set[str]:
     return out
 
 
+def repo_from_related(metadata: dict[str, Any]) -> str | None:
+    """Extract GitHub 'owner/repo' from a record's related_identifiers — the
+    `isSupplementTo` github.com link the archiver sets on every version. Returns
+    None if the record has no GitHub link (e.g. a non-software deposit)."""
+    for r in metadata.get("related_identifiers", []) or []:
+        ident = str(r.get("identifier", ""))
+        idx = ident.find("github.com/")
+        if idx < 0:
+            continue
+        path = ident[idx + len("github.com/"):].split("#")[0].split("?")[0].strip("/")
+        if path.endswith(".git"):
+            path = path[:-4]
+        parts = [p for p in path.split("/") if p]
+        if len(parts) >= 2:
+            return f"{parts[0]}/{parts[1]}"
+    return None
+
+
 def _identifier_is_repo(identifier: str, repo: str) -> bool:
     """True if a related-identifier URL points at exactly github.com/<repo> (not a
     different repo that merely shares a prefix, e.g. vcell vs vcell-solvers)."""

zenodo_maint/cli.py

@@ -113,6 +113,46 @@ def cmd_verify_token(args: argparse.Namespace) -> None:
         print(f"  - {title:44} v{m.get('version')}  concept={r.get('conceptrecid')}")
 
 
+def cmd_list_owned(args: argparse.Namespace) -> None:
+    """Every Zenodo concept this account owns, deduped to the latest version, with
+    the source GitHub repo — the authoritative 'which repos are tracked' list."""
+    cli = _client(args)
+    deps = cli.owned_records_all()
+    latest: dict[str, dict[str, Any]] = {}   # concept recid -> latest deposition
+    for d in deps:
+        cid = str(d.get("conceptrecid") or d.get("id"))
+        cur = latest.get(cid)
+        if cur is None or d.get("created", "") > cur.get("created", ""):
+            latest[cid] = d
+    rows = []
+    for cid, d in latest.items():
+        m = d.get("metadata", {})
+        rows.append({
+            "repo": api.repo_from_related(m) or "",
+            "concept_doi": d.get("conceptdoi") or "",
+            "concept_recid": cid,
+            "latest_doi": m.get("doi") or d.get("doi") or "",
+            "version": m.get("version") or "",
+            "date": m.get("publication_date") or "",
+            "title": m.get("title") or "",
+        })
+    if args.repo_only:
+        rows = [r for r in rows if r["repo"]]
+    rows.sort(key=lambda r: (r["repo"] == "", r["repo"].lower(), r["title"].lower()))
+
+    if args.json:
+        print(json.dumps(rows, indent=2))
+        return
+    linked = sum(1 for r in rows if r["repo"])
+    print(f"account owns {len(latest)} concept(s); {linked} linked to a GitHub repo:\n")
+    w = max([len(r["repo"]) for r in rows] + [4])
+    print(f"{'repo':{w}}  {'concept DOI':22}  {'latest':12}  {'date':10}  title")
+    print("-" * (w + 62))
+    for r in rows:
+        print(f"{r['repo'] or '—':{w}}  {(r['concept_doi'] or r['concept_recid']):22}  "
+              f"{r['version'][:12]:12}  {r['date']:10}  {r['title'][:50]}")
+
+
 def cmd_list_versions(args: argparse.Namespace) -> None:
     cli = _client(args)
     for x in cli.concept_versions(_concept(cli, args)):
@@ -493,6 +533,12 @@ def build_parser() -> argparse.ArgumentParser:
     sub = p.add_subparsers(dest="cmd", required=True)
 
     sub.add_parser("verify-token").set_defaults(func=cmd_verify_token)
+    lo = sub.add_parser("list-owned",
+                        help="list every concept/DOI this account owns, with source repo")
+    lo.add_argument("--json", action="store_true", help="output JSON")
+    lo.add_argument("--repo-only", action="store_true",
+                    help="only records linked to a GitHub repo")
+    lo.set_defaults(func=cmd_list_owned)
     sub.add_parser("list-versions").set_defaults(func=cmd_list_versions)
     sub.add_parser("check-drift").set_defaults(func=cmd_check_drift)
     sub.add_parser("doctor", help="check for integration conflicts, forks, and drift"