-
Notifications
You must be signed in to change notification settings - Fork 416
/
create_pr.py
185 lines (164 loc) · 6.15 KB
/
create_pr.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
"""
Creates PR given description.
"""
import os
import openai
from loguru import logger
import modal
from github.Repository import Repository
from sweepai.core.entities import FileChangeRequest, PullRequest
from sweepai.core.sweep_bot import SweepBot
from sweepai.handlers.on_review import review_pr
from sweepai.utils.config import SweepConfig
from sweepai.utils.event_logger import posthog
from sweepai.utils.github_utils import get_github_client
from sweepai.utils.constants import DB_NAME, PREFIX, DEFAULT_CONFIG, SWEEP_CONFIG_BRANCH, SWEEP_LOGIN
github_access_token = os.environ.get("GITHUB_TOKEN")
openai.api_key = os.environ.get("OPENAI_API_KEY")
update_index = modal.Function.lookup(DB_NAME, "update_index")
num_of_snippets_to_query = 10
max_num_of_snippets = 5
def create_pr(
file_change_requests: list[FileChangeRequest],
pull_request: PullRequest,
sweep_bot: SweepBot,
username: str,
installation_id: int,
issue_number: int | None = None
):
# Flow:
# 1. Get relevant files
# 2: Get human message
# 3. Get files to change
# 4. Get file changes
# 5. Create PR
organization, repo_name = sweep_bot.repo.full_name.split("/")
metadata = {
"repo_full_name": sweep_bot.repo.full_name,
"organization": organization,
"repo_name": repo_name,
"repo_description": sweep_bot.repo.description,
"username": username,
"installation_id": installation_id,
"function": "on_ticket",
"mode": PREFIX,
}
posthog.capture(username, "started", properties=metadata)
try:
logger.info("Making PR...")
pull_request.branch_name = sweep_bot.create_branch(pull_request.branch_name)
completed_count, fcr_count = sweep_bot.change_files_in_github(file_change_requests, pull_request.branch_name)
if completed_count == 0 and fcr_count != 0:
logger.info("No changes made")
posthog.capture(
username,
"failed",
properties={
"error": "No changes made",
"reason": "No changes made",
**metadata,
},
)
return {"success": False, "error": "No changes made"}
# Include issue number in PR description
if issue_number:
# If the #issue changes, then change on_ticket (f'Fixes #{issue_number}.\n' in pr.body:)
pr_description = f"{pull_request.content}\n\nFixes #{issue_number}.\n\nTo checkout this PR branch, run the following command in your terminal:\n```zsh\ngit checkout {pull_request.branch_name}\n```"
else:
pr_description = f"{pull_request.content}\n\nTo checkout this PR branch, run the following command in your terminal:\n```zsh\ngit checkout {pull_request.branch_name}\n```"
pr = sweep_bot.repo.create_pull(
title=pull_request.title,
body=pr_description,
head=pull_request.branch_name,
base=SweepConfig.get_branch(sweep_bot.repo),
)
except openai.error.InvalidRequestError as e:
logger.error(e)
posthog.capture(
username,
"failed",
properties={
"error": str(e),
"reason": "Invalid request error / context length",
**metadata,
},
)
raise e
except Exception as e:
logger.error(e)
posthog.capture(
username,
"failed",
properties={
"error": str(e),
"reason": "Unexpected error",
**metadata,
},
)
raise e
posthog.capture(username, "success", properties={**metadata})
logger.info("create_pr success")
return {"success": True, "pull_request": pr}
def safe_delete_sweep_branch(
pr, # Github PullRequest
repo: Repository,
) -> bool:
"""
Safely delete Sweep branch
1. Only edited by Sweep
2. Prefixed by sweep/
"""
pr_commits = pr.get_commits()
pr_commit_authors = set([commit.author.login for commit in pr_commits])
# Check if only Sweep has edited the PR, and sweep/ prefix
if len(pr_commit_authors) == 1 \
and SWEEP_LOGIN in pr_commit_authors \
and pr.head.ref.startswith("sweep/"):
branch = repo.get_git_ref(f"heads/{pr.head.ref}")
# pr.edit(state='closed')
branch.delete()
return True
else:
# Failed to delete branch as it was edited by someone else
return False
def create_config_pr(
sweep_bot: SweepBot,
):
title = "Create `sweep.yaml` Config File"
branch_name = SWEEP_CONFIG_BRANCH
branch_name = sweep_bot.create_branch(branch_name, retry=False)
try:
sweep_bot.repo.create_file(
'sweep.yaml',
'Create sweep.yaml config file',
DEFAULT_CONFIG,
branch=branch_name
)
except Exception as e:
logger.error(e)
# Check if the pull request from this branch to main already exists.
# If it does, then we don't need to create a new one.
pull_requests = sweep_bot.repo.get_pulls(
state="open",
sort="created",
base=SweepConfig.get_branch(sweep_bot.repo),
head=branch_name,
)
for pr in pull_requests:
if pr.title == title:
return pr
pr_description = "Config file allows for customization of Sweep."
pr = sweep_bot.repo.create_pull(
title=title,
body=
"""🎉 Thank you for installing Sweep! We're thrilled to announce the latest update for Sweep, your trusty AI junior developer on GitHub. This PR creates a `sweep.yaml` config file, allowing you to personalize Sweep's performance according to your project requirements.
## What's new?
- **Sweep is now configurable**.
- To configure Sweep, simply edit the `sweep.yaml` file in the root of your repository.
- If you need help, check out the [Sweep Default Config](https://github.com/sweepai/sweep/blob/main/sweep.yaml) or [Join Our Discord](https://discord.com/invite/sweep-ai) for help.
Thank you for using Sweep! 🧹
""",
head=branch_name,
base=SweepConfig.get_branch(sweep_bot.repo),
)
return pr