hermes-agent/devkit/gitea_client.py

#!/usr/bin/env python3
"""
Shared Gitea API client for wizard fleet automation.

Usage as CLI:
    python -m devkit.gitea_client issues --repo Timmy_Foundation/hermes-agent --state open
    python -m devkit.gitea_client issue --repo Timmy_Foundation/hermes-agent --number 142
    python -m devkit.gitea_client create-comment --repo Timmy_Foundation/hermes-agent --number 142 --body "Update from Bezalel"
    python -m devkit.gitea_client prs --repo Timmy_Foundation/hermes-agent --state open

Usage as module:
    from devkit.gitea_client import GiteaClient
    client = GiteaClient()
    issues = client.list_issues("Timmy_Foundation/hermes-agent", state="open")
"""

import argparse
import json
import os
import sys
from typing import Any, Dict, List, Optional

import urllib.request


DEFAULT_BASE_URL = os.getenv("GITEA_URL", "https://forge.alexanderwhitestone.com")
DEFAULT_TOKEN = os.getenv("GITEA_TOKEN", "")


class GiteaClient:
    def __init__(self, base_url: str = DEFAULT_BASE_URL, token: str = DEFAULT_TOKEN):
        self.base_url = base_url.rstrip("/")
        self.token = token or ""

    def _request(
        self,
        method: str,
        path: str,
        data: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
    ) -> Any:
        url = f"{self.base_url}/api/v1{path}"
        req_headers = {"Content-Type": "application/json", "Accept": "application/json"}
        if self.token:
            req_headers["Authorization"] = f"token {self.token}"
        if headers:
            req_headers.update(headers)

        body = json.dumps(data).encode() if data else None
        req = urllib.request.Request(url, data=body, headers=req_headers, method=method)

        try:
            with urllib.request.urlopen(req) as resp:
                return json.loads(resp.read().decode())
        except urllib.error.HTTPError as e:
            return {"error": True, "status": e.code, "body": e.read().decode()}

    def list_issues(self, repo: str, state: str = "open", limit: int = 50) -> List[Dict]:
        return self._request("GET", f"/repos/{repo}/issues?state={state}&limit={limit}") or []

    def get_issue(self, repo: str, number: int) -> Dict:
        return self._request("GET", f"/repos/{repo}/issues/{number}") or {}

    def create_comment(self, repo: str, number: int, body: str) -> Dict:
        return self._request(
            "POST", f"/repos/{repo}/issues/{number}/comments", {"body": body}
        )

    def update_issue(self, repo: str, number: int, **fields) -> Dict:
        return self._request("PATCH", f"/repos/{repo}/issues/{number}", fields)

    def list_prs(self, repo: str, state: str = "open", limit: int = 50) -> List[Dict]:
        return self._request("GET", f"/repos/{repo}/pulls?state={state}&limit={limit}") or []

    def get_pr(self, repo: str, number: int) -> Dict:
        return self._request("GET", f"/repos/{repo}/pulls/{number}") or {}

    def create_pr(self, repo: str, title: str, head: str, base: str, body: str = "") -> Dict:
        return self._request(
            "POST",
            f"/repos/{repo}/pulls",
            {"title": title, "head": head, "base": base, "body": body},
        )


def _fmt_json(obj: Any) -> str:
    return json.dumps(obj, indent=2, ensure_ascii=False)


def main(argv: List[str] = None) -> int:
    argv = argv or sys.argv[1:]
    parser = argparse.ArgumentParser(description="Gitea CLI for wizard fleet")
    parser.add_argument("--repo", default="Timmy_Foundation/hermes-agent", help="Repository full name")
    parser.add_argument("--token", default=DEFAULT_TOKEN, help="Gitea API token")
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="Gitea base URL")
    sub = parser.add_subparsers(dest="cmd")

    p_issues = sub.add_parser("issues", help="List issues")
    p_issues.add_argument("--state", default="open")
    p_issues.add_argument("--limit", type=int, default=50)

    p_issue = sub.add_parser("issue", help="Get single issue")
    p_issue.add_argument("--number", type=int, required=True)

    p_prs = sub.add_parser("prs", help="List PRs")
    p_prs.add_argument("--state", default="open")
    p_prs.add_argument("--limit", type=int, default=50)

    p_pr = sub.add_parser("pr", help="Get single PR")
    p_pr.add_argument("--number", type=int, required=True)

    p_comment = sub.add_parser("create-comment", help="Post comment on issue/PR")
    p_comment.add_argument("--number", type=int, required=True)
    p_comment.add_argument("--body", required=True)

    p_update = sub.add_parser("update-issue", help="Update issue fields")
    p_update.add_argument("--number", type=int, required=True)
    p_update.add_argument("--title", default=None)
    p_update.add_argument("--body", default=None)
    p_update.add_argument("--state", default=None)

    p_create_pr = sub.add_parser("create-pr", help="Create a PR")
    p_create_pr.add_argument("--title", required=True)
    p_create_pr.add_argument("--head", required=True)
    p_create_pr.add_argument("--base", default="main")
    p_create_pr.add_argument("--body", default="")

    args = parser.parse_args(argv)
    client = GiteaClient(base_url=args.base_url, token=args.token)

    if args.cmd == "issues":
        print(_fmt_json(client.list_issues(args.repo, args.state, args.limit)))
    elif args.cmd == "issue":
        print(_fmt_json(client.get_issue(args.repo, args.number)))
    elif args.cmd == "prs":
        print(_fmt_json(client.list_prs(args.repo, args.state, args.limit)))
    elif args.cmd == "pr":
        print(_fmt_json(client.get_pr(args.repo, args.number)))
    elif args.cmd == "create-comment":
        print(_fmt_json(client.create_comment(args.repo, args.number, args.body)))
    elif args.cmd == "update-issue":
        fields = {k: v for k, v in {"title": args.title, "body": args.body, "state": args.state}.items() if v is not None}
        print(_fmt_json(client.update_issue(args.repo, args.number, **fields)))
    elif args.cmd == "create-pr":
        print(_fmt_json(client.create_pr(args.repo, args.title, args.head, args.base, args.body)))
    else:
        parser.print_help()
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())