155 lines
4.6 KiB
Python
155 lines
4.6 KiB
Python
"""
|
||
Git 操作工具模块。
|
||
负责在 CI 环境中检出上一版本代码、获取变更文件列表及提交元信息。
|
||
"""
|
||
|
||
import os
|
||
import subprocess
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import List, Optional
|
||
|
||
|
||
@dataclass
|
||
class CommitInfo:
|
||
"""单次 Git 提交的元信息。"""
|
||
|
||
sha: str
|
||
author: str
|
||
commit_time: str
|
||
message: str
|
||
|
||
|
||
def run_git(args: List[str], cwd: Optional[Path] = None) -> str:
|
||
"""
|
||
执行 git 命令并返回标准输出。
|
||
|
||
:param args: git 子命令及参数，如 ["log", "-1", "--format=%H"]
|
||
:param cwd: 工作目录，默认为当前目录
|
||
:return: 命令 stdout 文本（已 strip）
|
||
:raises RuntimeError: git 命令执行失败时抛出
|
||
"""
|
||
cmd = ["git"] + args
|
||
result = subprocess.run(
|
||
cmd,
|
||
cwd=cwd,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f"Git 命令失败: {' '.join(cmd)}\n{result.stderr}")
|
||
return result.stdout.strip()
|
||
|
||
|
||
def get_current_commit() -> CommitInfo:
|
||
"""
|
||
获取当前 HEAD 提交的元信息（推送人、时间等，用于通知模板）。
|
||
|
||
:return: CommitInfo 对象
|
||
"""
|
||
sha = run_git(["rev-parse", "HEAD"])
|
||
author = run_git(["log", "-1", "--format=%an"])
|
||
commit_time = run_git(["log", "-1", "--format=%ci"])
|
||
message = run_git(["log", "-1", "--format=%s"])
|
||
return CommitInfo(sha=sha, author=author, commit_time=commit_time, message=message)
|
||
|
||
|
||
def get_previous_commit_sha() -> Optional[str]:
|
||
"""
|
||
获取上一次提交的 SHA（HEAD~1）。
|
||
若是首次提交则返回 None。
|
||
|
||
:return: 上一 commit SHA，或 None
|
||
"""
|
||
try:
|
||
return run_git(["rev-parse", "HEAD~1"])
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def checkout_commit(sha: str, worktree_dir: Path) -> None:
|
||
"""
|
||
将指定 commit 的代码检出到独立工作目录（不影响当前工作区）。
|
||
|
||
:param sha: 目标 commit SHA
|
||
:param worktree_dir: git worktree 目录
|
||
"""
|
||
worktree_dir.parent.mkdir(parents=True, exist_ok=True)
|
||
if worktree_dir.exists():
|
||
# 已存在则先移除旧 worktree
|
||
run_git(["worktree", "remove", "--force", str(worktree_dir)])
|
||
run_git(["worktree", "add", str(worktree_dir), sha])
|
||
|
||
|
||
def get_changed_java_controller_files(base_sha: str, head_sha: str) -> List[str]:
|
||
"""
|
||
获取两次提交之间变更的 Controller 相关 Java 文件路径。
|
||
|
||
:param base_sha: 基准 commit（旧版本）
|
||
:param head_sha: 目标 commit（新版本）
|
||
:return: 相对路径列表，如 ["src/main/java/.../UserController.java"]
|
||
"""
|
||
diff_output = run_git(["diff", "--name-only", base_sha, head_sha])
|
||
if not diff_output:
|
||
return []
|
||
|
||
changed = []
|
||
for line in diff_output.splitlines():
|
||
line = line.strip()
|
||
if line.endswith(".java") and "Controller" in line:
|
||
changed.append(line.replace("\\", "/"))
|
||
return changed
|
||
|
||
|
||
def get_controller_files_diff(base_sha: str, head_sha: str, changed_files: List[str]) -> str:
|
||
"""
|
||
获取变更 Controller 文件的 Git diff，供 LLM 审核接口参数变更时参考。
|
||
|
||
:param base_sha: 旧版本 commit SHA
|
||
:param head_sha: 新版本 commit SHA
|
||
:param changed_files: 变更文件相对路径列表
|
||
:return: diff 文本
|
||
"""
|
||
if not changed_files:
|
||
return ""
|
||
|
||
try:
|
||
return run_git(["diff", base_sha, head_sha, "--"] + changed_files)
|
||
except RuntimeError as exc:
|
||
print(f"[警告] 获取 Git diff 失败: {exc}")
|
||
return ""
|
||
|
||
|
||
def get_file_content_at_commit(commit_sha: str, file_path: str) -> Optional[str]:
|
||
"""
|
||
读取指定 commit 下某个文件的内容（无需 git worktree，更快）。
|
||
|
||
:param commit_sha: commit SHA
|
||
:param file_path: 相对仓库根目录的文件路径
|
||
:return: 文件内容；该 commit 中不存在则返回 None
|
||
"""
|
||
try:
|
||
return run_git(["show", f"{commit_sha}:{file_path}"])
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def prepare_worktrees(repo_root: Path) -> tuple:
|
||
"""
|
||
准备新旧两个版本的代码工作目录，供 AST 解析器分别扫描。
|
||
|
||
:param repo_root: 仓库根目录
|
||
:return: (新版本目录, 旧版本目录, 旧版本SHA)；首次提交时旧版本目录为 None
|
||
"""
|
||
prev_sha = get_previous_commit_sha()
|
||
current_dir = repo_root
|
||
|
||
if prev_sha is None:
|
||
return current_dir, None, None
|
||
|
||
prev_dir = repo_root / ".gitea" / ".cache" / "prev-worktree"
|
||
checkout_commit(prev_sha, prev_dir)
|
||
return current_dir, prev_dir, prev_sha
|