import json from .models import normalize_resource # type: ignore[reportMissingImports] JSON_FIELDS = "number,title,body,state,labels,comments,author,assignees,url,createdAt,updatedAt" class Adapter: provider = "" executable = "" def __init__(self, repo=None): self.repo = repo def _repo_args(self): return ["--repo", self.repo] if self.repo else [] def command(self, operation, **kwargs): method = getattr(self, f"command_{operation.replace('.', '_')}") return method(**kwargs) def normalize(self, value, kind): return normalize_resource(value, kind=kind, provider=self.provider) def json_value(self, stdout): text = (stdout or "").strip() if not text: return {} try: return json.loads(text) except (TypeError, ValueError): return {"output": text} class GitHubAdapter(Adapter): provider = "github" executable = "gh" def command_issue_create(self, title, body, labels, assignees): args = ["gh", "issue", "create", "--title", title, "--body", body] for label in labels: args += ["--label", label] for user in assignees: args += ["--assignee", user] return args + self._repo_args() def command_pr_create(self, title, body, head, base, labels, assignees): args = ["gh", "pr", "create", "--title", title, "--body", body] if head: args += ["--head", head] if base: args += ["--base", base] for label in labels: args += ["--label", label] for user in assignees: args += ["--assignee", user] return args + self._repo_args() def _view(self, kind, number, comments=True): command = "pr" if kind == "pr" else "issue" args = ["gh", command, "view", str(number)] if comments: args.append("--comments") return args + ["--json", JSON_FIELDS] + self._repo_args() def command_issue_get(self, number, comments=True): return self._view("issue", number, comments) def command_pr_get(self, number, comments=True): return self._view("pr", number, comments) def command_issue_list(self, state, labels, limit): args = ["gh", "issue", "list", "--state", state, "--limit", str(limit)] for label in labels: args += ["--label", label] return args + ["--json", JSON_FIELDS] + self._repo_args() def command_pr_list(self, state, limit): return ["gh", "pr", "list", "--state", state, "--limit", str(limit), "--json", JSON_FIELDS] + self._repo_args() def _edit(self, kind, number, title=None, body=None, add_label=None, remove_label=None, assignee=None, unassign=None): command = "pr" if kind == "pr" else "issue" args = ["gh", command, "edit", str(number)] if title is not None: args += ["--title", title] if body is not None: args += ["--body", body] if add_label: args += ["--add-label", add_label] if remove_label: args += ["--remove-label", remove_label] if assignee: args += ["--add-assignee", assignee] if unassign: args += ["--remove-assignee", unassign] return args + self._repo_args() def command_issue_edit(self, **kwargs): return self._edit("issue", **kwargs) def command_pr_edit(self, **kwargs): return self._edit("pr", **kwargs) def command_issue_comment(self, number, body): return ["gh", "issue", "comment", str(number), "--body", body] + self._repo_args() def command_pr_comment(self, number, body): return ["gh", "pr", "comment", str(number), "--body", body] + self._repo_args() def command_issue_close(self, number): return ["gh", "issue", "close", str(number)] + self._repo_args() def command_pr_close(self, number): return ["gh", "pr", "close", str(number)] + self._repo_args() def command_pr_diff(self, number): return ["gh", "pr", "diff", str(number)] + self._repo_args() def command_label_list(self): return ["gh", "label", "list", "--limit", "1000", "--json", "name,color,description"] + self._repo_args() def command_label_create(self, name, color, description): args = ["gh", "label", "create", name, "--color", color] if description: args += ["--description", description] return args + self._repo_args() def command_issue_resolve(self, number): return self.command_issue_get(number) class GitLabAdapter(Adapter): provider = "gitlab" executable = "glab" def _format(self): return ["-F", "json"] def _surface(self, kind): return "mr" if kind == "pr" else "issue" def command_issue_create(self, title, body, labels, assignees): args = ["glab", "issue", "create", "--title", title, "--description", body] if labels: args += ["--label", ",".join(labels)] if assignees: args += ["--assignee", ",".join(assignees)] return args + self._repo_args() def command_pr_create(self, title, body, head, base, labels, assignees): args = ["glab", "mr", "create", "--title", title, "--description", body] if head: args += ["--source-branch", head] if base: args += ["--target-branch", base] if labels: args += ["--label", ",".join(labels)] if assignees: args += ["--assignee", ",".join(assignees)] return args + self._repo_args() def command_issue_get(self, number, comments=True): args = ["glab", "issue", "view", str(number)] if comments: args.append("--comments") return args + self._format() + self._repo_args() def command_pr_get(self, number, comments=True): args = ["glab", "mr", "view", str(number)] if comments: args.append("--comments") return args + self._format() + self._repo_args() def command_issue_list(self, state, labels, limit): args = ["glab", "issue", "list", "--state", state, "--per-page", str(limit)] if labels: args += ["--label", ",".join(labels)] return args + self._format() + self._repo_args() def command_pr_list(self, state, limit): return ["glab", "mr", "list", "--state", state, "--per-page", str(limit)] + self._format() + self._repo_args() def _edit(self, kind, number, title=None, body=None, add_label=None, remove_label=None, assignee=None, unassign=None): args = ["glab", self._surface(kind), "update", str(number)] if title is not None: args += ["--title", title] if body is not None: args += ["--description", body] if add_label: args += ["--label", add_label] if remove_label: args += ["--unlabel", remove_label] if assignee: args += ["--assignee", assignee] if unassign: args += ["--unassign", unassign] return args + self._repo_args() def command_issue_edit(self, **kwargs): return self._edit("issue", **kwargs) def command_pr_edit(self, **kwargs): return self._edit("pr", **kwargs) def command_issue_comment(self, number, body): return ["glab", "issue", "note", str(number), "--message", body] + self._repo_args() def command_pr_comment(self, number, body): return ["glab", "mr", "note", str(number), "--message", body] + self._repo_args() def command_issue_close(self, number): return ["glab", "issue", "close", str(number)] + self._repo_args() def command_pr_close(self, number): return ["glab", "mr", "close", str(number)] + self._repo_args() def command_pr_diff(self, number): return ["glab", "mr", "diff", str(number)] + self._repo_args() def command_label_list(self): return ["glab", "label", "list"] + self._format() + self._repo_args() def command_label_create(self, name, color, description): args = ["glab", "label", "create", name, "--color", color] if description: args += ["--description", description] return args + self._repo_args() class GiteaAdapter(Adapter): provider = "gitea" executable = "tea" def _format(self): return ["-o", "json"] def _surface(self, kind): return "pr" if kind == "pr" else "issue" def command_issue_create(self, title, body, labels, assignees): args = ["tea", "issue", "create", "--title", title, "--description", body] if labels: args += ["--labels", ",".join(labels)] if assignees: args += ["--assignees", ",".join(assignees)] return args + self._repo_args() def command_pr_create(self, title, body, head, base, labels, assignees): args = ["tea", "pr", "create", "--title", title, "--description", body] if head: args += ["--head", head] if base: args += ["--base", base] if labels: args += ["--labels", ",".join(labels)] if assignees: args += ["--assignees", ",".join(assignees)] return args + self._repo_args() def _view(self, kind, number, comments=True): args = ["tea", self._surface(kind), str(number)] if comments: args.append("--comments") return args + self._format() + self._repo_args() def command_issue_get(self, number, comments=True): return self._view("issue", number, comments) def command_pr_get(self, number, comments=True): return self._view("pr", number, comments) def command_issue_list(self, state, labels, limit): args = ["tea", "issue", "list", "--state", state, "--limit", str(limit)] if labels: args += ["--labels", ",".join(labels)] return args + self._format() + self._repo_args() def command_pr_list(self, state, limit): return ["tea", "pr", "list", "--state", state, "--limit", str(limit)] + self._format() + self._repo_args() def _edit(self, kind, number, title=None, body=None, add_label=None, remove_label=None, assignee=None, unassign=None): args = ["tea", self._surface(kind), "edit", str(number)] if title is not None: args += ["--title", title] if body is not None: args += ["--description", body] if add_label: args += ["--add-label", add_label] if remove_label: args += ["--remove-label", remove_label] if assignee: args += ["--add-assignee", assignee] if unassign: args += ["--remove-assignee", unassign] return args + self._repo_args() def command_issue_edit(self, **kwargs): return self._edit("issue", **kwargs) def command_pr_edit(self, **kwargs): return self._edit("pr", **kwargs) def command_issue_comment(self, number, body): return ["tea", "comment", str(number), body] + self._repo_args() def command_pr_comment(self, number, body): return ["tea", "comment", str(number), body] + self._repo_args() def command_issue_close(self, number): return ["tea", "issue", "close", str(number)] + self._repo_args() def command_pr_close(self, number): return ["tea", "pr", "close", str(number)] + self._repo_args() def command_pr_diff(self, number): return ["tea", "api", f"/repos/{{owner}}/{{repo}}/pulls/{number}.diff"] + self._repo_args() def command_label_list(self): return ["tea", "label", "list"] + self._format() + self._repo_args() def command_label_create(self, name, color, description): args = ["tea", "label", "create", "--name", name, "--color", color] if description: args += ["--description", description] return args + self._repo_args() ADAPTERS = {"github": GitHubAdapter, "gitlab": GitLabAdapter, "gitea": GiteaAdapter}