Publish training skills collection

gitea-issues/scripts/create_repo.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Create or return a Gitea repository using env vars or a local env file."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+from gitea_common import DEFAULT_ENV_FILE, auth_config
+
+
+def request_json(method: str, url: str, token: str, payload: dict[str, Any] | None = None) -> Any:
+    data = None if payload is None else json.dumps(payload).encode("utf-8")
+    request = urllib.request.Request(url, data=data, method=method)
+    request.add_header("Authorization", f"token {token}")
+    request.add_header("Accept", "application/json")
+    if payload is not None:
+        request.add_header("Content-Type", "application/json")
+
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            body = response.read().decode("utf-8")
+    except urllib.error.HTTPError as exc:
+        error_body = exc.read().decode("utf-8", errors="replace")
+        raise GiteaHTTPError(exc.code, error_body) from exc
+    except urllib.error.URLError as exc:
+        raise SystemExit(f"Gitea API request failed: {exc.reason}") from exc
+
+    return json.loads(body) if body else None
+
+
+class GiteaHTTPError(Exception):
+    def __init__(self, status: int, body: str) -> None:
+        super().__init__(f"Gitea API request failed: HTTP {status}\n{body}")
+        self.status = status
+        self.body = body
+
+
+def api_root(base_url: str) -> str:
+    return f"{base_url.rstrip('/')}/api/v1"
+
+
+def repo_url(base_url: str, owner: str, repo: str) -> str:
+    encoded_owner = urllib.parse.quote(owner, safe="")
+    encoded_repo = urllib.parse.quote(repo, safe="")
+    return f"{api_root(base_url)}/repos/{encoded_owner}/{encoded_repo}"
+
+
+def get_authenticated_user(base_url: str, token: str) -> dict[str, Any]:
+    return request_json("GET", f"{api_root(base_url)}/user", token)
+
+
+def get_repo(base_url: str, token: str, owner: str, repo: str) -> dict[str, Any] | None:
+    try:
+        return request_json("GET", repo_url(base_url, owner, repo), token)
+    except GiteaHTTPError as exc:
+        if exc.status == 404:
+            return None
+        raise SystemExit(str(exc)) from exc
+
+
+def create_repo(
+    base_url: str,
+    token: str,
+    owner: str,
+    repo: str,
+    description: str,
+    private: bool,
+    default_branch: str,
+) -> dict[str, Any]:
+    user = get_authenticated_user(base_url, token)
+    login = user.get("login") or user.get("username") or ""
+    payload = {
+        "name": repo,
+        "description": description,
+        "private": private,
+        "auto_init": False,
+        "default_branch": default_branch,
+    }
+    if owner == login:
+        url = f"{api_root(base_url)}/user/repos"
+    else:
+        encoded_owner = urllib.parse.quote(owner, safe="")
+        url = f"{api_root(base_url)}/orgs/{encoded_owner}/repos"
+
+    try:
+        return request_json("POST", url, token, payload)
+    except GiteaHTTPError as exc:
+        if exc.status == 409:
+            existing = get_repo(base_url, token, owner, repo)
+            if existing:
+                return existing
+        raise SystemExit(str(exc)) from exc
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Create or return a Gitea repository.")
+    parser.add_argument("--name", required=True, help="Repository name to create.")
+    parser.add_argument("--owner", help="User or org owner. Defaults to GITEA_OWNER or authenticated user.")
+    parser.add_argument("--description", default="")
+    parser.add_argument("--private", action=argparse.BooleanOptionalAction, default=False)
+    parser.add_argument("--default-branch", default="main")
+    parser.add_argument("--env-file", type=Path, default=DEFAULT_ENV_FILE)
+    args = parser.parse_args()
+
+    cfg = auth_config(args.env_file.expanduser())
+    base_url = cfg["GITEA_BASE_URL"]
+    token = cfg["GITEA_TOKEN"]
+    owner = args.owner or cfg.get("GITEA_OWNER", "")
+    if not owner:
+        user = get_authenticated_user(base_url, token)
+        owner = user.get("login") or user.get("username") or ""
+    if not owner:
+        raise SystemExit("Could not determine Gitea owner. Pass --owner.")
+
+    repo = get_repo(base_url, token, owner, args.name)
+    if repo is None:
+        repo = create_repo(
+            base_url=base_url,
+            token=token,
+            owner=owner,
+            repo=args.name,
+            description=args.description,
+            private=args.private,
+            default_branch=args.default_branch,
+        )
+
+    print(repo.get("html_url") or repo.get("clone_url") or json.dumps(repo, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())