#!/usr/bin/env python3 """Gitea issue workflow helpers for ready-agent issue work.""" from __future__ import annotations import argparse import json import re import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any from gitea_common import DEFAULT_ENV_FILE, config STATUS_LABELS = { "status/needs-triage", "status/needs-info", "status/ready-for-agent", "status/ready-for-human", "status/in-progress", "status/review", "status/done", "status/wontfix", } def api_url(cfg: dict[str, str], suffix: str) -> str: encoded_owner = urllib.parse.quote(cfg["GITEA_OWNER"], safe="") encoded_repo = urllib.parse.quote(cfg["GITEA_REPO"], safe="") return f"{cfg['GITEA_BASE_URL'].rstrip('/')}/api/v1/repos/{encoded_owner}/{encoded_repo}/{suffix.lstrip('/')}" def request_json(method: str, cfg: dict[str, str], suffix: 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(api_url(cfg, suffix), data=data, method=method) request.add_header("Authorization", f"token {cfg['GITEA_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 GiteaError(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 GiteaError(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 labels_by_name(cfg: dict[str, str]) -> dict[str, dict[str, Any]]: labels = request_json("GET", cfg, "labels?limit=1000") return {label["name"]: label for label in labels} def issue_labels(issue: dict[str, Any]) -> set[str]: return {label["name"] for label in issue.get("labels", [])} def read_issue(cfg: dict[str, str], issue_number: int) -> dict[str, Any]: return request_json("GET", cfg, f"issues/{issue_number}") def child_issue_numbers(epic_body: str) -> list[int]: section_match = re.search( r"^## Child Issues.*?(?=^## |\Z)", epic_body, flags=re.MULTILINE | re.DOTALL, ) source = section_match.group(0) if section_match else epic_body seen: set[int] = set() numbers: list[int] = [] for match in re.finditer(r"#(\d+)", source): number = int(match.group(1)) if number not in seen: numbers.append(number) seen.add(number) return numbers def set_status(cfg: dict[str, str], issue_number: int, status: str) -> dict[str, Any]: if status not in STATUS_LABELS: raise SystemExit(f"Unsupported status label: {status}") labels = labels_by_name(cfg) missing = [label for label in [status, *STATUS_LABELS] if label not in labels] if missing: raise SystemExit("Missing Gitea labels: " + ", ".join(sorted(missing))) issue = read_issue(cfg, issue_number) current = issue_labels(issue) if status not in current: request_json("POST", cfg, f"issues/{issue_number}/labels", {"labels": [labels[status]["id"]]}) for label in sorted(current & (STATUS_LABELS - {status})): request_json("DELETE", cfg, f"issues/{issue_number}/labels/{labels[label]['id']}") return read_issue(cfg, issue_number) def set_ref(cfg: dict[str, str], issue_number: int, ref: str) -> dict[str, Any]: return request_json("PATCH", cfg, f"issues/{issue_number}", {"ref": ref}) def print_issue_summary(issue: dict[str, Any]) -> None: labels = ", ".join(sorted(issue_labels(issue))) or "(none)" print(f"#{issue['number']} {issue['title']}") print(f"URL: {issue['html_url']}") print(f"Labels: {labels}") print(f"Ref: {issue.get('ref') or '(none)'}") def cmd_next_ready(args: argparse.Namespace) -> int: cfg = config(args.env_file.expanduser()) epic = read_issue(cfg, args.epic) child_numbers = child_issue_numbers(epic.get("body") or "") for number in child_numbers: issue = read_issue(cfg, number) labels = issue_labels(issue) if issue.get("state") == "open" and "status/ready-for-agent" in labels: if args.claim: issue = set_status(cfg, number, "status/in-progress") print_issue_summary(issue) return 0 print(f"No open child issue with status/ready-for-agent found under epic #{args.epic}.", file=sys.stderr) return 2 def cmd_set_status(args: argparse.Namespace) -> int: cfg = config(args.env_file.expanduser()) issue = set_status(cfg, args.issue, args.status) print_issue_summary(issue) return 0 def cmd_set_ref(args: argparse.Namespace) -> int: cfg = config(args.env_file.expanduser()) issue = set_ref(cfg, args.issue, args.ref) print_issue_summary(issue) return 0 def cmd_comment(args: argparse.Namespace) -> int: cfg = config(args.env_file.expanduser()) body = args.body_file.read_text(encoding="utf-8") if args.body_file else args.body comment = request_json("POST", cfg, f"issues/{args.issue}/comments", {"body": body}) print(comment.get("html_url") or comment.get("url") or json.dumps(comment, indent=2)) return 0 def cmd_create_pr(args: argparse.Namespace) -> int: cfg = config(args.env_file.expanduser()) body = args.body_file.read_text(encoding="utf-8") if args.body_file else args.body payload = {"base": args.base, "head": args.head, "title": args.title, "body": body} try: pr = request_json("POST", cfg, "pulls", payload) except GiteaError as exc: if exc.status != 409: raise SystemExit(str(exc)) from exc owner = cfg["GITEA_OWNER"] pulls = request_json("GET", cfg, f"pulls?state=open&head={owner}:{args.head}&base={args.base}") or [] if not pulls: raise SystemExit(str(exc)) from exc pr = pulls[0] print(pr.get("html_url") or pr.get("url") or json.dumps(pr, indent=2)) return 0 def cmd_yolo_merge(args: argparse.Namespace) -> int: cfg = config(args.env_file.expanduser()) pr = request_json("GET", cfg, f"pulls/{args.pull}") if pr.get("state") != "open": raise SystemExit(f"Pull request #{args.pull} is not open.") if pr.get("merged"): raise SystemExit(f"Pull request #{args.pull} is already merged.") if pr.get("mergeable") is False: raise SystemExit(f"Pull request #{args.pull} is not mergeable.") payload = { "Do": args.method, "delete_branch_after_merge": args.delete_branch, } try: request_json("POST", cfg, f"pulls/{args.pull}/merge", payload) except GiteaError as exc: raise SystemExit(str(exc)) from exc pr_url = pr.get("html_url") or pr.get("url") or f"PR #{args.pull}" comment_body = args.comment or f"YOLO mode merged {pr_url} after automated implementation and verification." request_json("POST", cfg, f"issues/{args.issue}/comments", {"body": comment_body}) issue = set_status(cfg, args.issue, "status/done") print_issue_summary(issue) print(f"Merged PR: {pr_url}") return 0 def main() -> int: parser = argparse.ArgumentParser(description="Work with the configured Gitea issue workflow.") parser.add_argument("--env-file", type=Path, default=DEFAULT_ENV_FILE) subparsers = parser.add_subparsers(dest="command", required=True) next_ready = subparsers.add_parser("next-ready", help="Find the highest-priority ready child issue in an epic.") next_ready.add_argument("--epic", type=int, required=True) next_ready.add_argument("--claim", action="store_true", help="Move the selected issue to status/in-progress.") next_ready.set_defaults(func=cmd_next_ready) set_status_parser = subparsers.add_parser("set-status", help="Replace an issue status label.") set_status_parser.add_argument("--issue", type=int, required=True) set_status_parser.add_argument("--status", required=True) set_status_parser.set_defaults(func=cmd_set_status) set_ref_parser = subparsers.add_parser("set-ref", help="Attach a branch or tag ref to an issue.") set_ref_parser.add_argument("--issue", type=int, required=True) set_ref_parser.add_argument("--ref", required=True) set_ref_parser.set_defaults(func=cmd_set_ref) comment = subparsers.add_parser("comment", help="Add an issue comment.") comment.add_argument("--issue", type=int, required=True) comment.add_argument("--body", default="") comment.add_argument("--body-file", type=Path) comment.set_defaults(func=cmd_comment) create_pr = subparsers.add_parser("create-pr", help="Create or return an existing open pull request.") create_pr.add_argument("--head", required=True) create_pr.add_argument("--base", default="main") create_pr.add_argument("--title", required=True) create_pr.add_argument("--body", default="") create_pr.add_argument("--body-file", type=Path) create_pr.set_defaults(func=cmd_create_pr) yolo_merge = subparsers.add_parser( "yolo-merge", help="Merge a PR immediately, optionally delete its branch, comment, and mark the issue done.", ) yolo_merge.add_argument("--issue", type=int, required=True) yolo_merge.add_argument("--pull", type=int, required=True) yolo_merge.add_argument( "--method", default="merge", choices=("merge", "rebase", "rebase-merge", "squash"), help="Gitea merge method.", ) yolo_merge.add_argument( "--delete-branch", action=argparse.BooleanOptionalAction, default=True, help="Delete the PR branch after merge.", ) yolo_merge.add_argument("--comment", default="") yolo_merge.set_defaults(func=cmd_yolo_merge) args = parser.parse_args() return args.func(args) if __name__ == "__main__": raise SystemExit(main())