266 lines
10 KiB
Python
Executable File
266 lines
10 KiB
Python
Executable File
#!/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())
|