add gitlab workflows

This commit is contained in:
bobo.yang 2024-11-14 14:58:53 +08:00
parent 41a1de7bbb
commit 2c1a10c4ce
36 changed files with 2240 additions and 0 deletions

View File

@ -0,0 +1,24 @@
### code_task_summary
根据当前分支或指定的Issue,生成代码任务摘要。
#### 用途
- 自动生成简洁的代码任务描述
- 帮助开发者快速理解任务要点
- 用于更新项目配置或文档
#### 使用方法
执行命令: `/github.code_task_summary [issue_url]`
- 如不提供issue_url,将基于当前分支名称提取Issue信息
- 如提供issue_url,将直接使用该Issue的内容
#### 操作流程
1. 获取Issue信息
2. 生成代码任务摘要
3. 允许用户编辑摘要
4. 更新项目配置文件
#### 注意事项
- 确保Git仓库配置正确
- 需要有效的GitHub Token以访问API

View File

@ -0,0 +1,120 @@
import json
import os
import sys
from devchat.llm import chat_json
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit, ui_edit # noqa: E402
from git_api import ( # noqa: E402
check_git_installed,
get_current_branch,
get_gitlab_issue_repo,
get_issue_info,
is_issue_url,
read_issue_by_url,
)
def extract_issue_id(branch_name):
if "#" in branch_name:
return branch_name.split("#")[-1]
return None
# Function to generate a random branch name
PROMPT = (
"You are a coding engineer, required to summarize the ISSUE description into a coding task description of no more than 50 words. \n" # noqa: E501
"The ISSUE description is as follows: {issue_body}, please summarize the corresponding coding task description.\n" # noqa: E501
'The coding task description should be output in JSON format, in the form of: {{"summary": "code task summary"}}\n' # noqa: E501
)
@chat_json(prompt=PROMPT)
def generate_code_task_summary(issue_body):
pass
@ui_edit(ui_type="editor", description="Edit code task summary:")
def edit_code_task_summary(task_summary):
pass
def get_issue_or_task(task):
if is_issue_url(task):
issue = read_issue_by_url(task.strip())
assert_exit(not issue, "Failed to read issue.", exit_code=-1)
return json.dumps({"id": issue["iid"], "title": issue["title"], "description": issue["description"]})
else:
return task
def get_issue_json(issue_id, task):
issue = {"id": "no issue id", "title": "", "description": task}
if issue_id:
issue = get_issue_info(issue_id)
assert_exit(not issue, f"Failed to retrieve issue with ID: {issue_id}", exit_code=-1)
issue = {
"id": issue_id,
"web_url": issue["web_url"],
"title": issue["title"],
"description": issue["description"],
}
return issue
# Main function
def main():
print("Start update code task summary ...", end="\n\n", flush=True)
is_git_installed = check_git_installed()
assert_exit(not is_git_installed, "Git is not installed.", exit_code=-1)
task = sys.argv[1]
repo_name = get_gitlab_issue_repo()
branch_name = get_current_branch()
issue_id = extract_issue_id(branch_name)
# print basic info, repo_name, branch_name, issue_id
print("repo name:", repo_name, end="\n\n")
print("branch name:", branch_name, end="\n\n")
print("issue id:", issue_id, end="\n\n")
issue = get_issue_json(issue_id, task)
assert_exit(not issue["description"], f"Failed to retrieve issue with ID: {issue_id}", exit_code=-1)
# Generate 5 branch names
print("Generating code task summary ...", end="\n\n", flush=True)
code_task_summary = generate_code_task_summary(issue_body=issue["description"])
assert_exit(not code_task_summary, "Failed to generate code task summary.", exit_code=-1)
assert_exit(
not code_task_summary.get("summary", None),
"Failed to generate code task summary, missing summary field in result.",
exit_code=-1,
)
code_task_summary = code_task_summary["summary"]
# Select branch name
code_task_summary = edit_code_task_summary(code_task_summary)
assert_exit(not code_task_summary, "Failed to edit code task summary.", exit_code=-1)
code_task_summary = code_task_summary[0]
# create and checkout branch
print("Updating code task summary to config:")
config_file = os.path.join(".chat", "complete.config")
if os.path.exists(config_file):
with open(config_file, "r") as f:
config = json.load(f)
config["taskDescription"] = code_task_summary
else:
config = {"taskDescription": code_task_summary}
with open(config_file, "w") as f:
json.dump(config, f, indent=4)
print("Code task summary has updated")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new branch based current branch, and checkout new branch.'
input: optional
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"

View File

@ -0,0 +1 @@
description: Root of git commands.

View File

@ -0,0 +1,23 @@
### commit
自动生成提交信息并执行Git提交。
#### 用途
- 生成规范的提交信息
- 简化Git提交流程
- 保持提交历史的一致性
#### 使用方法
执行命令: `/github.commit [message]`
- message: 可选的用户输入,用于辅助生成提交信息
#### 操作流程
1. 选择要提交的文件
2. 生成提交信息
3. 允许用户编辑提交信息
4. 执行Git提交
#### 注意事项
- 确保已选择需要提交的文件
- 生成的提交信息可能需要进一步修改以符合项目规范

View File

View File

@ -0,0 +1,6 @@
description: 'Writes a well-formatted commit message for selected code changes and commits them via Git. Include an issue number if desired (e.g., input "/commit to close #12").'
hint: to close Issue #issue_number
input: optional
help: README.md
steps:
- run: $devchat_python $command_path/commit.py "$input" "english"

View File

