zhanggangyong/agent_management
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修复 Gitee 文件推送问题

Browse Source 
主要修复:
1. 修复 Gitea API 返回数组的处理
2. 修复获取 commit tree SHA 的路径问题
3. 添加空仓库初始提交支持
4. 添加 API 限流重试机制
5. 改为不自动初始化仓库，使用 Git Data API 推送初始提交

当前状态: Gitee 服务器严重 API 限流 (429)
This commit is contained in:
zhanggangyong
2026-01-29 15:55:38 +00:00
parent 8cb166fb58
commit 715139158e
1 changed files with 326 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitee_manager.py
+326 -46
View File
@@ -55,7 +55,8 @@ class GiteeManager:
self,
repo_name: str,
description: str = "",
private: bool = False
private: bool = False,
auto_init: bool = False
) -> Dict[str, Any]:
"""
创建 Gitee 仓库
@@ -64,6 +65,7 @@ class GiteeManager:
repo_name: 仓库名称
description: 仓库描述
private: 是否私有
auto_init: 是否自动初始化（默认不初始化，这样可以直接推送文件）
Returns:
仓库信息
@@ -74,50 +76,68 @@ class GiteeManager:
"name": repo_name,
"description": description,
"private": private,
"auto_init": True, # 自动初始化
"auto_init": auto_init, # 不自动初始化，后续直接推送文件
"default_branch": "main"
}
try:
response = requests.post(
url,
json=data,
headers=self._get_headers(),
timeout=30
)
if response.status_code == 201:
repo_info = response.json()
owner = repo_info.get("owner", {}).get("login", self.gitee_username)
logger.info(f"✅ Gitee 仓库创建成功: {owner}/{repo_name}")
return {
"success": True,
"repo_name": repo_name,
"owner": owner,
"clone_url": repo_info.get("clone_url"),
"html_url": repo_info.get("html_url"),
"ssh_url": repo_info.get("ssh_url")
}
elif response.status_code == 409:
# 仓库已存在
owner = self.gitee_username or self.gitee_owner
logger.warning(f"仓库已存在: {repo_name}")
return {
"success": True,
"repo_name": repo_name,
"owner": owner,
"clone_url": f"{self.gitee_base_url}/{owner}/{repo_name}.git",
"html_url": f"{self.gitee_base_url}/{owner}/{repo_name}",
"exists": True
}
else:
error_msg = response.json().get("message", response.text)
logger.error(f"创建仓库失败: {error_msg}")
return {"success": False, "error": error_msg}
import time
max_retries = 3
for attempt in range(max_retries):
try:
response = requests.post(
url,
json=data,
headers=self._get_headers(),
timeout=30
)
except Exception as e:
logger.error(f"创建仓库异常: {e}")
return {"success": False, "error": str(e)}
if response.status_code == 201:
repo_info = response.json()
owner = repo_info.get("owner", {}).get("login", self.gitee_username)
logger.info(f"✅ Gitee 仓库创建成功: {owner}/{repo_name}")
return {
"success": True,
"repo_name": repo_name,
"owner": owner,
"clone_url": repo_info.get("clone_url"),
"html_url": repo_info.get("html_url"),
"ssh_url": repo_info.get("ssh_url")
}
elif response.status_code == 409:
# 仓库已存在
owner = self.gitee_username or self.gitee_owner
logger.warning(f"仓库已存在: {repo_name}")
return {
"success": True,
"repo_name": repo_name,
"owner": owner,
"clone_url": f"{self.gitee_base_url}/{owner}/{repo_name}.git",
"html_url": f"{self.gitee_base_url}/{owner}/{repo_name}",
"exists": True
}
elif response.status_code == 429:
# API 限流，等待后重试
wait_time = (attempt + 1) * 2
logger.warning(f"API 限流，等待 {wait_time} 秒后重试 ({attempt + 1}/{max_retries})")
time.sleep(wait_time)
continue
else:
try:
error_msg = response.json().get("message", response.text)
except:
error_msg = response.text[:200] if response.text else f"HTTP {response.status_code}"
logger.error(f"创建仓库失败: {error_msg}")
return {"success": False, "error": error_msg}
except Exception as e:
logger.error(f"创建仓库异常 (尝试 {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
time.sleep(2)
continue
return {"success": False, "error": str(e)}
return {"success": False, "error": "API 限流，请稍后重试"}
def push_files(
self,
@@ -128,7 +148,7 @@ class GiteeManager:
owner: str = None
) -> Dict[str, Any]:
"""
推送文件到仓库
推送文件到仓库（使用 Git Data API 批量提交）
Args:
repo_name: 仓库名称
@@ -146,19 +166,51 @@ class GiteeManager:
repo_owner = owner or self.gitee_username or self.gitee_owner
results = {"success": True, "files": [], "owner": repo_owner}
logger.info(f"开始推送 {len(files)} 个文件到 {repo_owner}/{repo_name}")
# 检查仓库是否为空
is_empty = self._is_repo_empty(repo_owner, repo_name)
if is_empty:
# 空仓库：使用初始提交方式
logger.info("仓库为空，使用初始提交方式")
init_result = self._push_initial_commit(repo_owner, repo_name, files, commit_message, branch)
if init_result.get("success"):
logger.info(f"✅ 初始提交成功: {len(files)} 个文件")
results["files"] = [{"path": p, "status": "success"} for p in files.keys()]
return results
else:
logger.error(f"初始提交失败: {init_result.get('error')}")
results["success"] = False
results["error"] = init_result.get("error")
return results
# 非空仓库：尝试使用批量提交 API
try:
batch_result = self._push_files_batch(repo_owner, repo_name, files, commit_message, branch)
if batch_result.get("success"):
logger.info(f"✅ 批量推送成功: {len(files)} 个文件")
results["files"] = [{"path": p, "status": "success"} for p in files.keys()]
return results
else:
logger.warning(f"批量推送失败，尝试逐个推送: {batch_result.get('error')}")
except Exception as e:
logger.warning(f"批量推送异常，尝试逐个推送: {e}")
# 方法2：逐个推送文件
for file_path, content in files.items():
try:
# 使用 Contents API 创建/更新文件
url = f"{self.gitee_api_url}/repos/{repo_owner}/{repo_name}/contents/{file_path}"
# 检查文件是否存在
# 检查文件是否存在（获取 SHA）
check_response = requests.get(url, headers=self._get_headers(), timeout=10)
sha = None
if check_response.status_code == 200:
sha = check_response.json().get("sha")
data = {
"message": commit_message,
"message": f"{commit_message} - {file_path}",
"content": base64.b64encode(content.encode()).decode(),
"branch": branch
}
@@ -174,7 +226,7 @@ class GiteeManager:
)
# 添加短暂延迟避免 API 限流
time.sleep(0.5)
time.sleep(0.3)
if response.status_code in [200, 201]:
results["files"].append({
@@ -183,14 +235,17 @@ class GiteeManager:
})
logger.info(f"✅ 文件推送成功: {file_path}")
else:
error_msg = response.text[:200]
logger.error(f"❌ 文件推送失败: {file_path}, 状态码: {response.status_code}, 错误: {error_msg}")
results["files"].append({
"path": file_path,
"status": "failed",
"error": response.text
"error": error_msg
})
results["success"] = False
except Exception as e:
logger.error(f"❌ 文件推送异常: {file_path}, 错误: {e}")
results["files"].append({
"path": file_path,
"status": "failed",
@@ -198,8 +253,233 @@ class GiteeManager:
})
results["success"] = False
logger.info(f"推送完成: 成功 {sum(1 for f in results['files'] if f['status'] == 'success')}/{len(files)} 个文件")
return results
def _push_files_batch(
self,
owner: str,
repo_name: str,
files: Dict[str, str],
commit_message: str,
branch: str
) -> Dict[str, Any]:
"""
使用 Git Data API 批量推送文件
流程:
1. 获取当前分支的最新 commit SHA
2. 获取该 commit 的 tree SHA
3. 创建新的 blobs
4. 创建新的 tree
5. 创建新的 commit
6. 更新分支引用
"""
try:
headers = self._get_headers()
base_url = f"{self.gitee_api_url}/repos/{owner}/{repo_name}"
# 1. 获取当前分支的最新 commit
# Gitea API 返回的是数组
ref_url = f"{base_url}/git/refs/heads/{branch}"
ref_response = requests.get(ref_url, headers=headers, timeout=10)
if ref_response.status_code != 200:
return {"success": False, "error": f"获取分支引用失败: {ref_response.text}"}
ref_data = ref_response.json()
# 处理 Gitea API 可能返回数组或对象的情况
if isinstance(ref_data, list):
if not ref_data:
return {"success": False, "error": "无法获取分支引用"}
ref_data = ref_data[0]
latest_commit_sha = ref_data.get("object", {}).get("sha")
if not latest_commit_sha:
return {"success": False, "error": "无法获取最新 commit SHA"}
logger.info(f"获取到最新 commit SHA: {latest_commit_sha[:8]}...")
# 2. 获取该 commit 的 tree
commit_url = f"{base_url}/git/commits/{latest_commit_sha}"
commit_response = requests.get(commit_url, headers=headers, timeout=10)
if commit_response.status_code != 200:
return {"success": False, "error": f"获取 commit 信息失败: {commit_response.text}"}
commit_data = commit_response.json()
# Gitea API: tree 在 commit.tree 中
base_tree_sha = None
if commit_data.get("commit", {}).get("tree"):
base_tree_sha = commit_data["commit"]["tree"].get("sha")
elif commit_data.get("tree"):
base_tree_sha = commit_data["tree"].get("sha")
if not base_tree_sha:
logger.error(f"无法获取 tree SHA，commit 数据: {commit_data}")
return {"success": False, "error": "无法获取 tree SHA"}
logger.info(f"获取到 base tree SHA: {base_tree_sha[:8]}...")
# 3. 创建 tree 项目
tree_items = []
for file_path, content in files.items():
tree_items.append({
"path": file_path,
"mode": "100644",
"type": "blob",
"content": content
})
# 4. 创建新 tree
tree_url = f"{base_url}/git/trees"
tree_data = {
"base_tree": base_tree_sha,
"tree": tree_items
}
logger.info(f"创建新 tree，包含 {len(tree_items)} 个文件...")
tree_response = requests.post(tree_url, json=tree_data, headers=headers, timeout=120)
if tree_response.status_code not in [200, 201]:
return {"success": False, "error": f"创建 tree 失败: {tree_response.text[:200]}"}
new_tree_sha = tree_response.json().get("sha")
logger.info(f"新 tree SHA: {new_tree_sha[:8]}...")
# 5. 创建新 commit
commit_create_url = f"{base_url}/git/commits"
commit_create_data = {
"message": commit_message,
"tree": new_tree_sha,
"parents": [latest_commit_sha]
}
logger.info(f"创建新 commit...")
commit_create_response = requests.post(commit_create_url, json=commit_create_data, headers=headers, timeout=30)
if commit_create_response.status_code not in [200, 201]:
return {"success": False, "error": f"创建 commit 失败: {commit_create_response.text[:200]}"}
new_commit_sha = commit_create_response.json().get("sha")
logger.info(f"新 commit SHA: {new_commit_sha[:8]}...")
# 6. 更新分支引用
update_ref_data = {
"sha": new_commit_sha,
"force": False
}
logger.info(f"更新分支引用到新 commit...")
update_ref_response = requests.patch(ref_url, json=update_ref_data, headers=headers, timeout=10)
if update_ref_response.status_code not in [200, 201]:
return {"success": False, "error": f"更新分支引用失败: {update_ref_response.text[:200]}"}
logger.info(f"✅ 批量推送成功，commit: {new_commit_sha[:8]}")
return {"success": True, "commit_sha": new_commit_sha}
except Exception as e:
import traceback
logger.error(f"批量推送异常: {e}")
logger.error(traceback.format_exc())
return {"success": False, "error": str(e)}
def _is_repo_empty(self, owner: str, repo_name: str) -> bool:
"""检查仓库是否为空"""
try:
url = f"{self.gitee_api_url}/repos/{owner}/{repo_name}"
response = requests.get(url, headers=self._get_headers(), timeout=10)
if response.status_code == 200:
repo_data = response.json()
return repo_data.get("empty", True)
return True
except Exception as e:
logger.warning(f"检查仓库是否为空失败: {e}")
return True
def _push_initial_commit(
self,
owner: str,
repo_name: str,
files: Dict[str, str],
commit_message: str,
branch: str
) -> Dict[str, Any]:
"""
推送初始提交到空仓库
使用 Git Data API 创建初始 tree 和 commit
"""
try:
headers = self._get_headers()
base_url = f"{self.gitee_api_url}/repos/{owner}/{repo_name}"
# 1. 创建 tree（不需要 base_tree，这是初始提交）
tree_items = []
for file_path, content in files.items():
tree_items.append({
"path": file_path,
"mode": "100644",
"type": "blob",
"content": content
})
tree_url = f"{base_url}/git/trees"
tree_data = {
"tree": tree_items
}
logger.info(f"创建初始 tree，包含 {len(tree_items)} 个文件...")
tree_response = requests.post(tree_url, json=tree_data, headers=headers, timeout=120)
if tree_response.status_code not in [200, 201]:
return {"success": False, "error": f"创建初始 tree 失败: {tree_response.text[:300]}"}
new_tree_sha = tree_response.json().get("sha")
logger.info(f"初始 tree SHA: {new_tree_sha[:8]}...")
# 2. 创建初始 commit（没有 parents）
commit_create_url = f"{base_url}/git/commits"
commit_create_data = {
"message": commit_message,
"tree": new_tree_sha
# 不设置 parents，这是初始提交
}
logger.info(f"创建初始 commit...")
commit_create_response = requests.post(commit_create_url, json=commit_create_data, headers=headers, timeout=30)
if commit_create_response.status_code not in [200, 201]:
return {"success": False, "error": f"创建初始 commit 失败: {commit_create_response.text[:300]}"}
new_commit_sha = commit_create_response.json().get("sha")
logger.info(f"初始 commit SHA: {new_commit_sha[:8]}...")
# 3. 创建分支引用
ref_url = f"{base_url}/git/refs"
ref_data = {
"ref": f"refs/heads/{branch}",
"sha": new_commit_sha
}
logger.info(f"创建分支引用 {branch}...")
ref_response = requests.post(ref_url, json=ref_data, headers=headers, timeout=10)
if ref_response.status_code not in [200, 201]:
return {"success": False, "error": f"创建分支引用失败: {ref_response.text[:300]}"}
logger.info(f"✅ 初始提交成功，commit: {new_commit_sha[:8]}")
return {"success": True, "commit_sha": new_commit_sha}
except Exception as e:
import traceback
logger.error(f"初始提交异常: {e}")
logger.error(traceback.format_exc())
return {"success": False, "error": str(e)}
def get_action_status(self, repo_name: str) -> Dict[str, Any]:
"""
获取 Gitee Action 运行状态