<commit_before>""" Entry point for tsrc push """

import argparse
import sys
from typing import Optional

import github3
from github3 import GitHub
from github3.pulls import PullRequest
import cli_ui as ui

import tsrc
from tsrc.github_client.api_client import login as github_login
from tsrc.cli.push import RepositoryInfo


class PushAction(tsrc.cli.push.PushAction):
    def __init__(
        self,
        repository_info: RepositoryInfo,
        args: argparse.Namespace,
        github_api: Optional[GitHub] = None,
    ) -> None:
        super().__init__(repository_info, args)
        self.github_api = github_api
        self.repository = None
        self.pull_request = None

    def setup_service(self) -> None:
        if not self.github_api:
            self.github_api = github_login()
        assert self.project_name
        owner, name = self.project_name.split("/")
        self.repository = self.github_api.repository(owner, name)

    def post_push(self) -> None:
        self.pull_request = self.ensure_pull_request()
        assert self.pull_request
        if self.args.close:
            ui.info_2("Closing merge request #%s" % self.pull_request.number)
            self.pull_request.close()
            return
        params = {}
        if self.requested_target_branch:
            params["base"] = self.requested_target_branch

        if self.requested_title:
            params["title"] = self.requested_title

        self.pull_request.update(**params)

        if self.requested_reviewers:
            message = ["Requesting review from", ", ".join(self.requested_reviewers)]
            ui.info_2(*message)
            tsrc.github.request_reviewers(
                self.repository, self.pull_request.number, self.requested_reviewers
            )

        if self.requested_assignee:
            ui.info_2("Assigning to", self.requested_assignee)
            self.assign_pull_request()

        if self.args.merge:
            self.merge_pull_request()

        ui.info(
            ui.green, "::", ui.reset, "See pull request at", self.pull_request.html_url
        )

    def find_opened_pull_request(self) -> Optional[PullRequest]:
        assert self.repository
        for pull_request in self.repository.pull_requests():
            if pull_request.head.ref == self.remote_branch:
                if pull_request.state == "open":
                    return pull_request
        return None

    def create_pull_request(self) -> PullRequest:
        assert self.repository
        ui.info_2("Creating pull request", ui.ellipsis, end="")
        title = self.requested_title or self.remote_branch
        if self.requested_target_branch:
            target_branch = self.requested_target_branch
        else:
            target_branch = self.repository.default_branch
        try:
            pull_request = self.repository.create_pull(
                title, target_branch, self.remote_branch
            )
            ui.info("done", ui.check)
        except github3.GitHubError as github_error:
            ui.info()
            ui.error(ui.red, "\nCould not create pull request")
            for error in github_error.errors:
                ui.info(ui.red, error["message"])
            sys.exit(1)
        return pull_request

    def merge_pull_request(self) -> None:
        assert self.pull_request
        ui.info_2("Merging #", self.pull_request.number)
        self.pull_request.merge()

    def ensure_pull_request(self) -> PullRequest:
        pull_request = self.find_opened_pull_request()
        if pull_request:
            ui.info_2("Found existing pull request: #%s" % pull_request.number)
            return pull_request
        else:
            return self.create_pull_request()

    def assign_pull_request(self) -> None:
        assert self.pull_request
        issue = self.repository.issue(self.pull_request.number)
        issue.assign(self.requested_assignee)<commit_msg>push_github: use the GitHub client interfaces<commit_after>""" Entry point for tsrc push """

import argparse
import sys
from typing import Optional

import github3
from github3 import GitHub
from github3.pulls import PullRequest
import cli_ui as ui

import tsrc
from tsrc.cli.push import RepositoryInfo
from tsrc.github_client.interface import Client, PullRequest


class PushAction(tsrc.cli.push.PushAction):
    def __init__(
        self, repository_info: RepositoryInfo, args: argparse.Namespace, client: Client
    ) -> None:
        self.repository_info = repository_info
        self.args = args
        self.client = client
        project_name = repository_info.project_name
        owner, name = project_name.split("/")
        self.repository = self.client.get_repository(owner, name)

    def process(self) -> None:
        pull_request = self.ensure_pull_request()
        if self.args.close:
            ui.info_2(
                "Closing pull request", ui.bold, pull_request.get_short_description()
            )
            pull_request.close()
            return

        base = None
        if self.requested_target_branch:
            base = self.requested_target_branch

        title = None
        if self.requested_title:
            title = self.requested_title

        pull_request.update(base=base, title=title)

        if self.requested_reviewers:
            message = ["Requesting review from", ", ".join(self.requested_reviewers)]
            message = ["Requesting review from", ", ".join(requested_reviewers)]
            ui.info_2(*message)
            pull_request.request_reviewers(self.requested_reviewers)

        if self.requested_assignee:
            ui.info_2("Assigning to", self.requested_assignee)
            pull_request.assign(requested_assignee)

        if self.args.merge:
            pull_request.merge()

        ui.info(
            ui.green, "::", ui.reset, "See pull request at", pull_request.get_html_url()
        )

    def find_opened_pull_request(self) -> Optional[PullRequest]:
        remote_branch = self.repository_info.remote_branch
        pull_requests = self.repository.find_pull_requests(
            state="open", head=remote_branch
        )
        if pull_requests:
            return pull_requests[0]
        else:
            return None

    def create_pull_request(self) -> PullRequest:
        ui.info_2("Creating pull request")
        remote_branch = self.repository_info.remote_branch
        title = self.args.title or remote_branch
        head = remote_branch
        requested_target_branch = self.args.target_branch
        if requested_target_branch:
            base = requested_target_branch
        else:
            base = self.repository.get_default_branch()
        pull_request = self.repository.create_pull_request(
            title=title, head=head, base=base
        )
        return pull_request

    def ensure_pull_request(self) -> PullRequest:
        pull_request = self.find_opened_pull_request()
        if pull_request:
            ui.info_2(
                "Found existing pull request",
                ui.bold,
                pull_request.get_short_description(),
            )
            return pull_request
        else:
            return self.create_pull_request()
{'fraction_non_alphanumeric': 0.08403710369652499, 'fraction_numerical': 0.0027689325764917623, 'mean_word_length': 2.291116173120729, 'pattern_counts': {'":': 3, '<': 3, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2452155', 'n_tokens_mistral': 2044, 'n_tokens_neox': 1973, 'n_words': 428}