157 lines
6.6 KiB
Python
157 lines
6.6 KiB
Python
import json
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
# Mixed into DevCommand (bin/dev), which provides PROJECT_NAME, _project_root
|
|
# and _exec_in_root.
|
|
class GiteaCommand:
|
|
##
|
|
# Checks out a branch named after a Gitea issue URL.
|
|
# If the issue is from this project's repo, the branch name will be:
|
|
# issue-<issue_number>/<issue_title>
|
|
# If the issue is from another repo (presumably upstream), the branch name will be:
|
|
# <issue_owner>-issue-<issue_number>/<issue_title>
|
|
def branch_from(self, *args, **kwargs):
|
|
issue_url = args[0] if args else None
|
|
if issue_url is None:
|
|
raise ValueError("Must provide a Gitea issue URL.")
|
|
|
|
base_url, issue_repo, issue_number = self._parse_issue_url(issue_url)
|
|
issue_title = self._fetch_issue_title(base_url, issue_repo, issue_number)
|
|
branch_name = self._format_branch_name(issue_repo, issue_number, issue_title)
|
|
|
|
exists = subprocess.run(
|
|
["git", "-C", self._project_root(), "rev-parse", "--verify", "--quiet", f"refs/heads/{branch_name}"],
|
|
stdout=subprocess.DEVNULL,
|
|
).returncode == 0
|
|
flag = "" if exists else "-b "
|
|
self._exec_in_root(f"git checkout {flag}{shlex.quote(branch_name)}", replace=True)
|
|
|
|
##
|
|
# Creates an issue on this project's Gitea repo.
|
|
# dev issue <title> [-b|--body <text>] [-l|--label <name>]... [-a|--assignee <user>]... [-o|--open]
|
|
# The body is read from stdin if --body is not given and stdin is piped.
|
|
# Gitea's API cannot add issues to a project (only the web UI can), so the
|
|
# issue URL is printed and --open opens it in the browser to do so.
|
|
def issue(self, *args, **kwargs):
|
|
title, body, labels, assignees, open_in_browser = self._parse_issue_args(args)
|
|
|
|
repo = self._origin_repo()
|
|
if repo is None:
|
|
raise RuntimeError("could not determine the Gitea repo from the origin remote")
|
|
|
|
payload = {"title": title, "body": body}
|
|
if labels:
|
|
label_ids = self._fetch_label_ids(repo)
|
|
unknown = [label for label in labels if label not in label_ids]
|
|
if unknown:
|
|
raise ValueError(f"Unknown label(s): {', '.join(unknown)} (available: {', '.join(sorted(label_ids))})")
|
|
payload["labels"] = [label_ids[label] for label in labels]
|
|
if assignees:
|
|
payload["assignees"] = assignees
|
|
|
|
issue = self._tea_api(f"repos/{repo}/issues", payload)
|
|
print(f"Created issue #{issue['number']}: {issue['title']}")
|
|
print(issue["html_url"])
|
|
print(f"Add it to the project from the issue sidebar: Projects > {self.PROJECT_NAME}")
|
|
|
|
if open_in_browser:
|
|
opener = shutil.which("xdg-open") or shutil.which("open")
|
|
if opener is None:
|
|
raise RuntimeError("xdg-open not found on PATH")
|
|
subprocess.Popen([opener, issue["html_url"]], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
def _parse_issue_url(self, url):
|
|
match = re.match(r"^(https?://[^/]+)/([^/]+)/([^/]+)/issues/(\d+)", url)
|
|
if match is None:
|
|
raise ValueError(f"Not a Gitea issue URL: {url}")
|
|
return match[1], f"{match[2]}/{match[3]}", match[4]
|
|
|
|
def _fetch_issue_title(self, base_url, repo, issue_number):
|
|
if shutil.which("tea") is None:
|
|
raise RuntimeError("tea not found on PATH (install gitea tea and run `tea login add`)")
|
|
|
|
endpoint = f"{base_url}/api/v1/repos/{repo}/issues/{issue_number}"
|
|
print(f"Running: tea api {endpoint}")
|
|
result = subprocess.run(["tea", "api", endpoint], stdout=subprocess.PIPE, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"failed to fetch issue title from {endpoint}")
|
|
return json.loads(result.stdout)["title"]
|
|
|
|
def _parse_issue_args(self, args):
|
|
title = None
|
|
body = None
|
|
labels = []
|
|
assignees = []
|
|
open_in_browser = False
|
|
|
|
args = list(args)
|
|
while args:
|
|
arg = args.pop(0)
|
|
if arg in ("-b", "--body", "-l", "--label", "-a", "--assignee"):
|
|
if not args:
|
|
raise ValueError(f"{arg} requires a value")
|
|
value = args.pop(0)
|
|
if arg in ("-b", "--body"):
|
|
body = value
|
|
elif arg in ("-l", "--label"):
|
|
labels.append(value)
|
|
else:
|
|
assignees.append(value)
|
|
elif arg in ("-o", "--open"):
|
|
open_in_browser = True
|
|
elif arg.startswith("-"):
|
|
raise ValueError(f"Unknown option: {arg}")
|
|
elif title is None:
|
|
title = arg
|
|
else:
|
|
title = f"{title} {arg}"
|
|
|
|
if not title:
|
|
raise ValueError("Must provide an issue title. Usage: dev issue <title> [-b <body>] [-l <label>]... [-a <user>]... [-o]")
|
|
if body is None:
|
|
body = "" if sys.stdin.isatty() else sys.stdin.read()
|
|
return title, body, labels, assignees, open_in_browser
|
|
|
|
def _tea_api(self, endpoint, payload=None):
|
|
if shutil.which("tea") is None:
|
|
raise RuntimeError("tea not found on PATH (install gitea tea and run `tea login add`)")
|
|
|
|
command = ["tea", "api", endpoint]
|
|
if payload is not None:
|
|
command += ["-d", "@-"]
|
|
print(f"Running: {' '.join(command)}")
|
|
result = subprocess.run(
|
|
command,
|
|
input=json.dumps(payload) if payload is not None else None,
|
|
stdout=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"tea api request failed: {endpoint}")
|
|
return json.loads(result.stdout)
|
|
|
|
def _fetch_label_ids(self, repo):
|
|
labels = self._tea_api(f"repos/{repo}/labels?limit=100")
|
|
return {label["name"]: label["id"] for label in labels}
|
|
|
|
def _format_branch_name(self, issue_repo, issue_number, issue_title):
|
|
formatted_title = re.sub(r"-+", "-", re.sub(r"[^a-z0-9\-]", "", re.sub(r"\s+", "-", issue_title.lower().strip())))
|
|
if issue_repo == self._origin_repo():
|
|
return f"issue-{issue_number}/{formatted_title}"
|
|
issue_owner = issue_repo.split("/")[0]
|
|
return f"{issue_owner}-issue-{issue_number}/{formatted_title}"
|
|
|
|
def _origin_repo(self):
|
|
result = subprocess.run(
|
|
["git", "-C", self._project_root(), "remote", "get-url", "origin"],
|
|
stdout=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
match = re.search(r"[:/]([^/:]+/[^/]+?)(?:\.git)?$", result.stdout.strip())
|
|
return match[1] if match else None
|