@ -0,0 +1,427 @@
# flake8: noqa: E402
import os
import re
import subprocess
import sys
from devchat.llm import chat_completion_stream
from lib.chatmark import Checkbox, Form, TextEditor
from lib.ide_service import IDEService
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit # noqa: E402
from git_api import get_issue_info, subprocess_check_output, subprocess_run
diff_too_large_message_en = (
"Commit failed. The modified content is too long "
"and exceeds the model's length limit. "
"You can try to make partial changes to the file and submit multiple times. "
"Making small changes and submitting them multiple times is a better practice."
)
diff_too_large_message_zh = (
"提交失败。修改内容太长,超出模型限制长度,"
"可以尝试选择部分修改文件多次提交,小修改多提交是更好的做法。"
)
COMMIT_PROMPT_LIMIT_SIZE = 20000
def extract_markdown_block(text):
"""
Extracts the first Markdown code block from the given text without the language specifier.
:param text: A string containing Markdown text
:return: The content of the first Markdown code block, or None if not found
"""
# 正则表达式匹配Markdown代码块忽略可选的语言类型标记
pattern = r"```(?:\w+)?\s*\n(.*?)\n```"
match = re.search(pattern, text, re.DOTALL)
if match:
# 返回第一个匹配的代码块内容,去除首尾的反引号和语言类型标记
# 去除块结束标记前的一个换行符,但保留其他内容
block_content = match.group(1)
return block_content
else:
return text
# Read the prompt from the diffCommitMessagePrompt.txt file
def read_prompt_from_file(filename):
"""
Reads the content of a file and returns it as a string.
This function is designed to read a prompt message from a text file.
It expects the file to be encoded in UTF-8 and will strip any leading
or trailing whitespace from the content of the file. If the file does
not exist or an error occurs during reading, the function logs an error
message and exits the script.
Parameters:
- filename (str): The path to the file that contains the prompt message.
Returns:
- str: The content of the file as a string.
Raises:
- FileNotFoundError: If the file does not exist.
- Exception: If any other error occurs during file reading.
"""
try:
with open(filename, "r", encoding="utf-8") as file:
return file.read().strip()
except FileNotFoundError:
IDEService().ide_logging(
"error",
f"File {filename} not found. "
"Please make sure it exists in the same directory as the script.",
)
sys.exit(1)
except Exception as e:
IDEService().ide_logging(
"error", f"An error occurred while reading the file {filename}: {e}"
)
sys.exit(1)
# Read the prompt content from the file
script_path = os.path.dirname(__file__)
PROMPT_FILENAME = os.path.join(script_path, "diffCommitMessagePrompt.txt")
PROMPT_COMMIT_MESSAGE_BY_DIFF_USER_INPUT = read_prompt_from_file(PROMPT_FILENAME)
prompt_commit_message_by_diff_user_input_llm_config = {
"model": os.environ.get("LLM_MODEL", "gpt-3.5-turbo-1106")
}
language = ""
def assert_value(value, message):
"""
判断给定的value是否为True如果是则输出指定的message并终止程序
Args:
value: 用于判断的值
message: 如果value为True时需要输出的信息
Returns:
无返回值
"""
if value:
print(message, file=sys.stderr, flush=True)
sys.exit(-1)
def decode_path(encoded_path):
octal_pattern = re.compile(r"\\[0-7]{3}")
if octal_pattern.search(encoded_path):
bytes_path = encoded_path.encode("utf-8").decode("unicode_escape").encode("latin1")
decoded_path = bytes_path.decode("utf-8")
return decoded_path
else:
return encoded_path
def get_modified_files():
"""
获取当前修改文件列表以及已经staged的文件列表
Args:
Returns:
tuple: 包含两个list的元组第一个list包含当前修改过的文件第二个list包含已经staged的文件
"""
""" 获取当前修改文件列表以及已经staged的文件列表"""
output = subprocess_check_output(["git", "status", "-s", "-u"], text=True, encoding="utf-8")
lines = output.split("\n")
modified_files = []
staged_files = []
def strip_file_name(file_name):
file = file_name.strip()
if file.startswith('"'):
file = file[1:-1]
return file
for line in lines:
if len(line) > 2:
status, filename = line[:2], decode_path(line[3:])
# check wether filename is a directory
if os.path.isdir(filename):
continue
modified_files.append(os.path.normpath(strip_file_name(filename)))
if status == "M " or status == "A " or status == "D ":
staged_files.append(os.path.normpath(strip_file_name(filename)))
return modified_files, staged_files
def get_marked_files(modified_files, staged_files):
"""
根据给定的参数获取用户选中以供提交的文件
Args:
modified_files (List[str]): 用户已修改文件列表
staged_files (List[str]): 用户已staged文件列表
Returns:
List[str]: 用户选中的文件列表
"""
# Create two Checkbox instances for staged and unstaged files
staged_checkbox = Checkbox(staged_files, [True] * len(staged_files))
unstaged_files = [file for file in modified_files if file not in staged_files]
unstaged_checkbox = Checkbox(unstaged_files, [False] * len(unstaged_files))
# Create a Form with both Checkbox instances
form_list = []
if len(staged_files) > 0:
form_list.append("Staged:\n\n")
form_list.append(staged_checkbox)
if len(unstaged_files) > 0:
form_list.append("Unstaged:\n\n")
form_list.append(unstaged_checkbox)
form = Form(form_list, submit_button_name="Continue")
# Render the Form and get user input
form.render()
# Retrieve the selected files from both Checkbox instances
staged_checkbox_selections = staged_checkbox.selections if staged_checkbox.selections else []
unstaged_selections = unstaged_checkbox.selections if unstaged_checkbox.selections else []
selected_staged_files = [staged_files[idx] for idx in staged_checkbox_selections]
selected_unstaged_files = [unstaged_files[idx] for idx in unstaged_selections]
# Combine the selections from both checkboxes
selected_files = selected_staged_files + selected_unstaged_files
return selected_files
def rebuild_stage_list(user_files):
"""
根据用户选中文件重新构建stage列表
Args:
user_files: 用户选中的文件列表
Returns:
None
"""
# Unstage all files
subprocess_check_output(["git", "reset"])
# Stage all user_files
for file in user_files:
subprocess_run(["git", "add", file])
def get_diff():
"""
获取暂存区文件的Diff信息
Args:
Returns:
bytes: 返回bytes类型是git diff --cached命令的输出结果
"""
return subprocess_check_output(["git", "diff", "--cached"])
def get_current_branch():
try:
# 使用git命令获取当前分支名称
result = subprocess_check_output(
["git", "branch", "--show-current"], stderr=subprocess.STDOUT
).strip()
# 将结果从bytes转换为str
current_branch = result.decode("utf-8")
return current_branch
except subprocess.CalledProcessError:
# 如果发生错误,打印错误信息
return None
except FileNotFoundError:
# 如果未找到git命令可能是没有安装git或者不在PATH中
return None
def generate_commit_message_base_diff(user_input, diff, issue):
"""
根据diff信息通过AI生成一个commit消息
Args:
user_input (str): 用户输入的commit信息
diff (str): 提交的diff信息
Returns:
str: 生成的commit消息
"""
global language
language_prompt = "You must response commit message in chinese。\n" if language == "zh" else ""
prompt = (
PROMPT_COMMIT_MESSAGE_BY_DIFF_USER_INPUT.replace("{__DIFF__}", f"{diff}")
.replace("{__USER_INPUT__}", f"{user_input + language_prompt}")
.replace("{__ISSUE__}", f"{issue}")
)
model_token_limit_error = (
diff_too_large_message_en if language == "en" else diff_too_large_message_zh
)
if len(str(prompt)) > COMMIT_PROMPT_LIMIT_SIZE:
print(model_token_limit_error, flush=True)
sys.exit(0)
messages = [{"role": "user", "content": prompt}]
response = chat_completion_stream(messages, prompt_commit_message_by_diff_user_input_llm_config)
if (
not response["content"]
and response.get("error", None)
and f'{response["error"]}'.find("This model's maximum context length is") > 0
):
print(model_token_limit_error)
sys.exit(0)
assert_value(not response["content"], response.get("error", ""))
response["content"] = extract_markdown_block(response["content"])
return response
def display_commit_message_and_commit(commit_message):
"""
展示提交信息并提交
Args:
commit_message: 提交信息
Returns:
None
"""
text_editor = TextEditor(commit_message, submit_button_name="Commit")
text_editor.render()
new_commit_message = text_editor.new_text
if not new_commit_message:
return None
return subprocess_check_output(["git", "commit", "-m", new_commit_message])
def extract_issue_id(branch_name):
if "#" in branch_name:
return branch_name.split("#")[-1]
return None
def get_issue_json(issue_id):
issue = {"id": "no issue id", "title": "", "description": ""}
if issue_id:
issue = get_issue_info(issue_id)
assert_exit(not issue, f"Failed to retrieve issue with ID: {issue_id}", exit_code=-1)
issue = {
"id": issue_id,
"web_url": issue["web_url"],
"title": issue["title"],
"description": issue["description"],
}
return issue
def check_git_installed():
try:
subprocess.run(
["git", "--version"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True,
)
return True
except subprocess.CalledProcessError:
print("Git is not installed on your system.", file=sys.stderr, flush=True)
except FileNotFoundError:
print("Git is not installed on your system.", file=sys.stderr, flush=True)
except Exception:
print("Git is not installed on your system.", file=sys.stderr, flush=True)
return False
def main():
global language
try:
print("Let's follow the steps below.\n\n")
# Ensure enough command line arguments are provided
if len(sys.argv) < 3:
print("Usage: python script.py <user_input> <language>", file=sys.stderr, flush=True)
sys.exit(-1)
user_input = sys.argv[1]
language = sys.argv[2]
if not check_git_installed():
sys.exit(-1)
print(
"Step 1/2: Select the files you've changed that you wish to include in this commit, "
"then click 'Submit'.",
end="\n\n",
flush=True,
)
modified_files, staged_files = get_modified_files()
if len(modified_files) == 0:
print("No files to commit.", file=sys.stderr, flush=True)
sys.exit(-1)
selected_files = get_marked_files(modified_files, staged_files)
if not selected_files:
print("No files selected, commit aborted.")
return
rebuild_stage_list(selected_files)
print(
"Step 2/2: Review the commit message I've drafted for you. "
"Edit it below if needed. Then click 'Commit' to proceed with "
"the commit using this message.",
end="\n\n",
flush=True,
)
diff = get_diff()
branch_name = get_current_branch()
issue_id = extract_issue_id(branch_name)
issue = str(get_issue_json(issue_id))
if branch_name:
user_input += "\ncurrent repo branch name is:" + branch_name
commit_message = generate_commit_message_base_diff(user_input, diff, issue)
# TODO
# remove Closes #IssueNumber in commit message
commit_message["content"] = (
commit_message["content"]
.replace("Closes #IssueNumber", "")
.replace("No specific issue to close", "")
.replace("No specific issue mentioned.", "")
)
commit_result = display_commit_message_and_commit(commit_message["content"])
if not commit_result:
print("Commit aborted.", flush=True)
else:
print("Commit completed.", flush=True)
sys.exit(0)
except Exception as err:
print("Exception:", err, file=sys.stderr, flush=True)
sys.exit(-1)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,39 @@
Objective:** Generate a commit message that succinctly describes the codebase changes reflected in the provided diff, while incorporating any extra context or guidance from the user.
**Commit Message Structure:**
1. **Title Line:** Choose a type such as `feat`, `fix`, `docs`, `style`, `refactor`, `perf`, `test`, `build`, `ci`, `chore`, and so on, and couple it with a succinct title. Use the format: `type: Title`. Only one title line is permissible.
2. **Summary:** Summarize all adjustments concisely within a maximum of three detailed message lines. Prefix each line with a \"-\".
3. **Closing Reference (Conditional):** Include the line `Closes #IssueNumber` only if a specific, relevant issue number has been mentioned in the user input.
**Response Format:**
```
type: Title
- Detail message line 1
- Detail message line 2
- Detail message line 3
Closes #IssueNumber
```
Only append the \"Closes #IssueNumber\" if the user input explicitly references an issue to close.
**Constraints:**
- Exclude markdown code block indicators (```) and the placeholder \"commit_message\" from your response.
- Follow commit message best practices:
- Limit the title length to 50 characters.
- Limit each summary line to 72 characters.
- If the precise issue number is not known or not stated by the user, do not include the closing reference.
**User Input:** `{__USER_INPUT__}`
Determine if `{__USER_INPUT__}` contains a reference to closing an issue. If so, include the closing reference in the commit message. Otherwise, exclude it.
**Code Changes:**
```
{__DIFF__}
```
Related issue:
{__ISSUE__}
Utilize the provided format to craft a commit message that adheres to the stipulated criteria.

View File

@ -0,0 +1,5 @@
description: '为你选定的代码变更生成格式规范的提交信息,并通过 Git 提交。如需要可包含对应 issue 编号(例如,输入“/commit to close #12”'
hint: to close Issue #issue_number
input: optional
steps:
- run: $devchat_python $command_path/../commit.py "$input" "chinese"

View File

@ -0,0 +1,78 @@
import functools
import sys
from lib.chatmark import Checkbox, Form, Radio, TextEditor
def create_ui_objs(ui_decls, args):
ui_objs = []
editors = []
for i, ui in enumerate(ui_decls):
editor = ui[0](args[i])
if ui[1]:
# this is the title of UI object
editors.append(ui[1])
editors.append(editor)
ui_objs.append(editor)
return ui_objs, editors
def edit_form(uis, args):
ui_objs, editors = create_ui_objs(uis, args)
form = Form(editors)
form.render()
values = []
for obj in ui_objs:
if isinstance(obj, TextEditor):
values.append(obj.new_text)
elif isinstance(obj, Radio):
values.append(obj.selection)
else:
# TODO
pass
return values
def editor(description):
def decorator_edit(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
uis = wrapper.uis[::-1]
return edit_form(uis, args)
if hasattr(func, "uis"):
wrapper.uis = func.uis
else:
wrapper.uis = []
wrapper.uis.append((TextEditor, description))
return wrapper
return decorator_edit
def ui_edit(ui_type, description):
def decorator_edit(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
uis = wrapper.uis[::-1]
return edit_form(uis, args)
if hasattr(func, "uis"):
wrapper.uis = func.uis
else:
wrapper.uis = []
ui_type_class = {"editor": TextEditor, "radio": Radio, "checkbox": Checkbox}[ui_type]
wrapper.uis.append((ui_type_class, description))
return wrapper
return decorator_edit
def assert_exit(condition, message, exit_code=-1):
if condition:
if exit_code == 0:
print(message, end="\n\n", flush=True)
else:
print(message, end="\n\n", file=sys.stderr, flush=True)
sys.exit(exit_code)

View File

@ -0,0 +1,19 @@
### config
配置GitHub工作流所需的设置。
#### 用途
- 设置Issue仓库URL
- 配置GitHub Token
#### 使用方法
执行命令: `/github.config`
#### 操作流程
1. 输入Issue仓库URL(可选)
2. 输入GitHub Token
3. 保存配置信息
#### 注意事项
- GitHub Token应妥善保管,不要泄露
- 配置信息将保存在本地文件中

View File

@ -0,0 +1,88 @@
import json
import os
import sys
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import editor # noqa: E402
def read_issue_url():
config_path = os.path.join(os.getcwd(), ".chat", ".workflow_config.json")
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
if "git_issue_repo" in config_data:
return config_data["git_issue_repo"]
return ""
def save_issue_url(issue_url):
config_path = os.path.join(os.getcwd(), ".chat", ".workflow_config.json")
# make dirs
os.makedirs(os.path.dirname(config_path), exist_ok=True)
config_data = {}
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
config_data["git_issue_repo"] = issue_url
with open(config_path, "w+", encoding="utf-8") as f:
json.dump(config_data, f, indent=4)
def read_gitlab_token():
config_path = os.path.join(os.path.expanduser("~/.chat"), ".workflow_config.json")
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
if "gitlab_token" in config_data:
return config_data["gitlab_token"]
return ""
def save_gitlab_token(github_token):
config_path = os.path.join(os.path.expanduser("~/.chat"), ".workflow_config.json")
config_data = {}
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
config_data["gitlab_token"] = github_token
with open(config_path, "w+", encoding="utf-8") as f:
json.dump(config_data, f, indent=4)
@editor(
"Please specify the issue's repository, "
"If the issue is within this repository, no need to specify. "
"Otherwise, format as: username/repository-name"
)
@editor("Input your github TOKEN to access github api:")
def edit_issue(issue_url, github_token):
pass
def main():
print("start config git settings ...", end="\n\n", flush=True)
issue_url = read_issue_url()
github_token = read_gitlab_token()
issue_url, github_token = edit_issue(issue_url, github_token)
if issue_url:
save_issue_url(issue_url)
if github_token:
save_gitlab_token(github_token)
else:
print("Please specify the github token to access github api.")
sys.exit(0)
print("config git settings successfully.")
sys.exit(0)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,4 @@
description: 'Config required settings for GIT workflows.'
help: README.md
steps:
- run: $devchat_python $command_path/command.py

604
merico/gitlab/git_api.py Normal file
View File

@ -0,0 +1,604 @@
import json
import os
import re
import subprocess
import sys
import time
import requests
from lib.chatmark import TextEditor
from lib.ide_service import IDEService
def read_gitlab_token():
config_path = os.path.join(os.path.expanduser("~/.chat"), ".workflow_config.json")
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
if "gitlab_token" in config_data:
return config_data["gitlab_token"]
# ask user to input gitlab token
server_access_token_editor = TextEditor("", "Please input your GitLab access TOKEN to access:")
server_access_token_editor.render()
server_access_token = server_access_token_editor.new_text
if not server_access_token:
print("Please input your GitLab access TOKEN to continue.")
sys.exit(-1)
return server_access_token
current_repo_dir = None
def get_current_repo():
"""
获取当前文件所在的仓库信息
"""
global current_repo_dir
if not current_repo_dir:
selected_data = IDEService().get_selected_range().dict()
current_file = selected_data.get("abspath", None)
if not current_file:
return None
current_dir = os.path.dirname(current_file)
try:
# 获取仓库根目录
current_repo_dir = (
subprocess.check_output(
["git", "rev-parse", "--show-toplevel"],
stderr=subprocess.DEVNULL,
cwd=current_dir,
)
.decode("utf-8")
.strip()
)
except subprocess.CalledProcessError:
# 如果发生错误可能不在git仓库中
return None
return current_repo_dir
def subprocess_check_output(*popenargs, timeout=None, **kwargs):
# 将 current_dir 添加到 kwargs 中的 cwd 参数
current_repo = get_current_repo()
if current_repo:
kwargs["cwd"] = kwargs.get("cwd", current_repo)
# 调用 subprocess.check_output
return subprocess.check_output(*popenargs, timeout=timeout, **kwargs)
def subprocess_run(
*popenargs, input=None, capture_output=False, timeout=None, check=False, **kwargs
):
current_repo = get_current_repo()
if current_repo:
kwargs["cwd"] = kwargs.get("cwd", current_repo)
# 调用 subprocess.run
return subprocess.run(
*popenargs,
input=input,
capture_output=capture_output,
timeout=timeout,
check=check,
**kwargs,
)
def subprocess_call(*popenargs, timeout=None, **kwargs):
current_repo = get_current_repo()
if current_repo:
kwargs["cwd"] = kwargs.get("cwd", current_repo)
# 调用 subprocess.call
return subprocess.call(*popenargs, timeout=timeout, **kwargs)
def subprocess_check_call(*popenargs, timeout=None, **kwargs):
current_repo = get_current_repo()
if current_repo:
kwargs["cwd"] = kwargs.get("cwd", current_repo)
# 调用 subprocess.check_call
return subprocess.check_call(*popenargs, timeout=timeout, **kwargs)
GITLAB_ACCESS_TOKEN = read_gitlab_token()
GITLAB_API_URL = "https://gitlab.com/api/v4"
def create_issue(title, description):
headers = {
"Private-Token": GITLAB_ACCESS_TOKEN,
"Content-Type": "application/json",
}
data = {
"title": title,
"description": description,
}
project_id = get_gitlab_project_id()
issue_api_url = f"{GITLAB_API_URL}/projects/{project_id}/issues"
response = requests.post(issue_api_url, headers=headers, json=data)
if response.status_code == 201:
print("Issue created successfully!")
return response.json()
else:
print(f"Failed to create issue: {response.content}", file=sys.stderr, end="\n\n")
return None
def update_issue_body(issue_iid, issue_body):
headers = {
"Private-Token": GITLAB_ACCESS_TOKEN,
"Content-Type": "application/json",
}
data = {
"description": issue_body,
}
project_id = get_gitlab_project_id()
api_url = f"{GITLAB_API_URL}/projects/{project_id}/issues/{issue_iid}"
response = requests.put(api_url, headers=headers, json=data)
if response.status_code == 200:
print("Issue updated successfully!")
return response.json()
else:
print(f"Failed to update issue: {response.status_code}")
return None
def get_gitlab_project_id():
try:
result = subprocess_check_output(
["git", "remote", "get-url", "origin"], stderr=subprocess.STDOUT
).strip()
repo_url = result.decode("utf-8")
print(f"Original repo URL: {repo_url}", file=sys.stderr)
if repo_url.startswith("git@"):
# Handle SSH URL format
parts = repo_url.split(':')
project_path = parts[1].replace(".git", "")
elif repo_url.startswith("https://"):
# Handle HTTPS URL format
parts = repo_url.split('/')
project_path = '/'.join(parts[3:]).replace(".git", "")
else:
raise ValueError(f"Unsupported Git URL format: {repo_url}")
print(f"Extracted project path: {project_path}", file=sys.stderr)
encoded_project_path = requests.utils.quote(project_path, safe='')
print(f"Encoded project path: {encoded_project_path}", file=sys.stderr)
return encoded_project_path
except subprocess.CalledProcessError as e:
print(f"Error executing git command: {e}", file=sys.stderr)
return None
except Exception as e:
print(f"Error in get_gitlab_project_id: {e}", file=sys.stderr)
return None
# parse sub tasks in issue description
def parse_sub_tasks(description):
sub_tasks = []
lines = description.split("\n")
for line in lines:
if line.startswith("- ["):
sub_tasks.append(line[2:])
return sub_tasks
def update_sub_tasks(description, tasks):
# remove all existing tasks
lines = description.split("\n")
updated_body = "\n".join(line for line in lines if not line.startswith("- ["))
# add new tasks
updated_body += "\n" + "\n".join(f"- {task}" for task in tasks)
return updated_body
def update_task_issue_url(description, task, issue_url):
# task is like:
# [ ] task name
# [x] task name
# replace task name with issue url, like:
# [ ] [task name](url)
# [x] [task name](url)
if task.find("] ") == -1:
return None
task = task[task.find("] ") + 2 :]
return description.replace(task, f"[{task}]({issue_url})")
def check_git_installed():
"""
Check if Git is installed on the local machine.
Tries to execute 'git --version' command to determine the presence of Git.
Returns:
bool: True if Git is installed, False otherwise.
"""
try:
subprocess_run(
["git", "--version"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
return True
except subprocess.CalledProcessError:
print("Git is not installed on your system.")
return False
def create_and_checkout_branch(branch_name):
subprocess_run(["git", "checkout", "-b", branch_name], check=True)
def is_issue_url(task):
task = task.strip()
# 使用正则表达式匹配 http 或 https 开头issues/数字 结尾的 URL
pattern = r'^(http|https)://.*?/issues/\d+$'
is_issue = bool(re.match(pattern, task))
# print(f"Task to check: {task}", file=sys.stderr)
# print(f"Is issue URL: {is_issue}", file=sys.stderr)
return is_issue
def read_issue_by_url(issue_url):
# Extract the issue number and project path from the URL
issue_url = issue_url.replace("/-/", "/")
parts = issue_url.split('/')
issue_number = parts[-1]
project_path = '/'.join(parts[3:-2]) # Assumes URL format: https://gitlab.com/project/path/-/issues/number
# URL encode the project path
encoded_project_path = requests.utils.quote(project_path, safe='')
# Construct the API endpoint URL
api_url = f"{GITLAB_API_URL}/projects/{encoded_project_path}/issues/{issue_number}"
# Send a GET request to the API endpoint
headers = {
"Private-Token": GITLAB_ACCESS_TOKEN,
"Content-Type": "application/json",
}
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
return response.json()
else:
print(f"Error fetching issue: {response.status_code}", file=sys.stderr)
print(f"Response content: {response.text}", file=sys.stderr)
return None
def get_gitlab_issue_repo(issue_repo=False):
try:
config_path = os.path.join(os.getcwd(), ".chat", ".workflow_config.json")
if os.path.exists(config_path) and issue_repo:
with open(config_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
if "git_issue_repo" in config_data:
issue_repo = requests.utils.quote(config_data["git_issue_repo"], safe='')
print(
"current issue repo:",
config_data["git_issue_repo"],
end="\n\n",
file=sys.stderr,
flush=True,
)
return config_data["git_issue_repo"]
return get_gitlab_project_id()
except subprocess.CalledProcessError as e:
print(e)
# 如果发生错误,打印错误信息
return None
except FileNotFoundError:
# 如果未找到git命令可能是没有安装git或者不在PATH中
print("==> File not found...")
return None
# 获取当前分支名称
def get_current_branch():
try:
# 使用git命令获取当前分支名称
result = subprocess_check_output(
["git", "branch", "--show-current"], stderr=subprocess.STDOUT
).strip()
# 将结果从bytes转换为str
current_branch = result.decode("utf-8")
return current_branch
except subprocess.CalledProcessError:
# 如果发生错误,打印错误信息
return None
except FileNotFoundError:
# 如果未找到git命令可能是没有安装git或者不在PATH中
return None
def get_parent_branch():
current_branch = get_current_branch()
if current_branch is None:
return None
try:
# 使用git命令获取当前分支的父分支引用
result = subprocess_check_output(
["git", "rev-parse", "--abbrev-ref", f"{current_branch}@{1}"], stderr=subprocess.STDOUT
).strip()
# 将结果从bytes转换为str
parent_branch_ref = result.decode("utf-8")
if parent_branch_ref == current_branch:
# 如果父分支引用和当前分支相同,说明当前分支可能是基于一个没有父分支的提交创建的
return None
# 使用git命令获取父分支的名称
result = subprocess_check_output(
["git", "name-rev", "--name-only", "--exclude=tags/*", parent_branch_ref],
stderr=subprocess.STDOUT,
).strip()
parent_branch_name = result.decode("utf-8")
return parent_branch_name
except subprocess.CalledProcessError as e:
print(e)
# 如果发生错误,打印错误信息
return None
except FileNotFoundError:
# 如果未找到git命令可能是没有安装git或者不在PATH中
print("==> File not found...")
return None
def get_issue_info(issue_id):
# 获取 GitLab 项目 ID
project_id = get_gitlab_issue_repo()
# 构造 GitLab API 端点 URL
api_url = f"{GITLAB_API_URL}/projects/{project_id}/issues/{issue_id}"
# 发送 GET 请求到 API 端点
headers = {
"Private-Token": GITLAB_ACCESS_TOKEN,
"Content-Type": "application/json",
}
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
return response.json()
else:
print(f"Failed to get issue info. Status code: {response.status_code}", file=sys.stderr)
print(f"Response content: {response.text}", file=sys.stderr)
return None
def get_issue_info_by_url(issue_url):
# get issue id from issue_url
def get_issue_id(issue_url):
# Extract the issue id from the issue_url
issue_id = issue_url.split("/")[-1]
return issue_id
return get_issue_info(get_issue_id(issue_url))
# 获取当前分支自从与base_branch分叉以来的历史提交信息
def get_commit_messages(base_branch):
# 找到当前分支与base_branch的分叉点
merge_base = subprocess_run(
["git", "merge-base", "HEAD", base_branch],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
# 检查是否成功找到分叉点
if merge_base.returncode != 0:
raise RuntimeError(f"Error finding merge base: {merge_base.stderr.strip()}")
# 获取分叉点的提交哈希
merge_base_commit = merge_base.stdout.strip()
# 获取从分叉点到当前分支的所有提交信息
result = subprocess_run(
["git", "log", f"{merge_base_commit}..HEAD"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
# 检查git log命令是否成功执行
if result.returncode != 0:
raise RuntimeError(f"Error retrieving commit messages: {result.stderr.strip()}")
# 返回提交信息列表
return result.stdout
# 创建PR
def create_pull_request(title, description, source_branch, target_branch, project_id):
url = f"{GITLAB_API_URL}/projects/{project_id}/merge_requests"
headers = {"Private-Token": GITLAB_ACCESS_TOKEN, "Content-Type": "application/json"}
payload = {
"title": title,
"description": description,
"source_branch": source_branch,
"target_branch": target_branch
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 201:
response_json = response.json()
return response_json
print(response.text, end="\n\n", file=sys.stderr)
return None
def get_recently_mr(project_id):
project_id = requests.utils.quote(project_id, safe='')
url = f"{GITLAB_API_URL}/projects/{project_id}/merge_requests?state=opened&order_by=updated_at&sort=desc"
headers = {
"Private-Token": GITLAB_ACCESS_TOKEN,
"Content-Type": "application/json",
}
response = requests.get(url, headers=headers)
branch_name = get_current_branch()
if response.status_code == 200:
mrs = response.json()
for mr in mrs:
if mr["source_branch"] == branch_name:
return mr
return None
else:
return None
def run_command_with_retries(command, retries=3, delay=5):
for attempt in range(retries):
try:
subprocess_check_call(command)
return True
except subprocess.CalledProcessError as e:
print(f"Command failed: {e}")
if attempt < retries - 1:
print(f"Retrying... (attempt {attempt + 1}/{retries})")
time.sleep(delay)
else:
print("All retries failed.")
return False
def update_mr(project_id, mr_iid, title, description):
project_id = requests.utils.quote(project_id, safe='')
url = f"{GITLAB_API_URL}/projects/{project_id}/merge_requests/{mr_iid}"
headers = {"Private-Token": GITLAB_ACCESS_TOKEN, "Content-Type": "application/json"}
payload = {"title": title, "description": description}
response = requests.put(url, headers=headers, json=payload)
if response.status_code == 200:
print(f"MR updated successfully: {response.json()['web_url']}")
return response.json()
else:
print("Failed to update MR.")
return None
def check_unpushed_commits():
try:
# 获取当前分支的本地提交和远程提交的差异
result = subprocess_check_output(["git", "cherry", "-v"]).decode("utf-8").strip()
# 如果结果不为空说明存在未push的提交
return bool(result)
except subprocess.CalledProcessError as e:
print(f"Error checking for unpushed commits: {e}")
return True
def auto_push():
# 获取当前分支名
if not check_unpushed_commits():
return True
try:
branch = (
subprocess_check_output(["git", "rev-parse", "--abbrev-ref", "HEAD"])
.strip()
.decode("utf-8")
)
except subprocess.CalledProcessError as e:
print(f"Error getting current branch: {e}")
return False
# 检查当前分支是否有对应的远程分支
remote_branch_exists = subprocess_call(["git", "ls-remote", "--exit-code", "origin", branch])
push_command = ["git", "push", "origin", branch]
if remote_branch_exists == 0:
# 如果存在远程分支则直接push提交
return run_command_with_retries(push_command)
else:
# 如果不存在远程分支则发布并push提交
push_command.append("-u")
return run_command_with_retries(push_command)
def get_recently_pr(repo):
url = f"{GITLAB_API_URL}/repos/{repo}/pulls?state=open&sort=updated"
headers = {
"Authorization": f"token {GITLAB_ACCESS_TOKEN}",
"Accept": "application/vnd.github.v3+json",
}
response = requests.get(url, headers=headers)
branch_name = get_current_branch()
if response.status_code == 200:
prs = response.json()
for pr in prs:
if pr["head"]["ref"] == branch_name:
return pr
return None
else:
return None
def update_pr(pr_number, title, description, repo_name):
url = f"{GITLAB_API_URL}/repos/{repo_name}/pulls/{pr_number}"
headers = {"Authorization": f"token {GITLAB_ACCESS_TOKEN}", "Content-Type": "application/json"}
payload = {"title": title, "description": description}
response = requests.patch(url, headers=headers, data=json.dumps(payload))
if response.status_code == 200:
print(f"PR updated successfully: {response.json()['web_url']}")
return response.json()
else:
print("Failed to update PR.")
return None
def get_last_base_branch(default_branch):
"""read last base branch from config file"""
def read_config_item(config_path, item):
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
return config.get(item)
return None
project_config_path = os.path.join(os.getcwd(), ".chat", ".workflow_config.json")
last_base_branch = read_config_item(project_config_path, "last_base_branch")
if last_base_branch:
return last_base_branch
return default_branch
def save_last_base_branch(base_branch=None):
"""save last base branch to config file"""
def save_config_item(config_path, item, value):
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
else:
config = {}
config[item] = value
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=4)
if not base_branch:
base_branch = get_current_branch()
project_config_path = os.path.join(os.getcwd(), ".chat", ".workflow_config.json")
save_config_item(project_config_path, "last_base_branch", base_branch)

View File

@ -0,0 +1,19 @@
### list_issue_tasks
列出指定Issue中的任务列表。
#### 用途
- 查看Issue中的子任务
- 跟踪任务进度
#### 使用方法
执行命令: `/github.list_issue_tasks <issue_url>`
#### 操作流程
1. 获取指定Issue的信息
2. 解析Issue内容中的任务列表
3. 显示任务列表
#### 注意事项
- 需要提供有效的Issue URL
- 任务应以特定格式在Issue中列出(如: - [ ] 任务描述)

View File

@ -0,0 +1,53 @@
import os
import sys
from devchat.llm import chat_json
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit, editor # noqa: E402
from git_api import create_issue # noqa: E402
# Function to generate issue title and description using LLM
PROMPT = (
"Based on the following description, "
"suggest a title and a detailed description for a GitHub issue:\n\n"
"Description: {description}\n\n"
'Output format: {{"title": "<title>", "description": "<description>"}} '
)
@chat_json(prompt=PROMPT)
def generate_issue_content(description):
pass
@editor("Edit issue title:")
@editor("Edit issue description:")
def edit_issue(title, description):
pass
# Main function
def main():
print("start new_issue ...", end="\n\n", flush=True)
assert_exit(len(sys.argv) < 2, "Missing argument.", exit_code=-1)
description = sys.argv[1]
print("Generating issue content ...", end="\n\n", flush=True)
issue_json_ob = generate_issue_content(description=description)
assert_exit(not issue_json_ob, "Failed to generate issue content.", exit_code=-1)
issue_title, issue_body = edit_issue(issue_json_ob["title"], issue_json_ob["description"])
assert_exit(not issue_title, "Issue creation cancelled.", exit_code=0)
print("New Issue:", issue_title, "description:", issue_body, end="\n\n", flush=True)
print("Creating issue ...", end="\n\n", flush=True)
issue = create_issue(issue_title, issue_body)
assert_exit(not issue, "Failed to create issue.", exit_code=-1)
print("New Issue:", issue["web_url"], end="\n\n", flush=True)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new issue.'
input: required
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"

View File

@ -0,0 +1,21 @@
### new_branch
基于当前分支创建新分支并切换到新分支。
#### 用途
- 快速创建新的功能或修复分支
- 保持工作区隔离
#### 使用方法
执行命令: `/github.new_branch <description>`
- description: 新分支的简短描述或相关Issue URL
#### 操作流程
1. 生成多个分支名建议
2. 用户选择或编辑分支名
3. 创建新分支并切换
#### 注意事项
- 确保当前分支的更改已提交
- 如提供Issue URL,会自动关联Issue编号到分支名

View File

@ -0,0 +1,95 @@
import json
import os
import sys
from devchat.llm import chat_json
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit, ui_edit # noqa: E402
from git_api import ( # noqa: E402
check_git_installed,
create_and_checkout_branch,
is_issue_url,
read_issue_by_url,
save_last_base_branch,
)
# Function to generate a random branch name
PROMPT = (
"Give me 5 different git branch names, "
"mainly hoping to express: {task}, "
"Good branch name should looks like: <type>/<main content>,"
"the final result is output in JSON format, "
'as follows: {{"names":["name1", "name2", .. "name5"]}}\n'
)
@chat_json(prompt=PROMPT)
def generate_branch_name(task):
pass
@ui_edit(ui_type="radio", description="Select a branch name")
def select_branch_name_ui(branch_names):
pass
def select_branch_name(branch_names):
[branch_selection] = select_branch_name_ui(branch_names)
assert_exit(branch_selection is None, "No branch selected.", exit_code=0)
return branch_names[branch_selection]
def get_issue_or_task(task):
if is_issue_url(task):
issue = read_issue_by_url(task.strip())
assert_exit(not issue, "Failed to read issue.", exit_code=-1)
return json.dumps(
{"id": issue["iid"], "title": issue["title"], "description": issue["description"]}
), issue["iid"]
else:
return task, None
# Main function
def main():
print("Start create branch ...", end="\n\n", flush=True)
is_git_installed = check_git_installed()
assert_exit(not is_git_installed, "Git is not installed.", exit_code=-1)
task = sys.argv[1]
assert_exit(
not task,
"You need input something about the new branch, or input a issue url.",
exit_code=-1,
)
# read issue by url
task, issue_id = get_issue_or_task(task)
# Generate 5 branch names
print("Generating branch names ...", end="\n\n", flush=True)
branch_names = generate_branch_name(task=task)
assert_exit(not branch_names, "Failed to generate branch names.", exit_code=-1)
branch_names = branch_names["names"]
for index, branch_name in enumerate(branch_names):
if issue_id:
branch_names[index] = f"{branch_name}-#{issue_id}"
# Select branch name
selected_branch = select_branch_name(branch_names)
# save base branch name
save_last_base_branch()
# create and checkout branch
print(f"Creating and checking out branch: {selected_branch}")
create_and_checkout_branch(selected_branch)
print("Branch has create and checkout")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new branch based current branch, and checkout new branch.'
input: required
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"

View File

@ -0,0 +1,21 @@
### new_issue
创建新的GitHub Issue。
#### 用途
- 快速创建标准格式的Issue
- 记录任务、bug或功能请求
#### 使用方法
执行命令: `/github.new_issue <description>`
- description: Issue的简短描述
#### 操作流程
1. 基于描述生成Issue标题和正文
2. 允许用户编辑Issue内容
3. 创建GitHub Issue
#### 注意事项
- 需要有创建Issue的权限
- 生成的内容可能需要进一步完善

View File

@ -0,0 +1,52 @@
import os
import sys
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit, editor # noqa: E402
from devchat.llm import chat_json # noqa: E402
from git_api import create_issue # noqa: E402
# Function to generate issue title and description using LLM
PROMPT = (
"Based on the following description, "
"suggest a title and a detailed description for a GitHub issue:\n\n"
"Description: {description}\n\n"
'Output as valid JSON format: {{"title": "<title>", "description": "<description> use \\n as new line flag."}} ' # noqa: E501
)
@chat_json(prompt=PROMPT)
def generate_issue_content(description):
pass
@editor("Edit issue title:")
@editor("Edit issue description:")
def edit_issue(title, description):
pass
# Main function
def main():
print("start new_issue ...", end="\n\n", flush=True)
assert_exit(len(sys.argv) < 2, "Missing argument.", exit_code=-1)
description = sys.argv[1]
print("Generating issue content ...", end="\n\n", flush=True)
issue_json_ob = generate_issue_content(description=description)
assert_exit(not issue_json_ob, "Failed to generate issue content.", exit_code=-1)
issue_title, issue_body = edit_issue(issue_json_ob["title"], issue_json_ob["description"])
assert_exit(not issue_title, "Issue creation cancelled.", exit_code=0)
print("New Issue:", issue_title, "description:", issue_body, end="\n\n", flush=True)
print("Creating issue ...", end="\n\n", flush=True)
issue = create_issue(issue_title, issue_body)
assert_exit(not issue, "Failed to create issue.", exit_code=-1)
print("New Issue:", issue["web_url"], end="\n\n", flush=True)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new issue.'
input: required
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"

View File

@ -0,0 +1,94 @@
import os
import sys
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".."))
from common_util import assert_exit, editor, ui_edit # noqa: E402
from devchat.llm import chat_json # noqa: E402
from git_api import ( # noqa: E402
create_issue,
get_issue_info_by_url,
parse_sub_tasks,
update_issue_body,
update_task_issue_url,
)
# Function to generate issue title and description using LLM
PROMPT = (
"Following is parent issue content:\n"
"{issue_content}\n\n"
"Based on the following issue task: {task}"
"suggest a title and a detailed description for a GitHub issue:\n\n"
'Output format: {{"title": "<title>", "description": "<description>"}} '
)
@chat_json(prompt=PROMPT)
def generate_issue_content(issue_content, task):
pass
@editor("Edit issue title:")
@editor("Edit issue description:")
def edit_issue(title, description):
pass
@ui_edit(ui_type="radio", description="Select a task to create issue:")
def select_task(tasks):
pass
def get_issue_json(issue_url):
issue = get_issue_info_by_url(issue_url)
assert_exit(not issue, f"Failed to retrieve issue with ID: {issue_id}", exit_code=-1)
return {
"id": issue["iid"],
"web_url": issue["web_url"],
"title": issue["title"],
"description": issue["description"],
}
# Main function
def main():
print("start new_issue ...", end="\n\n", flush=True)
assert_exit(len(sys.argv) < 2, "Missing argument.", exit_code=-1)
issue_url = sys.argv[1]
old_issue = get_issue_json(issue_url)
assert_exit(not old_issue, "Failed to retrieve issue with: {issue_url}", exit_code=-1)
tasks = parse_sub_tasks(old_issue["get_issue_json"])
assert_exit(not tasks, "No tasks in issue description.")
# select task from tasks
[task] = select_task(tasks)
assert_exit(task is None, "No task selected.")
task = tasks[task]
print("task:", task, end="\n\n", flush=True)
print("Generating issue content ...", end="\n\n", flush=True)
issue_json_ob = generate_issue_content(issue_content=old_issue, task=task)
assert_exit(not issue_json_ob, "Failed to generate issue content.", exit_code=-1)
issue_title, issue_body = edit_issue(issue_json_ob["title"], issue_json_ob["description"])
assert_exit(not issue_title, "Issue creation cancelled.", exit_code=0)
print("New Issue:", issue_title, "description:", issue_body, end="\n\n", flush=True)
print("Creating issue ...", end="\n\n", flush=True)
issue = create_issue(issue_title, issue_body)
assert_exit(not issue, "Failed to create issue.", exit_code=-1)
print("New Issue:", issue["web_url"], end="\n\n", flush=True)
# update issue task with new issue url
new_body = update_task_issue_url(old_issue["description"], task, issue["web_url"])
assert_exit(not new_body, f"{task} parse error.")
new_issue = update_issue_body(issue_url, new_body)
assert_exit(not new_issue, "Failed to update issue description.")
print("Issue tasks updated successfully!", end="\n\n", flush=True)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new issue.'
input: required
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"

View File

@ -0,0 +1,22 @@
### new_pr
创建新的Pull Request。
#### 用途
- 自动生成PR标题和描述
- 简化代码审查流程
#### 使用方法
执行命令: `/github.new_pr [additional_info]`
- additional_info: 可选的附加信息
#### 操作流程
1. 获取当前分支信息和相关Issue
2. 生成PR标题和描述
3. 允许用户编辑PR内容
4. 创建Pull Request
#### 注意事项
- 确保当前分支有未合并的更改
- 需要有创建PR的权限

View File

@ -0,0 +1,120 @@
import json
import os
import sys
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit, ui_edit # noqa: E402
from devchat.llm import chat_json # noqa: E402
from git_api import ( # noqa: E402
auto_push,
create_pull_request,
get_commit_messages,
get_current_branch,
get_gitlab_issue_repo,
get_issue_info,
get_last_base_branch,
save_last_base_branch,
)
# 从分支名称中提取issue id
def extract_issue_id(branch_name):
if "#" in branch_name:
return branch_name.split("#")[-1]
return None
# 使用LLM模型生成PR内容
PROMPT = (
"Create a pull request title and description based on "
"the following issue and commit messages, if there is an "
"issue, close that issue in PR description as <user>/<repo>#issue_id:\n"
"Issue: {issue}\n"
"Commits:\n{commit_messages}\n"
"Other information:\n{user_input}\n\n"
"The response result should format as JSON object as following:\n"
'{{"title": "pr title", "description": "pr description"}}'
)
@chat_json(prompt=PROMPT)
def generate_pr_content_llm(issue, commit_message, user_input):
pass
def generate_pr_content(issue, commit_messages, user_input):
response = generate_pr_content_llm(
issue=json.dumps(issue), commit_messages=commit_messages, user_input=user_input
)
assert_exit(not response, "Failed to generate PR content.", exit_code=-1)
return response.get("title"), response.get("description")
@ui_edit(ui_type="editor", description="Edit PR title:")
@ui_edit(ui_type="editor", description="Edit PR description:")
def edit_pr(title, description):
pass
@ui_edit(ui_type="editor", description="Edit base branch:")
def edit_base_branch(base_branch):
pass
def get_issue_json(issue_id):
issue = {"id": "no issue id", "title": "", "description": ""}
if issue_id:
issue = get_issue_info(issue_id)
assert_exit(not issue, f"Failed to retrieve issue with ID: {issue_id}", exit_code=-1)
issue = {
"id": issue_id,
"web_url": issue["web_url"],
"title": issue["title"],
"description": issue["description"],
}
return issue
# 主函数
def main():
print("start new_pr ...", end="\n\n", flush=True)
base_branch = get_last_base_branch("main")
base_branch = edit_base_branch(base_branch)
if isinstance(base_branch, list) and len(base_branch) > 0:
base_branch = base_branch[0]
save_last_base_branch(base_branch)
repo_name = get_gitlab_issue_repo()
branch_name = get_current_branch()
issue_id = extract_issue_id(branch_name)
# print basic info, repo_name, branch_name, issue_id
print("repo name:", repo_name, end="\n\n")
print("branch name:", branch_name, end="\n\n")
print("issue id:", issue_id, end="\n\n")
issue = get_issue_json(issue_id)
commit_messages = get_commit_messages(base_branch)
print("generating pr title and description ...", end="\n\n", flush=True)
user_input = sys.argv[1]
pr_title, pr_body = generate_pr_content(issue, commit_messages, user_input)
assert_exit(not pr_title, "Failed to generate PR content.", exit_code=-1)
pr_title, pr_body = edit_pr(pr_title, pr_body)
assert_exit(not pr_title, "PR creation cancelled.", exit_code=0)
is_push_success = auto_push()
assert_exit(not is_push_success, "Failed to push changes.", exit_code=-1)
pr = create_pull_request(pr_title, pr_body, branch_name, base_branch, repo_name)
assert_exit(not pr, "Failed to create PR.", exit_code=-1)
print(f"PR created successfully: {pr['web_url']}")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new PR.'
input: optional
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"

View File

@ -0,0 +1,22 @@
### update_issue_tasks
更新指定Issue中的任务列表。
#### 用途
- 添加、修改或删除Issue中的子任务
- 更新任务进度
#### 使用方法
执行命令: `/github.update_issue_tasks`
#### 操作流程
1. 输入Issue URL
2. 显示当前任务列表
3. 用户输入更新建议
4. 生成新的任务列表
5. 允许用户编辑新任务列表
6. 更新Issue内容
#### 注意事项
- 需要有编辑Issue的权限
- 小心不要删除或覆盖重要信息

View File

@ -0,0 +1,101 @@
import os
import sys
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit, editor # noqa: E402
from devchat.llm import chat_json # noqa: E402
from git_api import ( # noqa: E402
get_issue_info_by_url,
parse_sub_tasks,
update_issue_body,
update_sub_tasks,
)
TASKS_PROMPT = (
"Following is my git issue content.\n"
"{issue_data}\n\n"
"Sub task in issue is like:- [ ] task name\n"
"'[ ] task name' will be as sub task content\n\n"
"Following is my idea to update sub tasks:\n"
"{user_input}\n\n"
"Please output all tasks in JSON format as:"
'{{"tasks": ["[ ] task1", "[ ] task2"]}}'
)
@chat_json(prompt=TASKS_PROMPT)
def generate_issue_tasks(issue_data, user_input):
pass
def to_task_str(tasks):
task_str = ""
for task in tasks:
task_str += task + "\n"
return task_str
@editor("Edit issue old tasks:")
@editor("Edit issue new tasks:")
def edit_issue_tasks(old_tasks, new_tasks):
pass
@editor("Input ISSUE url:")
def input_issue_url(url):
pass
@editor("How to update tasks:")
def update_tasks_input(user_input):
pass
def get_issue_json(issue_url):
issue = get_issue_info_by_url(issue_url)
assert_exit(not issue, f"Failed to retrieve issue with ID: {issue_id}", exit_code=-1)
return {
"id": issue["iid"],
"web_url": issue["web_url"],
"title": issue["title"],
"description": issue["description"],
}
# Main function
def main():
print("start issue tasks update ...", end="\n\n", flush=True)
[issue_url] = input_issue_url("")
assert_exit(not issue_url, "No issue url.")
print("issue url:", issue_url, end="\n\n", flush=True)
issue = get_issue_json(issue_url)
old_tasks = parse_sub_tasks(issue["description"])
print(f"```tasks\n{to_task_str(old_tasks)}\n```", end="\n\n", flush=True)
[user_input] = update_tasks_input("")
assert_exit(not user_input, "No user input")
new_tasks = generate_issue_tasks(issue_data=issue, user_input=user_input)
assert_exit(not new_tasks, "No new tasks.")
print("new_tasks:", new_tasks, end="\n\n", flush=True)
assert_exit(not new_tasks.get("tasks", []), "No new tasks.")
print("new tasks:", to_task_str(new_tasks["tasks"]), end="\n\n", flush=True)
new_tasks = new_tasks["tasks"]
[old_tasks, new_tasks] = edit_issue_tasks(to_task_str(old_tasks), to_task_str(new_tasks))
assert_exit(not new_tasks, "No new tasks.")
print("new tasks:", new_tasks, end="\n\n", flush=True)
new_body = update_sub_tasks(issue["description"], new_tasks.split("\n"))
new_issue = update_issue_body(issue_url, new_body)
assert_exit(not new_issue, "Failed to update issue description.")
print("Issue tasks updated successfully!", end="\n\n", flush=True)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new issue.'
input: required
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"

View File

@ -0,0 +1,20 @@
### update_pr
更新现有的Pull Request。
#### 用途
- 更新PR的标题和描述
- 反映最新的代码变更
#### 使用方法
执行命令: `/github.update_pr`
#### 操作流程
1. 获取最近的PR信息
2. 重新生成PR标题和描述
3. 允许用户编辑PR内容
4. 更新Pull Request
#### 注意事项
- 确保有更新PR的权限
- 更新前请确认是否有新的提交需要推送

View File

@ -0,0 +1,122 @@
import json
import os
import sys
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from common_util import assert_exit, ui_edit # noqa: E402
from devchat.llm import ( # noqa: E402
chat_json,
)
from git_api import ( # noqa: E402
auto_push,
get_commit_messages,
get_current_branch,
get_gitlab_issue_repo,
get_issue_info,
get_last_base_branch,
get_recently_pr,
save_last_base_branch,
update_pr,
)
# 从分支名称中提取issue id
def extract_issue_id(branch_name):
if "#" in branch_name:
return branch_name.split("#")[-1]
return None
# 使用LLM模型生成PR内容
PROMPT = (
"Create a pull request title and description based on "
"the following issue and commit messages, if there is an "
"issue, close that issue in PR description as <user>/<repo>#issue_id:\n"
"Issue: {issue}\n"
"Commits:\n{commit_messages}\n"
"The response result should format as JSON object as following:\n"
'{{"title": "pr title", "description": "pr description"}}'
)
@chat_json(prompt=PROMPT)
def generate_pr_content_llm(issue, commit_messages):
pass
def generate_pr_content(issue, commit_messages):
response = generate_pr_content_llm(issue=json.dumps(issue), commit_messages=commit_messages)
assert_exit(not response, "Failed to generate PR content.", exit_code=-1)
return response.get("title"), response.get("description")
@ui_edit(ui_type="editor", description="Edit PR title:")
@ui_edit(ui_type="editor", description="Edit PR description:")
def edit_pr(title, description):
pass
@ui_edit(ui_type="editor", description="Edit base branch:")
def edit_base_branch(base_branch):
pass
def get_issue_json(issue_id):
issue = {"id": "no issue id", "title": "", "description": ""}
if issue_id:
issue = get_issue_info(issue_id)
assert_exit(not issue, f"Failed to retrieve issue with ID: {issue_id}", exit_code=-1)
issue = {
"id": issue_id,
"web_url": issue["web_url"],
"title": issue["title"],
"description": issue["description"],
}
return issue
# 主函数
def main():
print("start update_pr ...", end="\n\n", flush=True)
base_branch = get_last_base_branch("main")
base_branch = edit_base_branch(base_branch)
if isinstance(base_branch, list) and len(base_branch) > 0:
base_branch = base_branch[0]
save_last_base_branch(base_branch)
repo_name = get_gitlab_issue_repo()
branch_name = get_current_branch()
issue_id = extract_issue_id(branch_name)
# print basic info, repo_name, branch_name, issue_id
print("repo name:", repo_name, end="\n\n")
print("branch name:", branch_name, end="\n\n")
print("issue id:", issue_id, end="\n\n")
issue = get_issue_json(issue_id)
commit_messages = get_commit_messages(base_branch)
recent_pr = get_recently_pr(repo_name)
assert_exit(not recent_pr, "Failed to get recent PR.", exit_code=-1)
print("generating pr title and description ...", end="\n\n", flush=True)
pr_title, pr_body = generate_pr_content(issue, commit_messages)
assert_exit(not pr_title, "Failed to generate PR content.", exit_code=-1)
pr_title, pr_body = edit_pr(pr_title, pr_body)
assert_exit(not pr_title, "PR creation cancelled.", exit_code=0)
is_push_success = auto_push()
assert_exit(not is_push_success, "Failed to push changes.", exit_code=-1)
pr = update_pr(recent_pr["iid"], pr_title, pr_body, repo_name)
assert_exit(not pr, "Failed to update PR.", exit_code=-1)
print(f"PR updated successfully: {pr['web_url']}")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,5 @@
description: 'Create new PR.'
input: required
help: README.md
steps:
- run: $devchat_python $command_path/command.py "$input"