skills/tracker/adapters.py

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}