#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import shlex
import subprocess
import tempfile
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from collections import defaultdict
import os
import re
def prompt_for_user():
while True:
try:
user_input = input(" Name or email (case insensitive): ")
except (KeyboardInterrupt, EOFError):
return None
clean_input = user_input.strip().lower()
if clean_input != "":
return clean_input
def update_trailers(body, trailer):
with tempfile.NamedTemporaryFile() as fp:
fp.write(body.encode())
fp.flush()
cmd = f"git interpret-trailers --if-exists replace --trailer '{trailer}' {fp.name} "
p = subprocess.run(shlex.split(cmd), capture_output=True, text=True)
fp.close()
return p.stdout
def append_message_to_pr_body(pr: int , message: str):
try:
pr_url = f"https://github.com/apache/kafka/pull/{pr}"
cmd_get_pr = f"gh pr view {pr_url} --json title,body"
result = subprocess.run(shlex.split(cmd_get_pr), capture_output=True, text=True, check=True)
current_pr_body = json.loads(result.stdout).get("body", {}).strip() + " "
pr_title = json.loads(result.stdout).get("title", {})
updated_pr_body = update_trailers(current_pr_body, message)
except subprocess.CalledProcessError as e:
print("Failed to retrieve PR body:", e.stderr)
return
print(f"""New PR body will be: --- {updated_pr_body}--- """)
choice = input(f'Update the body of "{pr_title}"? [Y/n] ').strip().lower()
if choice not in ['', 'y']:
print("Abort.")
return
try:
cmd_edit_body = f"gh pr edit {pr_url} --body {shlex.quote(updated_pr_body)}"
subprocess.run(shlex.split(cmd_edit_body), check=True)
print("PR body updated successfully!")
except subprocess.CalledProcessError as e:
print("Failed to update PR body:", e.stderr)
if __name__ == "__main__":
print("Utility to help generate 'Reviewers' string for Pull Requests. Use Ctrl+D or Ctrl+C to exit")
command = r"git log | grep 'Reviewers\|Author'"
stream = os.popen(command)
lines = stream.readlines()
all_reviewers = defaultdict(int)
for line in lines:
stripped = line.strip().lstrip("Reviewers: ").lstrip("Author: ")
reviewers = stripped.split(",")
for reviewer in reviewers:
all_reviewers[reviewer.strip()] += 1
parsed_reviewers = []
for item in all_reviewers.items():
patterns = r"(?P<name>.*)\s<(?P<email>.*)>"
m = re.match(patterns, item[0])
if m is not None and len(m.groups()) == 2:
if item[1] > 2:
parsed_reviewers.append((m.group("name"), m.group("email"), item[1]))
selected_reviewers = []
while True:
if selected_reviewers:
print(f"Reviewers so far: {selected_reviewers}")
user_input = prompt_for_user()
if user_input is None:
break
candidates = []
for reviewer, email, count in parsed_reviewers:
if reviewer.lower().startswith(user_input) or email.lower().startswith(user_input):
candidates.append((reviewer, email, count))
if len(candidates) == 10:
break
if not candidates:
continue
print(" Possible matches (in order of most recent):")
for i, candidate in zip(range(10), candidates):
print(f"[{i+1}] {candidate[0]} {candidate[1]} ({candidate[2]})")
try:
selection_input = input(" Make a selection: ")
selected_candidate = candidates[int(selection_input)-1]
selected_reviewers.append(selected_candidate)
except (EOFError, KeyboardInterrupt):
break
except (ValueError, IndexError):
print("Invalid selection")
continue
if selected_reviewers:
reviewer_message = "Reviewers: "
reviewer_message += ", ".join([f"{name} <{email}>" for name, email, _ in selected_reviewers])
print(f" {reviewer_message} ")
try:
pr_number = int(input(" Pull Request (Ctrl+D or Ctrl+C to skip): "))
append_message_to_pr_body(pr_number, reviewer_message)
except (EOFError, KeyboardInterrupt):
exit(0)