diff --git a/pr_agent/tools/pr_code_suggestions.py b/pr_agent/tools/pr_code_suggestions.py index 777ffd99..d08c723b 100644 --- a/pr_agent/tools/pr_code_suggestions.py +++ b/pr_agent/tools/pr_code_suggestions.py @@ -35,7 +35,6 @@ class PRCodeSuggestions: get_logger().info(f"Setting max_model_tokens to {MAX_CONTEXT_TOKENS_IMPROVE} for PR improve") get_settings().config.max_model_tokens = MAX_CONTEXT_TOKENS_IMPROVE - # extended mode try: self.is_extended = self._get_is_extended(args or []) @@ -116,7 +115,7 @@ class PRCodeSuggestions: # require self-review if get_settings().pr_code_suggestions.demand_code_suggestions_self_review: - text= get_settings().pr_code_suggestions.code_suggestions_self_review_text + text = get_settings().pr_code_suggestions.code_suggestions_self_review_text pr_body += f"\n\n- [ ] {text}" if get_settings().pr_code_suggestions.approve_pr_on_self_review: pr_body += ' ' @@ -193,8 +192,9 @@ class PRCodeSuggestions: # self-reflect on suggestions if get_settings().pr_code_suggestions.self_reflect_on_suggestions: - model = get_settings().config.model_turbo # use turbo model for self-reflection, since it is an easier task - response_reflect = await self.self_reflect_on_suggestions(data["code_suggestions"], patches_diff, model=model) + model = get_settings().config.model_turbo # use turbo model for self-reflection, since it is an easier task + response_reflect = await self.self_reflect_on_suggestions(data["code_suggestions"], patches_diff, + model=model) if response_reflect: response_reflect_yaml = load_yaml(response_reflect) code_suggestions_feedback = response_reflect_yaml["code_suggestions"] @@ -203,7 +203,7 @@ class PRCodeSuggestions: try: suggestion["score"] = code_suggestions_feedback[i]["suggestion_score"] suggestion["score_why"] = code_suggestions_feedback[i]["why"] - except Exception as e: # + except Exception as e: # get_logger().error(f"Error processing suggestion score {i}", artifact={"suggestion": suggestion, "code_suggestions_feedback": code_suggestions_feedback[i]}) @@ -226,7 +226,7 @@ class PRCodeSuggestions: suggestion['improved_code'] = suggestion['improved_code'][:max_code_suggestion_length] suggestion['improved_code'] += f"\n{suggestion_truncation_message}" get_logger().info(f"Truncated suggestion from {len(suggestion['improved_code'])} " - f"characters to {max_code_suggestion_length} characters") + f"characters to {max_code_suggestion_length} characters") return suggestion def _prepare_pr_code_suggestions(self, predictions: str) -> Dict: @@ -240,17 +240,24 @@ class PRCodeSuggestions: one_sentence_summary_list = [] for i, suggestion in enumerate(data['code_suggestions']): try: - if (not suggestion or 'one_sentence_summary' not in suggestion or - 'label' not in suggestion or 'relevant_file' not in suggestion): - get_logger().debug(f"Skipping suggestion {i + 1}, because it is invalid: {suggestion}") + needed_keys = ['one_sentence_summary', 'label', 'relevant_file', 'relevant_lines_start', 'relevant_lines_end'] + is_valid_keys = True + for key in needed_keys: + if key not in suggestion: + is_valid_keys = False + get_logger().debug(f"Skipping suggestion {i + 1}, because it does not contain '{key}':\n'{suggestion}") + break + if not is_valid_keys: continue if suggestion['one_sentence_summary'] in one_sentence_summary_list: get_logger().debug(f"Skipping suggestion {i + 1}, because it is a duplicate: {suggestion}") continue - if 'const' in suggestion['suggestion_content'] and 'instead' in suggestion['suggestion_content'] and 'let' in suggestion['suggestion_content']: - get_logger().debug(f"Skipping suggestion {i + 1}, because it uses 'const instead let': {suggestion}") + if 'const' in suggestion['suggestion_content'] and 'instead' in suggestion[ + 'suggestion_content'] and 'let' in suggestion['suggestion_content']: + get_logger().debug( + f"Skipping suggestion {i + 1}, because it uses 'const instead let': {suggestion}") continue if ('existing_code' in suggestion) and ('improved_code' in suggestion): @@ -258,7 +265,7 @@ class PRCodeSuggestions: get_logger().debug( f"edited improved suggestion {i + 1}, because equal to existing code: {suggestion['existing_code']}") if get_settings().pr_code_suggestions.commitable_code_suggestions: - suggestion['improved_code'] = "" # we need 'existing_code' to locate the code in the PR + suggestion['improved_code'] = "" # we need 'existing_code' to locate the code in the PR else: suggestion['existing_code'] = "" suggestion = self._truncate_if_needed(suggestion) @@ -279,12 +286,15 @@ class PRCodeSuggestions: if not data['code_suggestions']: get_logger().info('No suggestions found to improve this PR.') if self.progress_response: - return self.git_provider.edit_comment(self.progress_response, body='No suggestions found to improve this PR.') + return self.git_provider.edit_comment(self.progress_response, + body='No suggestions found to improve this PR.') else: return self.git_provider.publish_comment('No suggestions found to improve this PR.') for d in data['code_suggestions']: try: + if get_settings().config.verbosity_level >= 2: + get_logger().info(f"suggestion: {d}") relevant_file = d['relevant_file'].strip() relevant_lines_start = int(d['relevant_lines_start']) # absolute position relevant_lines_end = int(d['relevant_lines_end']) @@ -300,8 +310,8 @@ class PRCodeSuggestions: else: body = f"**Suggestion:** {content} [{label}]\n```suggestion\n" + new_code_snippet + "\n```" code_suggestions.append({'body': body, 'relevant_file': relevant_file, - 'relevant_lines_start': relevant_lines_start, - 'relevant_lines_end': relevant_lines_end}) + 'relevant_lines_start': relevant_lines_start, + 'relevant_lines_end': relevant_lines_end}) except Exception: get_logger().info(f"Could not parse suggestion: {d}") @@ -477,14 +487,15 @@ class PRCodeSuggestions: # sort suggestions_labels by the suggestion with the highest score if get_settings().pr_code_suggestions.self_reflect_on_suggestions: - suggestions_labels = dict(sorted(suggestions_labels.items(), key=lambda x: max([s['score'] for s in x[1]]), reverse=True)) + suggestions_labels = dict( + sorted(suggestions_labels.items(), key=lambda x: max([s['score'] for s in x[1]]), reverse=True)) # sort the suggestions inside each label group by score for label, suggestions in suggestions_labels.items(): suggestions_labels[label] = sorted(suggestions, key=lambda x: x['score'], reverse=True) - + counter_suggestions = 0 for label, suggestions in suggestions_labels.items(): - num_suggestions=len(suggestions) + num_suggestions = len(suggestions) pr_body += f"""{label.capitalize()}\n""" for i, suggestion in enumerate(suggestions): @@ -508,8 +519,8 @@ class PRCodeSuggestions: suggestion_content = insert_br_after_x_chars(suggestion_content, 90) # pr_body += f"
{suggestion_content}" - existing_code = suggestion['existing_code'].rstrip()+"\n" - improved_code = suggestion['improved_code'].rstrip()+"\n" + existing_code = suggestion['existing_code'].rstrip() + "\n" + improved_code = suggestion['improved_code'].rstrip() + "\n" diff = difflib.unified_diff(existing_code.split('\n'), improved_code.split('\n'), n=999) @@ -518,7 +529,7 @@ class PRCodeSuggestions: example_code = "" example_code += f"```diff\n{patch}\n```\n" - if i==0: + if i == 0: pr_body += f"""\n\n""" else: pr_body += f"""\n\n""" @@ -529,13 +540,20 @@ class PRCodeSuggestions: pr_body += f"""\n\n
{suggestion_summary}\n\n___\n\n""" pr_body += f""" **{suggestion_content}** - + [{relevant_file} {range_str}]({code_snippet_link}) -{example_code} +{example_code.rstrip()} """ + if (get_settings().pr_code_suggestions.apply_suggestions_checkbox and + (isinstance(self.git_provider, GithubProvider) or isinstance(self.git_provider, + GitLabProvider))): + # add a checkbox line, to create a committal suggestion from the table suggestion + if '...' not in patch: + pr_body += f"""\n- [ ] **Apply this suggestion** \n\n""" + if get_settings().pr_code_suggestions.self_reflect_on_suggestions: - pr_body +=f"\n\n
Suggestion importance[1-10]: {suggestion['score']}\n\n" + pr_body += f"
Suggestion importance[1-10]: {suggestion['score']}\n\n" pr_body += f"Why: {suggestion['score_why']}\n\n" pr_body += f"
" @@ -546,7 +564,7 @@ class PRCodeSuggestions: pr_body += f"{suggestion['score']}\n\n" pr_body += f"" - + counter_suggestions += 1 # pr_body += "
" # pr_body += """""" @@ -570,14 +588,54 @@ class PRCodeSuggestions: "diff": patches_diff, 'num_code_suggestions': len(suggestion_list)} environment = Environment(undefined=StrictUndefined) - system_prompt_reflect = environment.from_string(get_settings().pr_code_suggestions_reflect_prompt.system).render( + system_prompt_reflect = environment.from_string( + get_settings().pr_code_suggestions_reflect_prompt.system).render( variables) - user_prompt_reflect = environment.from_string(get_settings().pr_code_suggestions_reflect_prompt.user).render(variables) + user_prompt_reflect = environment.from_string( + get_settings().pr_code_suggestions_reflect_prompt.user).render(variables) with get_logger().contextualize(command="self_reflect_on_suggestions"): response_reflect, finish_reason_reflect = await self.ai_handler.chat_completion(model=model, - system=system_prompt_reflect, - user=user_prompt_reflect) + system=system_prompt_reflect, + user=user_prompt_reflect) except Exception as e: get_logger().info(f"Could not reflect on suggestions, error: {e}") return "" - return response_reflect \ No newline at end of file + return response_reflect + + async def handle_apply_suggestion(self): + try: + get_logger().info('Processing "apply" suggestion...') + suggestion_number = get_settings().apply_suggestion + comment_after = get_settings().pr_code_suggestions.get('comment_after', None) + if suggestion_number is None or comment_after is None: + get_logger().error('Invalid suggestion number or comment_after') + return False + suggestions = parse_suggestions_content(comment_after) + if not suggestions: + get_logger().error('Failed to parse suggestions') + return False + suggestion = suggestions[suggestion_number] + if hasattr(self, 'main_language'): + self.git_provider.main_language = self.main_language + relevant_file = suggestion['suggestion_orig_location']['filename'] + relevant_lines_start = int(suggestion['suggestion_orig_location']['start_line']) + relevant_lines_end = int(suggestion['suggestion_orig_location']['end_line']) + content = suggestion['suggestion_summary'] + new_code_snippet = suggestion['new_code_snippet'] + label = suggestion['category'] + score = suggestion['score'] + if new_code_snippet: + new_code_snippet = self.dedent_code(relevant_file, relevant_lines_start, new_code_snippet) + body = f"**Suggestion:** {content} [{label}, importance: {score}]\n```suggestion\n" + new_code_snippet + "\n```" + original_suggestion = suggestion + code_suggestions = [({'original_suggestion': original_suggestion, + 'body': body, 'relevant_file': relevant_file, + 'relevant_lines_start': relevant_lines_start, + 'relevant_lines_end': relevant_lines_end})] + is_successful = self.git_provider.publish_code_suggestions(code_suggestions) + get_settings().set("suggestion_score", score) + get_settings().set("suggestion_label", label) + except Exception as e: + get_logger().info(f"Failed to apply suggestion, error: {e}") + is_successful = False + return is_successful