550 lines
24 KiB
Python
550 lines
24 KiB
Python
"""
|
||
Git命令封装工具类
|
||
实现Git仓库的克隆、更新检测、增量拉取等功能
|
||
"""
|
||
import subprocess
|
||
import os
|
||
from typing import Tuple, Dict, List
|
||
from loguru import logger
|
||
from config import settings
|
||
|
||
|
||
class GitTool:
|
||
def __init__(self, user_id: str = "test", repo_id: str = "test", git_config: dict = None,
|
||
git_url: str = None, branch: str = None, protocol: str = None,
|
||
https_token: str = None, ssh_key: str = None, password: str = None, local_repo_path: str = None):
|
||
"""
|
||
初始化Git工具类
|
||
|
||
Args:
|
||
user_id: 用户ID
|
||
repo_id: 仓库ID
|
||
git_config: Git配置信息
|
||
git_url: Git仓库URL（直接参数，优先级高于git_config）
|
||
branch: Git分支（直接参数，优先级高于git_config）
|
||
protocol: Git协议（直接参数，优先级高于git_config）
|
||
https_token: HTTPS令牌（直接参数，优先级高于git_config）
|
||
ssh_key: SSH密钥（直接参数，优先级高于git_config）
|
||
password: SSH密码（直接参数，优先级高于git_config）
|
||
local_repo_path: 本地仓库路径（直接参数，优先级高于git_config）
|
||
"""
|
||
self.user_id = user_id
|
||
self.repo_id = repo_id
|
||
|
||
# 优先使用直接参数，如果没有则使用git_config
|
||
if git_config:
|
||
self.git_url = git_url or git_config.get("git_url")
|
||
# 如果branch为空字符串，保持为空字符串，不使用默认值
|
||
self.branch = branch if branch is not None else git_config.get("branch", "")
|
||
self.protocol = protocol or git_config.get("protocol", "https")
|
||
self.ssh_key = ssh_key or git_config.get("ssh_key")
|
||
self.https_token = https_token or git_config.get("https_token")
|
||
self.password = password or git_config.get("password")
|
||
# 本地结构化存储路径
|
||
self.local_repo_path = local_repo_path or git_config.get("local_repo_path")
|
||
else:
|
||
self.git_url = git_url
|
||
# 如果branch为空字符串，保持为空字符串，不使用默认值
|
||
self.branch = branch if branch is not None else ""
|
||
self.protocol = protocol or "https"
|
||
self.ssh_key = ssh_key
|
||
self.https_token = https_token
|
||
self.password = password
|
||
self.local_repo_path = local_repo_path
|
||
|
||
if not self.local_repo_path:
|
||
self.local_repo_path = os.path.join(settings.GIT_LOCAL_STORAGE_ROOT, user_id, repo_id)
|
||
# 初始化Git环境
|
||
try:
|
||
self._init_git_env()
|
||
except Exception as e:
|
||
import traceback
|
||
logger.error(f"初始化Git环境失败: {e}")
|
||
logger.error(f"错误类型: {type(e).__name__}")
|
||
logger.error(f"错误堆栈: {traceback.format_exc()}")
|
||
raise
|
||
|
||
def _init_git_env(self):
|
||
"""
|
||
初始化Git环境（SSH密钥配置）
|
||
"""
|
||
logger.info(f"初始化Git环境，密码：{self.password is not None}, SSH密钥：{self.ssh_key is not None}")
|
||
|
||
# 设置 SSH variant 为 ssh，避免 "simple" variant 不支持端口问题
|
||
os.environ["GIT_SSH_VARIANT"] = "ssh"
|
||
|
||
# 提取SSH端口（如果在URL中指定）
|
||
ssh_port = 22
|
||
if self.git_url and self.git_url.startswith('ssh://'):
|
||
# 解析SSH URL以获取端口
|
||
import re
|
||
port_match = re.search(r'ssh://[^:]+:([0-9]+)/', self.git_url)
|
||
if port_match:
|
||
ssh_port = port_match.group(1)
|
||
|
||
if self.ssh_key:
|
||
# 解密SSH私钥，写入临时文件，配置Git SSH
|
||
ssh_key_path = f"/tmp/ssh_key_{self.user_id}_{self.repo_id}"
|
||
with open(ssh_key_path, "w") as f:
|
||
f.write(self.ssh_key)
|
||
if not self.ssh_key.endswith('\n'):
|
||
f.write('\n')
|
||
os.chmod(ssh_key_path, 0o600)
|
||
os.environ["GIT_SSH_COMMAND"] = f"ssh -i {ssh_key_path} -p {ssh_port} -o StrictHostKeyChecking=no"
|
||
logger.info(f"使用SSH密钥认证，命令：{os.environ['GIT_SSH_COMMAND']}")
|
||
elif self.password:
|
||
# 检查是否在Windows系统上
|
||
import platform
|
||
system = platform.system()
|
||
logger.info(f"本地系统类型：{system}")
|
||
|
||
# 使用 GIT_ASKPASS 机制，这是 Git 官方推荐的密码输入方式
|
||
# 创建 askpass 脚本
|
||
askpass_script = r'''#!/usr/bin/env python3
|
||
import sys
|
||
# 密码文件路径通过环境变量传递
|
||
import os
|
||
password_file = os.environ.get('GIT_ASKPASS_PASSWORD_FILE', '')
|
||
if password_file and os.path.exists(password_file):
|
||
with open(password_file, 'r', encoding='utf-8') as f:
|
||
print(f.read().strip())
|
||
else:
|
||
print('111111') # 默认密码
|
||
'''
|
||
|
||
import tempfile
|
||
import sys as sys_module
|
||
|
||
# 写入 askpass 脚本
|
||
askpass_file = tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8')
|
||
askpass_file.write(askpass_script)
|
||
askpass_file.close()
|
||
askpass_path = askpass_file.name
|
||
|
||
# 写入密码文件
|
||
password_file = tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8')
|
||
password_file.write(self.password)
|
||
password_file.close()
|
||
password_file_path = password_file.name
|
||
|
||
# 设置环境变量
|
||
os.environ['GIT_ASKPASS'] = sys_module.executable.replace('\\', '/') + ' "' + askpass_path.replace('\\', '/') + '"'
|
||
os.environ['GIT_ASKPASS_PASSWORD_FILE'] = password_file_path.replace('\\', '/')
|
||
os.environ['GIT_TERMINAL_PROMPT'] = '0' # 禁用 Git 的终端提示
|
||
|
||
# 设置 SSH 使用交互式模式
|
||
os.environ['GIT_SSH_COMMAND'] = f'ssh -o StrictHostKeyChecking=no -o BatchMode=no'
|
||
|
||
logger.info(f"使用 GIT_ASKPASS 机制进行密码认证")
|
||
logger.info(f"GIT_ASKPASS: {os.environ['GIT_ASKPASS']}")
|
||
logger.info(f"GIT_SSH_COMMAND: {os.environ['GIT_SSH_COMMAND']}")
|
||
logger.info("使用密码认证，自动使用配置中的密码")
|
||
|
||
def clone_repo(self) -> bool:
|
||
"""
|
||
克隆Git仓库
|
||
|
||
Returns:
|
||
bool: 是否成功克隆
|
||
"""
|
||
if not os.path.exists(self.local_repo_path):
|
||
os.makedirs(os.path.dirname(self.local_repo_path), exist_ok=True)
|
||
|
||
# 尝试两种URL格式：原始URL和带/不带.git后缀的URL
|
||
test_urls = [self.git_url]
|
||
|
||
# 如果是Git daemon协议，尝试两种格式
|
||
if self.git_url.startswith("git://"):
|
||
if ".git" in self.git_url:
|
||
# 尝试不带.git后缀的格式
|
||
test_urls.append(self.git_url.replace(".git", ""))
|
||
else:
|
||
# 尝试带.git后缀的格式
|
||
test_urls.append(f"{self.git_url}.git")
|
||
# 如果是SSH协议且路径不包含.git后缀，尝试带.git后缀的格式
|
||
elif self.git_url.startswith("ssh://") and ".git" not in self.git_url:
|
||
test_urls.append(f"{self.git_url}.git")
|
||
|
||
for url in test_urls:
|
||
# 根据分支是否为空决定克隆方式
|
||
if self.branch:
|
||
# 有指定分支，使用带分支克隆
|
||
cmd = [
|
||
"git", "clone", "--single-branch",
|
||
"--branch", self.branch, url, self.local_repo_path
|
||
]
|
||
logger.info(f"执行Git克隆命令: {' '.join(cmd)}")
|
||
# 捕获输出，避免密码提示 #BUG: 对于云服务器方式不能使用git pass密码验证，显示permission denied
|
||
res = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='replace')
|
||
|
||
if res.returncode == 0:
|
||
# 克隆成功，更新git_url
|
||
self.git_url = url
|
||
# 克隆后校验
|
||
self._check_repo_integrity()
|
||
logger.info(f"Git仓库克隆成功: {self.local_repo_path}")
|
||
return True
|
||
else:
|
||
logger.error(f"Git克隆失败: {res.stderr}")
|
||
# 清理失败的克隆尝试
|
||
if os.path.exists(self.local_repo_path):
|
||
import shutil
|
||
shutil.rmtree(self.local_repo_path)
|
||
|
||
# 尝试不带分支克隆（适用于空分支或分支不存在的情况）
|
||
cmd_no_branch = [
|
||
"git", "clone", url, self.local_repo_path
|
||
]
|
||
logger.info(f"尝试不带分支克隆: {' '.join(cmd_no_branch)}")
|
||
res_no_branch = subprocess.run(cmd_no_branch, capture_output=True, text=True, encoding='utf-8', errors='replace')
|
||
if res_no_branch.returncode == 0:
|
||
# 克隆成功，更新git_url
|
||
self.git_url = url
|
||
# 克隆后校验
|
||
self._check_repo_integrity()
|
||
logger.info(f"Git仓库克隆成功(不带分支): {self.local_repo_path}")
|
||
return True
|
||
else:
|
||
logger.error(f"不带分支克隆失败: {res_no_branch.stderr}")
|
||
|
||
# 继续尝试其他URL格式
|
||
continue
|
||
|
||
# 所有格式都尝试失败
|
||
raise Exception("所有Git克隆尝试都失败")
|
||
else:
|
||
# 仓库已存在，执行完整性检查
|
||
logger.info(f"Git仓库已存在: {self.local_repo_path}")
|
||
# 检查仓库完整性
|
||
self._check_repo_integrity()
|
||
# 尝试获取当前分支
|
||
try:
|
||
current_branch = subprocess.run(
|
||
["git", "rev-parse", "--abbrev-ref", "HEAD"],
|
||
cwd=self.local_repo_path,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding='utf-8',
|
||
errors='replace'
|
||
).stdout.strip()
|
||
if current_branch:
|
||
self.branch = current_branch
|
||
logger.info(f"检测到本地仓库分支: {current_branch}")
|
||
# 尝试获取远程URL
|
||
remote_url = subprocess.run(
|
||
["git", "config", "--get", "remote.origin.url"],
|
||
cwd=self.local_repo_path,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding='utf-8',
|
||
errors='replace'
|
||
).stdout.strip()
|
||
if remote_url:
|
||
self.git_url = remote_url
|
||
logger.info(f"检测到远程URL: {remote_url}")
|
||
except Exception as e:
|
||
logger.warning(f"获取本地仓库信息失败: {e}")
|
||
return False
|
||
|
||
def _check_repo_integrity(self):
|
||
"""
|
||
仓库完整性校验（git fsck）+ 支持的编程语言检测
|
||
"""
|
||
# 执行git fsck
|
||
try:
|
||
subprocess.run(["git", "fsck"], cwd=self.local_repo_path, check=True, capture_output=True, text=True, encoding='utf-8', errors='replace')
|
||
logger.info(f"Git仓库完整性校验成功: {self.local_repo_path}")
|
||
except subprocess.CalledProcessError as e:
|
||
logger.warning(f"Git仓库完整性校验失败: {e.stderr}")
|
||
|
||
# 扫描文件类型，记录支持的编程语言
|
||
support_lang = self._detect_support_lang()
|
||
logger.info(f"检测到支持的编程语言: {support_lang}")
|
||
return support_lang
|
||
|
||
def _detect_support_lang(self) -> List[str]:
|
||
"""
|
||
检测仓库支持的编程语言
|
||
|
||
Returns:
|
||
List[str]: 支持的编程语言列表
|
||
"""
|
||
lang_extensions = {
|
||
"python": [".py"],
|
||
"java": [".java"],
|
||
"go": [".go"],
|
||
"javascript": [".js", ".jsx"],
|
||
"typescript": [".ts", ".tsx"],
|
||
"c": [".c", ".h"],
|
||
"cpp": [".cpp", ".hpp", ".cc"],
|
||
"csharp": [".cs"],
|
||
"rust": [".rs"],
|
||
"php": [".php"],
|
||
"ruby": [".rb"],
|
||
"swift": [".swift"],
|
||
"kotlin": [".kt"],
|
||
"scala": [".scala"]
|
||
}
|
||
|
||
support_lang = []
|
||
for root, dirs, files in os.walk(self.local_repo_path):
|
||
# 跳过.git目录
|
||
if ".git" in dirs:
|
||
dirs.remove(".git")
|
||
# 跳过其他常见的非代码目录
|
||
dirs_to_skip = ["node_modules", "venv", "dist", "build", "__pycache__"]
|
||
dirs[:] = [d for d in dirs if d not in dirs_to_skip]
|
||
|
||
for file in files:
|
||
for lang, extensions in lang_extensions.items():
|
||
if any(file.endswith(ext) for ext in extensions):
|
||
if lang not in support_lang:
|
||
support_lang.append(lang)
|
||
break
|
||
|
||
return support_lang
|
||
|
||
def detect_remote_update(self) -> Tuple[bool, str, str]:
|
||
"""
|
||
远程更新检测
|
||
|
||
Returns:
|
||
Tuple[bool, str, str]: (是否有更新, 本地commit ID, 远程commit ID)
|
||
"""
|
||
# 确保仓库存在
|
||
if not os.path.exists(self.local_repo_path):
|
||
raise Exception(f"Git仓库不存在: {self.local_repo_path}")
|
||
|
||
# 拉取远程commit记录
|
||
try:
|
||
# 修改fetch命令，只获取远程更新而不直接更新本地分支
|
||
subprocess.run(["git", "fetch", "origin", self.branch],
|
||
cwd=self.local_repo_path, check=True, capture_output=True, text=True, encoding='utf-8',errors="replace")
|
||
except subprocess.CalledProcessError as e:
|
||
logger.error(f"Git fetch失败: {e.stderr}")
|
||
raise
|
||
|
||
# 获取本地/远程commit ID
|
||
local_commit = subprocess.run(["git", "rev-parse", "HEAD"],
|
||
cwd=self.local_repo_path, capture_output=True, text=True, encoding='utf-8', errors='replace').stdout.strip()
|
||
remote_commit = subprocess.run(["git", "rev-parse", f"origin/{self.branch}"],
|
||
cwd=self.local_repo_path, capture_output=True, text=True, encoding='utf-8', errors='replace').stdout.strip()
|
||
|
||
has_update = local_commit != remote_commit
|
||
logger.info(f"Git更新检测: 本地={local_commit[:7]}, 远程={remote_commit[:7]}, 有更新={has_update}")
|
||
return has_update, local_commit, remote_commit
|
||
|
||
def incremental_pull(self, local_commit: str, remote_commit: str) -> Dict[str, List[str]]:
|
||
"""
|
||
增量拉取代码+解析文件变更
|
||
|
||
Args:
|
||
local_commit: 本地commit ID
|
||
remote_commit: 远程commit ID
|
||
|
||
Returns:
|
||
Dict[str, List[str]]: 文件变更集
|
||
"""
|
||
# 快进合并到远程最新版本
|
||
try:
|
||
subprocess.run(["git", "merge", "--ff-only", f"origin/{self.branch}"],
|
||
cwd=self.local_repo_path, check=True, capture_output=True, text=True, encoding='utf-8', errors="replace")
|
||
logger.info(f"Git快进合并成功: {self.branch}")
|
||
except subprocess.CalledProcessError as e:
|
||
logger.error(f"Git合并失败: {e.stderr}")
|
||
raise
|
||
|
||
# 提取增量commit的文件变更
|
||
delta_commits = subprocess.run(["git", "log", "--pretty=format:%H", f"{local_commit}..{remote_commit}"],
|
||
cwd=self.local_repo_path, capture_output=True, text=True, encoding='utf-8', errors='replace').stdout.split()
|
||
|
||
# 解析文件变更为ADD/MODIFY/DELETE
|
||
delta_files = self._parse_delta_files(delta_commits)
|
||
logger.info(f"Git增量变更: ADD={len(delta_files['ADD'])}, MODIFY={len(delta_files['MODIFY'])}, DELETE={len(delta_files['DELETE'])}, RENAME={len(delta_files['RENAME'])}")
|
||
return delta_files
|
||
|
||
def _parse_delta_files(self, delta_commits: List[str]) -> Dict[str, List[str]]:
|
||
"""
|
||
解析文件变更集
|
||
|
||
Args:
|
||
delta_commits: 增量commit列表
|
||
|
||
Returns:
|
||
Dict[str, List[str]]: 文件变更集
|
||
"""
|
||
add_files, modify_files, delete_files, rename_files = [], [], [], []
|
||
|
||
for commit in delta_commits:
|
||
# git show --name-status 获取文件变更
|
||
res = subprocess.run(["git", "show", "--name-status", commit],
|
||
cwd=self.local_repo_path, capture_output=True, text=True, encoding='utf-8', errors='replace').stdout
|
||
|
||
for line in res.splitlines():
|
||
if not line:
|
||
continue
|
||
# 解析状态和文件路径
|
||
if "\t" in line:
|
||
status, file_path = line.split("\t", 1)
|
||
|
||
if status.startswith("R"):
|
||
# 处理重命名：分割旧路径和新路径
|
||
old_path, new_path = file_path.split("\t", 1)
|
||
old_full_path = os.path.join(self.local_repo_path, old_path)
|
||
new_full_path = os.path.join(self.local_repo_path, new_path)
|
||
if status == "R100":
|
||
# 对于R100（仅重命名无内容变更），处理为仅修改对应document的元数据（file_path、chunk_id、func_id）
|
||
rename_files.append((old_full_path, new_full_path))
|
||
else:
|
||
# 旧路径视为删除，新路径视为添加
|
||
delete_files.append(old_full_path)
|
||
add_files.append(new_full_path)
|
||
elif status == "A":
|
||
full_path = os.path.join(self.local_repo_path, file_path)
|
||
add_files.append(full_path)
|
||
elif status == "M":
|
||
full_path = os.path.join(self.local_repo_path, file_path)
|
||
modify_files.append(full_path)
|
||
elif status == "D":
|
||
full_path = os.path.join(self.local_repo_path, file_path)
|
||
delete_files.append(full_path)
|
||
|
||
# 去重并返回
|
||
return {
|
||
"ADD": list(set(add_files)),
|
||
"MODIFY": list(set(modify_files)),
|
||
"DELETE": list(set(delete_files)),
|
||
"RENAME": list(set(rename_files)),
|
||
}
|
||
|
||
def get_current_commit(self) -> str:
|
||
"""
|
||
获取当前commit ID
|
||
|
||
Returns:
|
||
str: 当前commit ID
|
||
"""
|
||
if not os.path.exists(self.local_repo_path):
|
||
raise Exception(f"Git仓库不存在: {self.local_repo_path}")
|
||
|
||
commit_id = subprocess.run(["git", "rev-parse", "HEAD"],
|
||
cwd=self.local_repo_path, capture_output=True, text=True, encoding='utf-8', errors='replace').stdout.strip()
|
||
return commit_id
|
||
|
||
def get_repo_info(self) -> Dict[str, str]:
|
||
"""
|
||
获取仓库信息
|
||
|
||
Returns:
|
||
Dict[str, str]: 仓库信息
|
||
"""
|
||
if not os.path.exists(self.local_repo_path):
|
||
raise Exception(f"Git仓库不存在: {self.local_repo_path}")
|
||
|
||
# 获取仓库URL
|
||
remote_url = subprocess.run(["git", "config", "--get", "remote.origin.url"],
|
||
cwd=self.local_repo_path, capture_output=True, text=True, encoding='utf-8', errors='replace').stdout.strip()
|
||
|
||
# 获取当前分支
|
||
current_branch = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"],
|
||
cwd=self.local_repo_path, capture_output=True, text=True, encoding='utf-8', errors='replace').stdout.strip()
|
||
|
||
# 获取当前commit
|
||
current_commit = self.get_current_commit()
|
||
|
||
return {
|
||
"remote_url": remote_url,
|
||
"current_branch": current_branch,
|
||
"current_commit": current_commit,
|
||
"local_path": self.local_repo_path
|
||
}
|
||
|
||
def test_connection(self) -> bool:
|
||
"""
|
||
测试Git连接
|
||
|
||
Returns:
|
||
bool: 是否连接成功
|
||
"""
|
||
if not self.git_url:
|
||
raise Exception("Git仓库URL未设置")
|
||
|
||
# 尝试两种URL格式：原始URL和带/不带.git后缀的URL
|
||
test_urls = [self.git_url]
|
||
|
||
# 如果是Git daemon协议，尝试两种格式
|
||
if self.git_url.startswith("git://"):
|
||
if ".git" in self.git_url:
|
||
# 尝试不带.git后缀的格式
|
||
test_urls.append(self.git_url.replace(".git", ""))
|
||
else:
|
||
# 尝试带.git后缀的格式
|
||
test_urls.append(f"{self.git_url}.git")
|
||
|
||
for url in test_urls:
|
||
logger.info(f"测试Git连接: {url}")
|
||
|
||
# 尝试执行git ls-remote命令来测试连接
|
||
try:
|
||
cmd = ["git", "ls-remote", "--heads", url, f"refs/heads/{self.branch}"]
|
||
logger.info(f"执行Git连接测试命令: {' '.join(cmd)}")
|
||
|
||
res = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='replace')
|
||
|
||
if res.returncode == 0:
|
||
if self.branch in res.stdout or not res.stdout:
|
||
logger.info("Git连接测试成功！")
|
||
self.git_url = url
|
||
return True
|
||
else:
|
||
logger.warning(f"Git连接测试失败：分支 {self.branch} 不存在")
|
||
continue
|
||
else:
|
||
logger.error(f"Git连接测试失败: {res.stderr}")
|
||
continue
|
||
|
||
except Exception as e:
|
||
logger.error(f"Git连接测试异常: {e}")
|
||
continue
|
||
|
||
logger.error("所有Git连接测试格式都失败")
|
||
return False
|
||
|
||
def get_file_blob_sha(self, file_path: str) -> str:
|
||
"""获取文件的Git blob SHA值"""
|
||
try:
|
||
rel_path = os.path.relpath(file_path, self.local_repo_path)
|
||
|
||
cmd = ["git", "ls-files", "-s", rel_path]
|
||
result = subprocess.run(cmd, cwd=self.local_repo_path, capture_output=True, text=True)
|
||
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
parts = result.stdout.strip().split()
|
||
if len(parts) >= 3:
|
||
return parts[1]
|
||
except Exception as e:
|
||
logger.warning(f"获取文件blob SHA失败 {file_path}: {e}")
|
||
|
||
return ""
|
||
|
||
def get_all_file_shas(self) -> Dict[str, str]:
|
||
"""获取仓库中所有支持语言文件的blob SHA"""
|
||
from sync.ast_parser import ASTParser
|
||
|
||
file_shas = {}
|
||
support_lang = self._detect_support_lang()
|
||
|
||
for root, dirs, files in os.walk(self.local_repo_path):
|
||
if ".git" in dirs:
|
||
dirs.remove(".git")
|
||
|
||
for file in files:
|
||
file_path = os.path.join(root, file)
|
||
lang = ASTParser.detect_language(file_path)
|
||
if lang and lang in support_lang:
|
||
blob_sha = self.get_file_blob_sha(file_path)
|
||
if blob_sha:
|
||
file_shas[file_path] = blob_sha
|
||
|
||
return file_shas
|