xiaxinhang/pr-agent
1
0
Fork 0
mirror of https://github.com/qodo-ai/pr-agent.git synced 2025-07-02 11:50:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

persistent release notes

Browse Source
This commit is contained in:
Hussam.lawen
2024-07-03 16:38:13 +03:00
parent b05e15e9ec
commit 0c3940b6a7
5 changed files with 141 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
pr_agent/tools/pr_code_suggestions.py
View File

@ -17,6 +17,7 @@ from pr_agent.log import get_logger
from pr_agent.servers.help import HelpMessage
from pr_agent.tools.pr_description import insert_br_after_x_chars
import difflib
import re
class PRCodeSuggestions:
@ -132,11 +133,12 @@ class PRCodeSuggestions:
if get_settings().pr_code_suggestions.persistent_comment:
final_update_message = False
self.git_provider.publish_persistent_comment(pr_body,
self.publish_persistent_comment_with_history(pr_body,
initial_header="## PR Code Suggestions ✨",
update_header=True,
name="suggestions",
final_update_message=final_update_message, )
final_update_message = final_update_message,
max_previous_comments = get_settings().pr_code_suggestions.max_history_len)
if self.progress_response:
self.progress_response.delete()
else:
@ -160,6 +162,101 @@ class PRCodeSuggestions:
self.git_provider.publish_comment(f"Failed to generate code suggestions for PR")
except Exception as e:
pass
def publish_persistent_comment_with_history(self, pr_comment: str,
initial_header: str,
update_header: bool = True,
name='review',
final_update_message=True,
max_previous_comments=4):
history_header = f"#### Previous Suggestions\n"
last_commit_num = self.git_provider.get_latest_commit_url().split('/')[-1][:7]
latest_suggestion_header = f"Latest Suggestions up to {last_commit_num}"
latest_commit_html_comment = f"<!-- {last_commit_num} -->"
found_comment = None
if max_previous_comments > 0:
try:
prev_comments = list(self.git_provider.get_issue_comments())
for comment in prev_comments:
if comment.body.startswith(initial_header):
prev_suggestions = comment.body
found_comment = comment
comment_url = self.git_provider.get_comment_url(comment)
if history_header.strip() not in comment.body:
# no history section
# extract everything between <table> and </table> in comment.body including <table> and </table>
table_index = comment.body.find("<table>")
if table_index == -1:
self.git_provider.edit_comment(comment, pr_comment)
continue
# find http link from comment.body[:table_index]
up_to_commit_txt = self.extract_link(comment.body[:table_index])
prev_suggestion_table = comment.body[table_index:comment.body.rfind("</table>") + len("</table>")]
tick = "✅ " if "✅" in prev_suggestion_table else ""
# surround with details tag
prev_suggestion_table = f"<details><summary>{tick}{name.capitalize()}{up_to_commit_txt}</summary>\n<br>{prev_suggestion_table}\n\n</details>"
new_suggestion_table = pr_comment.replace(initial_header, "").strip()
pr_comment_updated = f"{initial_header}\n{latest_commit_html_comment}\n\n"
pr_comment_updated += f"{latest_suggestion_header}\n{new_suggestion_table}\n\n___\n\n"
pr_comment_updated += f"{history_header}{prev_suggestion_table}\n"
else:
# get the text of the previous suggestions until the latest commit
sections = prev_suggestions.split(history_header.strip())
latest_table = sections[0].strip()
prev_suggestion_table = sections[1].replace(history_header, "").strip()
# get text after the latest_suggestion_header in comment.body
table_ind = latest_table.find("<table>")
up_to_commit_txt = self.extract_link(latest_table[:table_ind])
latest_table = latest_table[table_ind:latest_table.rfind("</table>") + len("</table>")]
# enforce max_previous_comments
count = prev_suggestions.count(f"\n<details><summary>{name.capitalize()}")
count += prev_suggestions.count(f"\n<details><summary>✅ {name.capitalize()}")
if count >= max_previous_comments:
# remove the oldest suggestion
prev_suggestion_table = prev_suggestion_table[:prev_suggestion_table.rfind(f"<details><summary>{name.capitalize()} up to commit")]
tick = "✅ " if "✅" in latest_table else ""
# Add to the prev_suggestions section
last_prev_table = f"\n<details><summary>{tick}{name.capitalize()}{up_to_commit_txt}</summary>\n<br>{latest_table}\n\n</details>"
prev_suggestion_table = last_prev_table + "\n" + prev_suggestion_table
new_suggestion_table = pr_comment.replace(initial_header, "").strip()
pr_comment_updated = f"{initial_header}\n"
pr_comment_updated += f"{latest_commit_html_comment}\n\n"
pr_comment_updated += f"{latest_suggestion_header}\n\n{new_suggestion_table}\n\n"
pr_comment_updated += "___\n\n"
pr_comment_updated += f"{history_header}\n"
pr_comment_updated += f"{prev_suggestion_table}\n"
get_logger().info(f"Persistent mode - updating comment {comment_url} to latest {name} message")
self.git_provider.edit_comment(comment, pr_comment_updated)
return
except Exception as e:
get_logger().exception(f"Failed to update persistent review, error: {e}")
pass
body = pr_comment.replace(initial_header, "").strip()
pr_comment = f"{initial_header}\n\n{latest_commit_html_comment}\n\n{body}\n\n"
if found_comment is not None:
self.git_provider.edit_comment(found_comment, pr_comment)
else:
self.git_provider.publish_comment(pr_comment)
def extract_link(self, s):
r = re.compile(r"<!--.*?-->")
match = r.search(s)
up_to_commit_txt = ""
if match:
up_to_commit_txt = f" up to commit {match.group(0)[4:-3].strip()}"
return up_to_commit_txt
async def _prepare_prediction(self, model: str) -> dict:
self.patches_diff = get_pr_diff(self.git_provider,