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 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200
| import subprocess import json import os import requests import sys import io
BASE_URL = os.environ.get("OPENAI_API_BASE", "https://yourbaseurl/v1")
API_KEY = os.environ.get("OPENAI_API_KEY", "sk-ssfgyygvfsdfdgfhjhytgrfOIcQ")
MODEL_NAME = "gpt-4.1-mini"
MAX_TOKENS = 60
TEMPERATURE = 0.3
SYSTEM_PROMPT = """你是一个专业的Git提交信息生成器。 请根据提供的 'git diff --staged' 输出或 'git show HEAD' 输出,生成一个简短、清晰且符合 Conventional Commits 规范的提交信息。 规范格式为: <type>(<scope>): <subject> 例如: feat: add new login button fix(parser): handle malformed input docs: update README with API instructions
- type: feat, fix, docs, style, refactor, perf, test, chore - scope: 可选,指明影响范围 - subject: 动词开头,现在时态,简洁描述。
只输出最终的commit信息本身,不要包含任何解释、前缀如 "Commit message:" 或 markdown 代码块标记。 请使用中文回答。 """
def get_staged_diff(): """获取 git staged 的 diff 内容""" try: result = subprocess.run( ["git", "diff", "--staged", "--patch", "--unified=0"], capture_output=True, check=True ) return result.stdout.decode('utf-8', errors='replace').strip() except subprocess.CalledProcessError as e: stderr_msg = e.stderr.decode('utf-8', errors='replace').strip() if e.stderr else "N/A" print(f"获取 git diff 失败: {e}. Stderr: {stderr_msg}", file=sys.stderr) return None except FileNotFoundError: print("错误: git 命令未找到。请确保 git 已安装并在 PATH 中。", file=sys.stderr) return None
def generate_commit_message(diff_content): """调用 AI API 生成 commit message""" if not diff_content or not diff_content.strip(): return "chore: no meaningful changes detected to generate commit message"
if BASE_URL == "YOUR_OPENAI_COMPATIBLE_BASE_URL/v1": print("错误:请在脚本中或通过环境变量 OPENAI_API_BASE 设置 BASE_URL。", file=sys.stderr) return None if "openai.com" in BASE_URL and (not API_KEY or API_KEY == "YOUR_API_KEY"): print("错误:使用 OpenAI 官方 API 时必须提供 API_KEY。", file=sys.stderr) return None
headers = { "Content-Type": "application/json; charset=utf-8", } if API_KEY and API_KEY != "YOUR_API_KEY": headers["Authorization"] = f"Bearer {API_KEY}"
user_prompt = f"请为以下 git diff/show 内容生成一个 commit message:\n\n```diff\n{diff_content}\n```"
payload = { "model": MODEL_NAME, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_prompt} ], "max_tokens": MAX_TOKENS, "temperature": TEMPERATURE, "stream": False }
api_endpoint = f"{BASE_URL.rstrip('/')}/chat/completions"
try: response = requests.post(api_endpoint, headers=headers, json=payload, timeout=45) response.raise_for_status() response_data = response.json() if "choices" in response_data and len(response_data["choices"]) > 0: message_content = response_data["choices"][0].get("message", {}).get("content", "") message_content = message_content.strip() if message_content.startswith("```") and message_content.endswith("```"): message_content = message_content[message_content.find('\n')+1:-3].strip() if '\n' in message_content else message_content[3:-3].strip()
message_content = message_content.removeprefix("`").removesuffix("`") message_content = message_content.strip('"').strip("'") return message_content.strip() else: error_details = response_data.get("error", {}).get("message", "No error message in response.") print(f"错误:API 响应中没有找到有效的 'choices'。API 返回错误: {error_details} 完整响应: {response.text}", file=sys.stderr) return None
except requests.exceptions.Timeout: print(f"API 请求超时 ({api_endpoint})。", file=sys.stderr) return None except requests.exceptions.RequestException as e: print(f"API 请求失败: {e}", file=sys.stderr) if hasattr(e, 'response') and e.response is not None: print(f"API 响应状态码: {e.response.status_code}, 内容: {e.response.text}", file=sys.stderr) return None except json.JSONDecodeError: print(f"错误:解析API响应失败,不是有效的JSON。响应: {response.text}", file=sys.stderr) return None
if __name__ == "__main__": sys.stdin = io.TextIOWrapper(sys.stdin.buffer, encoding='utf-8', errors='surrogateescape') sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
if sys.platform.startswith('win'): try: import ctypes kernel32 = ctypes.windll.kernel32 except Exception as e: print(f"无法设置Windows控制台编码: {e}", file=sys.stderr)
diff_content_input = "" diff_source_description = ""
if not sys.stdin.isatty(): diff_source_description = "标准输入 (stdin)" input_bytes = sys.stdin.buffer.read() try: diff_content_input = input_bytes.decode('utf-8') except UnicodeDecodeError: print(f"警告: 来自 {diff_source_description} 的输入不是有效的 UTF-8。尝试 GBK 解码...", file=sys.stderr) try: diff_content_input = input_bytes.decode('gbk', errors='replace') except UnicodeDecodeError: print(f"警告: 来自 {diff_source_description} 的输入也不是有效的 GBK。最终使用 UTF-8 并替换无效字符...", file=sys.stderr) diff_content_input = input_bytes.decode('utf-8', errors='replace') else: diff_source_description = "'git diff --staged'" diff_content_input = get_staged_diff()
if diff_content_input is None: exit(1)
if not diff_content_input.strip(): if diff_source_description == "'git diff --staged'": print("没有暂存的更改可以提交。", file=sys.stderr) exit(0) else: print(f"从 {diff_source_description} 读取的内容为空或仅含空白字符。将尝试让AI处理此情况。", file=sys.stderr)
max_diff_length = 800000 if len(diff_content_input) > max_diff_length: print(f"警告: 来自 {diff_source_description} 的 Diff 内容过长 ({len(diff_content_input)} characters),将截断为 {max_diff_length} characters。", file=sys.stderr) diff_content_input = diff_content_input[:max_diff_length] + "\n... (diff truncated)"
commit_msg = generate_commit_message(diff_content_input)
if commit_msg: print(commit_msg, flush=True) exit(0) else: print("最终未能生成 commit message。", file=sys.stderr) exit(1)
|