#!/usr/bin/env python3 """Move issue labels from repository-local labels to organization labels.""" from __future__ import annotations import argparse import os import pathlib import re import sys from typing import Any from gitea_common import ( GiteaClient, GiteaError, infer_target, load_dotenv, org_path, repo_path, repo_root, require_token, ) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=pathlib.Path, default=pathlib.Path.cwd(), help="repository root used for Gitea URL inference") parser.add_argument("--remote", default="origin", help="git remote to use for target inference") parser.add_argument("--env-file", type=pathlib.Path, default=pathlib.Path("/home/zemion/.config/gitea/gitea.env")) parser.add_argument("--org", help="organization owner; defaults to inferred owner") parser.add_argument("--repo", action="append", default=[], help="repository name to process; may be repeated") parser.add_argument("--repo-regex", default=".*", help="only process repositories whose name matches this regex") parser.add_argument( "--issue-mode", choices=("replace", "delete-add"), default="replace", help="replace labels in one request, or delete local ids before adding org ids", ) parser.add_argument("--delete-repo-labels", action="store_true", help="delete matching repository-local labels after issue and PR migration") parser.add_argument("--apply", action="store_true", help="apply changes; omit for dry-run") args = parser.parse_args() try: root = repo_root(args.root) load_dotenv(args.env_file or root / ".env") target = infer_target(root, args.remote) owner = args.org or target.owner client = GiteaClient(target, require_token() if args.apply else os.environ.get("GITEA_TOKEN")) if client.token is None: raise GiteaError("GITEA_TOKEN is required for dry-run and apply because migration inspects live issue labels.") repo_filter = re.compile(args.repo_regex) org_labels = _labels_by_name(client.paginate(org_path(owner, "/labels"), limit=100)) if not org_labels: raise GiteaError(f"{owner} has no organization labels") repos = _selected_repositories(client, owner, args.repo, repo_filter) totals = Totals() print(f"Organization: {owner}") print(f"Repositories: {len(repos)}") print(f"Mode: {'apply' if args.apply else 'dry-run'}") print(f"Delete repository labels: {args.delete_repo_labels}") for index, repo in enumerate(repos, start=1): repo_result = migrate_repository( client, owner=owner, repo=repo, org_labels=org_labels, issue_mode=args.issue_mode, apply=args.apply, delete_repo_labels=args.delete_repo_labels, ) totals.add(repo_result) if repo_result.has_work: print( f"[{index}/{len(repos)}] {repo}: " f"issue-labels={repo_result.issue_label_migrations}, " f"delete-labels={repo_result.repo_label_deletions}, " f"errors={len(repo_result.errors)}" ) for error in repo_result.errors: print(f" error: {error}") else: print(f"[{index}/{len(repos)}] {repo}: no matching local labels") print("Summary:") print(f" repositories processed: {totals.repositories}") print(f" repositories with matching local labels: {totals.repositories_with_work}") print(f" issue/PR local labels migrated: {totals.issue_label_migrations}") print(f" repository labels deleted: {totals.repo_label_deletions}") print(f" errors: {totals.errors}") if totals.errors: return 1 return 0 except GiteaError as exc: print(f"error: {exc}", file=sys.stderr) return 1 class RepoResult: def __init__(self) -> None: self.issue_label_migrations = 0 self.repo_label_deletions = 0 self.errors: list[str] = [] self.matched_local_labels = 0 @property def has_work(self) -> bool: return bool(self.matched_local_labels or self.issue_label_migrations or self.repo_label_deletions or self.errors) class Totals: def __init__(self) -> None: self.repositories = 0 self.repositories_with_work = 0 self.issue_label_migrations = 0 self.repo_label_deletions = 0 self.errors = 0 def add(self, result: RepoResult) -> None: self.repositories += 1 if result.has_work: self.repositories_with_work += 1 self.issue_label_migrations += result.issue_label_migrations self.repo_label_deletions += result.repo_label_deletions self.errors += len(result.errors) def migrate_repository( client: GiteaClient, *, owner: str, repo: str, org_labels: dict[str, dict[str, Any]], issue_mode: str, apply: bool, delete_repo_labels: bool, ) -> RepoResult: result = RepoResult() repo_labels = client.paginate(repo_path(owner, repo, "/labels"), limit=100) local_labels = { str(label.get("name") or ""): label for label in repo_labels if str(label.get("name") or "") in org_labels } result.matched_local_labels = len(local_labels) if not local_labels: return result local_by_id = {_label_id(label): label for label in local_labels.values() if _label_id(label) is not None} org_by_name = {name: _label_id(label) for name, label in org_labels.items()} for issue_type in ("issues", "pulls"): issues = client.paginate( repo_path(owner, repo, "/issues"), query={"state": "all", "type": issue_type}, limit=100, ) for issue in issues: issue_number = int(issue.get("number") or issue.get("index")) issue_label_ids = [_label_id(label) for label in issue.get("labels") or []] issue_label_ids = [label_id for label_id in issue_label_ids if label_id is not None] final_label_ids = list(issue_label_ids) changed = False migrated_count = 0 local_ids_to_remove: list[int] = [] org_ids_to_add: list[int] = [] for label in issue.get("labels") or []: local_id = _label_id(label) if local_id is None or local_id not in local_by_id: continue name = str(label.get("name") or "") org_id = org_by_name.get(name) if org_id is None: continue local_ids_to_remove.append(local_id) if org_id not in issue_label_ids and org_id not in org_ids_to_add: org_ids_to_add.append(org_id) final_label_ids = [label_id for label_id in final_label_ids if label_id != local_id] if org_id not in final_label_ids: final_label_ids.append(org_id) changed = True migrated_count += 1 if changed: if apply: if issue_mode == "delete-add": for local_id in local_ids_to_remove: _remove_issue_label(client, owner, repo, issue_number, local_id) if org_ids_to_add: client.request_json( "POST", repo_path(owner, repo, f"/issues/{issue_number}/labels"), body={"labels": org_ids_to_add}, ) else: client.request_json( "PUT", repo_path(owner, repo, f"/issues/{issue_number}/labels"), body={"labels": final_label_ids}, ) result.issue_label_migrations += migrated_count if delete_repo_labels: for name, label in sorted(local_labels.items()): label_id = _label_id(label) if label_id is None: result.errors.append(f"{name} has no label id") continue if apply: try: client.request_json("DELETE", repo_path(owner, repo, f"/labels/{label_id}")) except GiteaError as exc: result.errors.append(f"delete repository label {name}: {exc}") continue result.repo_label_deletions += 1 return result def _selected_repositories(client: GiteaClient, owner: str, explicit: list[str], repo_filter: re.Pattern[str]) -> list[str]: if explicit: return sorted(set(explicit)) repos = client.paginate(org_path(owner, "/repos"), query={"type": "all"}, limit=100) names = sorted(str(repo.get("name") or "") for repo in repos if repo.get("name")) return [name for name in names if repo_filter.search(name)] def _labels_by_name(labels: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: return {str(label.get("name")): label for label in labels if label.get("name") and _label_id(label) is not None} def _label_id(label: dict[str, Any]) -> int | None: value = label.get("id") or label.get("index") if value is None: return None return int(value) def _remove_issue_label(client: GiteaClient, owner: str, repo: str, issue_number: int, label_id: int) -> None: try: client.request_json("DELETE", repo_path(owner, repo, f"/issues/{issue_number}/labels/{label_id}")) except GiteaError as exc: if "HTTP 404" in str(exc): return raise if __name__ == "__main__": raise SystemExit(main())