| 49 | |
| 50 | |
| 51 | class GithubManager(Manager[GithubViewType]): |
| 52 | def __init__( |
| 53 | self, token_manager: TokenManager, data_collector: GitHubDataCollector |
| 54 | ): |
| 55 | self.token_manager = token_manager |
| 56 | self.data_collector = data_collector |
| 57 | self.github_integration = GithubIntegration( |
| 58 | auth=Auth.AppAuth(GITHUB_APP_CLIENT_ID, GITHUB_APP_PRIVATE_KEY) |
| 59 | ) |
| 60 | |
| 61 | self.jinja_env = Environment( |
| 62 | loader=FileSystemLoader(OPENHANDS_RESOLVER_TEMPLATES_DIR + 'github') |
| 63 | ) |
| 64 | |
| 65 | def _confirm_incoming_source_type(self, message: Message): |
| 66 | if message.source != SourceType.GITHUB: |
| 67 | raise ValueError(f'Unexpected message source {message.source}') |
| 68 | |
| 69 | def _get_full_repo_name(self, repo_obj: dict) -> str: |
| 70 | owner = repo_obj['owner']['login'] |
| 71 | repo_name = repo_obj['name'] |
| 72 | |
| 73 | return f'{owner}/{repo_name}' |
| 74 | |
| 75 | def _get_installation_access_token(self, installation_id: int) -> str: |
| 76 | token_data = self.github_integration.get_access_token(installation_id) |
| 77 | return token_data.token |
| 78 | |
| 79 | def _add_reaction( |
| 80 | self, github_view: ResolverViewInterface, reaction: str, installation_token: str |
| 81 | ): |
| 82 | """Add a reaction to the GitHub issue, PR, or comment. |
| 83 | |
| 84 | Args: |
| 85 | github_view: The GitHub view object containing issue/PR/comment info |
| 86 | reaction: The reaction to add (e.g. "eyes", "+1", "-1", "laugh", "confused", "heart", "hooray", "rocket") |
| 87 | installation_token: GitHub installation access token for API access |
| 88 | """ |
| 89 | with Github(auth=Auth.Token(installation_token)) as github_client: |
| 90 | repo = github_client.get_repo(github_view.full_repo_name) |
| 91 | # Add reaction based on view type |
| 92 | if isinstance(github_view, GithubInlinePRComment): |
| 93 | pr = repo.get_pull(github_view.issue_number) |
| 94 | inline_comment = pr.get_review_comment(github_view.comment_id) |
| 95 | inline_comment.create_reaction(reaction) |
| 96 | |
| 97 | elif isinstance(github_view, (GithubIssueComment, GithubPRComment)): |
| 98 | issue = repo.get_issue(github_view.issue_number) |
| 99 | comment = issue.get_comment(github_view.comment_id) |
| 100 | comment.create_reaction(reaction) |
| 101 | else: |
| 102 | issue = repo.get_issue(github_view.issue_number) |
| 103 | issue.create_reaction(reaction) |
| 104 | |
| 105 | def _user_has_write_access_to_repo( |
| 106 | self, installation_id: str, full_repo_name: str, username: str |
| 107 | ) -> bool: |
| 108 | """Check if the user is an owner, collaborator, or member of the repository.""" |
no outgoing calls