mitai-jinkendo/scripts/gitea/gitea_api.py

#!/usr/bin/env python3
"""
Minimal Gitea API client. Reads GITEA_* from environment or .env in repo root.
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from gitea_lib import (
    issues_comment,
    issues_create,
    issues_get,
    issues_list_all,
    issues_list_page,
    issues_patch,
    load_dotenv,
    repo_file_content,
    repo_root,
    require_config,
)


def cmd_issues_list(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    if args.all_pages:
        items = issues_list_all(
            base, token, owner, repo, state=args.state, limit=args.limit
        )
    else:
        _, items = issues_list_page(
            base,
            token,
            owner,
            repo,
            state=args.state,
            page=args.page,
            limit=args.limit,
        )
    for it in items:
        num = it.get("number")
        title = it.get("title")
        st = it.get("state")
        print(f"#{num} [{st}] {title}")


def cmd_issues_get(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    status, payload = issues_get(base, token, owner, repo, args.number)
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    if status >= 400:
        sys.exit(1)


def cmd_issues_create(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    body = args.body or ""
    if args.body_file:
        body = Path(args.body_file).read_text(encoding="utf-8")
    status, payload = issues_create(
        base,
        token,
        owner,
        repo,
        title=args.title,
        body=body,
        labels=args.labels or [],
    )
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    if status >= 400:
        sys.exit(1)


def cmd_issues_comment(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    body = args.body or ""
    if getattr(args, "body_file", None):
        body = Path(args.body_file).read_text(encoding="utf-8")
    if not body.strip():
        sys.stderr.write("issues comment: --body oder --body-file mit Inhalt erforderlich\n")
        sys.exit(2)
    status, payload = issues_comment(
        base, token, owner, repo, args.number, body
    )
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    if status >= 400:
        sys.exit(1)


def cmd_issues_close(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    status, payload = issues_patch(
        base, token, owner, repo, args.number, {"state": "closed"}
    )
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    if status >= 400:
        sys.exit(1)


def cmd_issues_reopen(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    status, payload = issues_patch(
        base, token, owner, repo, args.number, {"state": "open"}
    )
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    if status >= 400:
        sys.exit(1)


def cmd_issues_edit(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    fields: dict = {}
    if args.title is not None:
        fields["title"] = args.title.strip()
        if not fields["title"]:
            sys.stderr.write("issues edit: --title darf nicht leer sein\n")
            sys.exit(2)
    body: str | None = None
    if args.body_file:
        body = Path(args.body_file).read_text(encoding="utf-8")
    elif args.body is not None:
        body = args.body
    if body is not None:
        fields["body"] = body
    if not fields:
        sys.stderr.write(
            "issues edit: mindestens eines von --title, --body oder --body-file setzen\n"
        )
        sys.exit(2)
    status, payload = issues_patch(base, token, owner, repo, args.number, fields)
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    if status >= 400:
        sys.exit(1)


def cmd_repo_contents(args: argparse.Namespace, base: str, token: str, owner: str, repo: str) -> None:
    status, payload = repo_file_content(
        base, token, owner, repo, args.path, ref=args.ref or ""
    )
    if status >= 400:
        print(json.dumps(payload, indent=2, ensure_ascii=False))
        sys.exit(1)
    if isinstance(payload, dict) and payload.get("encoding") == "text":
        print(payload.get("content", ""))
    else:
        print(json.dumps(payload, indent=2, ensure_ascii=False))


def main() -> None:
    if hasattr(sys.stdout, "reconfigure"):
        try:
            sys.stdout.reconfigure(encoding="utf-8")
        except Exception:
            pass
    root = repo_root()
    load_dotenv(root)

    parser = argparse.ArgumentParser(description="Gitea API helper")
    sub = parser.add_subparsers(dest="domain", required=True)

    p_issues = sub.add_parser("issues", help="Issues")
    i_sub = p_issues.add_subparsers(dest="issues_cmd", required=True)

    p_il = i_sub.add_parser("list", help="List issues")
    p_il.add_argument("--state", default="open", choices=["open", "closed", "all"])
    p_il.add_argument("--limit", type=int, default=50)
    p_il.add_argument("--page", type=int, default=1)
    p_il.add_argument(
        "--all-pages",
        action="store_true",
        help="Alle Seiten abfragen (Vorsicht bei sehr vielen Issues)",
    )
    p_il.set_defaults(_handler=cmd_issues_list)

    p_ig = i_sub.add_parser("get", help="Get one issue")
    p_ig.add_argument("number", type=int)
    p_ig.set_defaults(_handler=cmd_issues_get)

    p_ic = i_sub.add_parser("create", help="Create issue")
    p_ic.add_argument("--title", required=True)
    p_ic.add_argument("--body", default="")
    p_ic.add_argument("--body-file")
    p_ic.add_argument("--labels", nargs="*", default=[])
    p_ic.set_defaults(_handler=cmd_issues_create)

    p_co = i_sub.add_parser("comment", help="Add comment")
    p_co.add_argument("number", type=int)
    p_co.add_argument("--body", default="")
    p_co.add_argument("--body-file", help="Kommentar aus Datei (UTF-8); überschreibt --body wenn gesetzt")
    p_co.set_defaults(_handler=cmd_issues_comment)

    p_cl = i_sub.add_parser("close", help="Close issue")
    p_cl.add_argument("number", type=int)
    p_cl.set_defaults(_handler=cmd_issues_close)

    p_ro = i_sub.add_parser("reopen", help="Reopen issue")
    p_ro.add_argument("number", type=int)
    p_ro.set_defaults(_handler=cmd_issues_reopen)

    p_ed = i_sub.add_parser(
        "edit",
        help="Issue per PATCH ändern (Titel und/oder Beschreibung; für große Texte --body-file)",
    )
    p_ed.add_argument("number", type=int)
    p_ed.add_argument("--title", default=None, help="Neuer Titel")
    p_ed.add_argument("--body", default=None, help="Neue Beschreibung (Markdown)")
    p_ed.add_argument(
        "--body-file",
        default=None,
        help="Beschreibung aus Datei (UTF-8); überschreibt --body wenn beides gesetzt",
    )
    p_ed.set_defaults(_handler=cmd_issues_edit)

    p_repo = sub.add_parser("repo", help="Repository (API)")
    r_sub = p_repo.add_subparsers(dest="repo_cmd", required=True)

    p_rc = r_sub.add_parser("file", help="Get file or directory metadata/content")
    p_rc.add_argument("path")
    p_rc.add_argument("--ref", default="", help="branch/tag/commit")
    p_rc.set_defaults(_handler=cmd_repo_contents)

    args = parser.parse_args()
    try:
        base, token, owner, reponame = require_config()
    except RuntimeError as e:
        sys.stderr.write(str(e) + "\n")
        sys.exit(1)

    handler = args._handler
    handler(args, base, token, owner, reponame)


if __name__ == "__main__":
    main()