diff --git a/README.md b/README.md index 1c865c5..30f4f5b 100644 --- a/README.md +++ b/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 diff --git a/tests/test_repo_from_related.py b/tests/test_repo_from_related.py new file mode 100644 index 0000000..26826b4 --- /dev/null +++ b/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() diff --git a/zenodo_maint/api.py b/zenodo_maint/api.py index 27a808e..dc385f5 100644 --- a/zenodo_maint/api.py +++ b/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/ (not a different repo that merely shares a prefix, e.g. vcell vs vcell-solvers).""" diff --git a/zenodo_maint/cli.py b/zenodo_maint/cli.py index 5820e30..4247e6c 100644 --- a/zenodo_maint/cli.py +++ b/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"