zhanggangyong
b8c61a07b2
修复内容: 1. 空仓库使用 Contents API 先创建初始文件 2. 批量推送失败时回退到逐个推送 3. 简化 Dockerfile 避免 WAF 拦截 4. 添加 API 限流重试机制 测试验证: 10 个文件全部推送成功
714 lines
27 KiB
Python
714 lines
27 KiB
Python
"""
|
||
Gitee 仓库管理模块
|
||
负责创建仓库、推送代码、查询 Action 状态
|
||
"""
|
||
|
||
import os
|
||
import logging
|
||
import requests
|
||
import base64
|
||
from typing import Dict, List, Optional, Any
|
||
from datetime import datetime
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class GiteeManager:
|
||
"""Gitee 仓库管理器"""
|
||
|
||
def __init__(self):
|
||
# Gitee 配置
|
||
self.gitee_api_url = os.getenv("GITEE_API_URL", "http://gitee.ath.cx:3000/api/v1")
|
||
self.gitee_token = os.getenv("GITEE_TOKEN", "")
|
||
self.gitee_username = os.getenv("GITEE_USERNAME", "zhanggangyong")
|
||
self.gitee_password = os.getenv("GITEE_PASSWORD", "")
|
||
self.gitee_owner = os.getenv("GITEE_OWNER", "xiaohei")
|
||
self.gitee_base_url = os.getenv("GITEE_BASE_URL", "http://gitee.ath.cx:3000")
|
||
|
||
# 模板仓库（参考 http://gitee.ath.cx:3000/xiaohei/cicd-AKS）
|
||
self.template_repo = os.getenv("GITEE_TEMPLATE_REPO", "cicd-AKS")
|
||
|
||
# ACR 配置
|
||
self.acr_registry = os.getenv("ACR_REGISTRY", "agnettaiji.azurecr.io")
|
||
self.acr_namespace = os.getenv("ACR_NAMESPACE", "ai-agents")
|
||
|
||
def _get_headers(self) -> Dict[str, str]:
|
||
"""获取请求头（支持 Token 或 Basic Auth）"""
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json"
|
||
}
|
||
|
||
if self.gitee_token:
|
||
headers["Authorization"] = f"token {self.gitee_token}"
|
||
elif self.gitee_username and self.gitee_password:
|
||
# 使用 Basic Auth
|
||
import base64
|
||
credentials = base64.b64encode(
|
||
f"{self.gitee_username}:{self.gitee_password}".encode()
|
||
).decode()
|
||
headers["Authorization"] = f"Basic {credentials}"
|
||
|
||
return headers
|
||
|
||
def create_repository(
|
||
self,
|
||
repo_name: str,
|
||
description: str = "",
|
||
private: bool = False,
|
||
auto_init: bool = False
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
创建 Gitee 仓库
|
||
|
||
Args:
|
||
repo_name: 仓库名称
|
||
description: 仓库描述
|
||
private: 是否私有
|
||
auto_init: 是否自动初始化（默认不初始化，这样可以直接推送文件）
|
||
|
||
Returns:
|
||
仓库信息
|
||
"""
|
||
url = f"{self.gitee_api_url}/user/repos"
|
||
|
||
data = {
|
||
"name": repo_name,
|
||
"description": description,
|
||
"private": private,
|
||
"auto_init": auto_init, # 不自动初始化，后续直接推送文件
|
||
"default_branch": "main"
|
||
}
|
||
|
||
import time
|
||
max_retries = 3
|
||
|
||
for attempt in range(max_retries):
|
||
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
|
||
}
|
||
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,
|
||
repo_name: str,
|
||
files: Dict[str, str],
|
||
commit_message: str = "Initial commit",
|
||
branch: str = "main",
|
||
owner: str = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
推送文件到仓库（使用 Git Data API 批量提交）
|
||
|
||
Args:
|
||
repo_name: 仓库名称
|
||
files: 文件字典 {path: content}
|
||
commit_message: 提交信息
|
||
branch: 分支名
|
||
owner: 仓库所有者（默认使用配置的用户名）
|
||
|
||
Returns:
|
||
推送结果
|
||
"""
|
||
import time
|
||
|
||
# 使用当前用户作为所有者
|
||
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": f"{commit_message} - {file_path}",
|
||
"content": base64.b64encode(content.encode()).decode(),
|
||
"branch": branch
|
||
}
|
||
|
||
if sha:
|
||
data["sha"] = sha
|
||
|
||
response = requests.put(
|
||
url,
|
||
json=data,
|
||
headers=self._get_headers(),
|
||
timeout=30
|
||
)
|
||
|
||
# 添加短暂延迟避免 API 限流
|
||
time.sleep(0.3)
|
||
|
||
if response.status_code in [200, 201]:
|
||
results["files"].append({
|
||
"path": file_path,
|
||
"status": "success"
|
||
})
|
||
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": error_msg
|
||
})
|
||
results["success"] = False
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 文件推送异常: {file_path}, 错误: {e}")
|
||
results["files"].append({
|
||
"path": file_path,
|
||
"status": "failed",
|
||
"error": str(e)
|
||
})
|
||
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，需要先用 Contents API 创建第一个文件
|
||
然后再使用 Git Data API 批量推送其余文件
|
||
"""
|
||
import time
|
||
|
||
try:
|
||
headers = self._get_headers()
|
||
base_url = f"{self.gitee_api_url}/repos/{owner}/{repo_name}"
|
||
|
||
logger.info(f"开始初始提交，共 {len(files)} 个文件...")
|
||
|
||
# 找出 README.md 作为初始文件（或者使用第一个文件）
|
||
init_file_path = None
|
||
init_file_content = None
|
||
|
||
if "README.md" in files:
|
||
init_file_path = "README.md"
|
||
init_file_content = files["README.md"]
|
||
else:
|
||
# 使用第一个文件
|
||
init_file_path = list(files.keys())[0]
|
||
init_file_content = files[init_file_path]
|
||
|
||
# 1. 使用 Contents API 创建第一个文件（初始化仓库）
|
||
init_url = f"{base_url}/contents/{init_file_path}"
|
||
init_data = {
|
||
"message": f"{commit_message} - {init_file_path}",
|
||
"content": base64.b64encode(init_file_content.encode()).decode(),
|
||
"branch": branch
|
||
}
|
||
|
||
logger.info(f"创建初始文件: {init_file_path}")
|
||
init_response = requests.post(init_url, json=init_data, headers=headers, timeout=30)
|
||
|
||
if init_response.status_code not in [200, 201]:
|
||
return {"success": False, "error": f"创建初始文件失败: {init_response.text[:300]}"}
|
||
|
||
logger.info(f"✓ 初始文件创建成功: {init_file_path}")
|
||
time.sleep(0.5)
|
||
|
||
# 2. 推送剩余文件
|
||
remaining_files = {k: v for k, v in files.items() if k != init_file_path}
|
||
|
||
if remaining_files:
|
||
logger.info(f"使用批量方式推送剩余 {len(remaining_files)} 个文件...")
|
||
|
||
# 现在仓库已初始化，使用 Git Data API 批量推送
|
||
batch_result = self._push_files_batch(owner, repo_name, remaining_files, commit_message, branch)
|
||
|
||
if not batch_result.get("success"):
|
||
# 批量失败，逐个推送
|
||
logger.warning(f"批量推送失败: {batch_result.get('error')}, 尝试逐个推送")
|
||
|
||
for file_path, content in remaining_files.items():
|
||
try:
|
||
file_url = f"{base_url}/contents/{file_path}"
|
||
file_data = {
|
||
"message": f"{commit_message} - {file_path}",
|
||
"content": base64.b64encode(content.encode()).decode(),
|
||
"branch": branch
|
||
}
|
||
|
||
file_response = requests.post(file_url, json=file_data, headers=headers, timeout=30)
|
||
|
||
if file_response.status_code in [200, 201]:
|
||
logger.info(f" ✓ {file_path}")
|
||
else:
|
||
logger.error(f" ✗ {file_path}: {file_response.text[:100]}")
|
||
|
||
time.sleep(0.3)
|
||
|
||
except Exception as e:
|
||
logger.error(f" ✗ {file_path}: {e}")
|
||
|
||
logger.info(f"✅ 初始提交完成")
|
||
return {"success": True}
|
||
|
||
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 运行状态
|
||
|
||
Args:
|
||
repo_name: 仓库名称
|
||
|
||
Returns:
|
||
Action 运行状态
|
||
"""
|
||
# Gitea Actions API
|
||
url = f"{self.gitee_api_url}/repos/{self.gitee_owner}/{repo_name}/actions/runs"
|
||
|
||
try:
|
||
response = requests.get(
|
||
url,
|
||
headers=self._get_headers(),
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
runs = data.get("workflow_runs", [])
|
||
|
||
if not runs:
|
||
return {
|
||
"success": True,
|
||
"status": "no_runs",
|
||
"message": "暂无 Action 运行记录"
|
||
}
|
||
|
||
# 获取最新的运行
|
||
latest_run = runs[0] if runs else None
|
||
|
||
if latest_run:
|
||
return {
|
||
"success": True,
|
||
"status": latest_run.get("status"),
|
||
"conclusion": latest_run.get("conclusion"),
|
||
"run_id": latest_run.get("id"),
|
||
"created_at": latest_run.get("created_at"),
|
||
"updated_at": latest_run.get("updated_at"),
|
||
"html_url": latest_run.get("html_url")
|
||
}
|
||
|
||
elif response.status_code == 404:
|
||
return {
|
||
"success": True,
|
||
"status": "not_configured",
|
||
"message": "Actions 未配置或仓库不存在"
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"error": response.text
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取 Action 状态失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def get_latest_workflow_run(self, repo_name: str, workflow_name: str = None) -> Dict[str, Any]:
|
||
"""
|
||
获取最新的 Workflow 运行信息
|
||
|
||
Args:
|
||
repo_name: 仓库名称
|
||
workflow_name: 工作流名称（可选）
|
||
|
||
Returns:
|
||
Workflow 运行信息
|
||
"""
|
||
url = f"{self.gitee_api_url}/repos/{self.gitee_owner}/{repo_name}/actions/runs"
|
||
|
||
try:
|
||
response = requests.get(
|
||
url,
|
||
headers=self._get_headers(),
|
||
params={"per_page": 10},
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
runs = data.get("workflow_runs", [])
|
||
|
||
# 如果指定了工作流名称，过滤
|
||
if workflow_name and runs:
|
||
runs = [r for r in runs if r.get("name") == workflow_name]
|
||
|
||
if runs:
|
||
latest = runs[0]
|
||
return {
|
||
"success": True,
|
||
"run": {
|
||
"id": latest.get("id"),
|
||
"name": latest.get("name"),
|
||
"status": latest.get("status"),
|
||
"conclusion": latest.get("conclusion"),
|
||
"created_at": latest.get("created_at"),
|
||
"updated_at": latest.get("updated_at"),
|
||
"run_number": latest.get("run_number")
|
||
}
|
||
}
|
||
else:
|
||
return {
|
||
"success": True,
|
||
"run": None,
|
||
"message": "暂无运行记录"
|
||
}
|
||
else:
|
||
return {"success": False, "error": response.text}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def delete_repository(self, repo_name: str) -> Dict[str, Any]:
|
||
"""删除仓库"""
|
||
url = f"{self.gitee_api_url}/repos/{self.gitee_owner}/{repo_name}"
|
||
|
||
try:
|
||
response = requests.delete(
|
||
url,
|
||
headers=self._get_headers(),
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code == 204:
|
||
logger.info(f"✅ 仓库删除成功: {repo_name}")
|
||
return {"success": True}
|
||
else:
|
||
return {"success": False, "error": response.text}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def set_repo_secret(
|
||
self,
|
||
repo_name: str,
|
||
secret_name: str,
|
||
secret_value: str,
|
||
owner: str = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
设置仓库 Action Secret
|
||
|
||
Args:
|
||
repo_name: 仓库名称
|
||
secret_name: Secret 名称
|
||
secret_value: Secret 值
|
||
owner: 仓库所有者
|
||
|
||
Returns:
|
||
设置结果
|
||
"""
|
||
repo_owner = owner or self.gitee_username or self.gitee_owner
|
||
url = f"{self.gitee_api_url}/repos/{repo_owner}/{repo_name}/actions/secrets/{secret_name}"
|
||
|
||
try:
|
||
response = requests.put(
|
||
url,
|
||
json={"data": secret_value},
|
||
headers=self._get_headers(),
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code in [200, 201, 204]:
|
||
logger.info(f"✅ Secret 设置成功: {secret_name}")
|
||
return {"success": True, "secret_name": secret_name}
|
||
else:
|
||
return {"success": False, "error": response.text}
|
||
|
||
except Exception as e:
|
||
logger.error(f"设置 Secret 失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def set_repo_secrets(
|
||
self,
|
||
repo_name: str,
|
||
secrets: Dict[str, str],
|
||
owner: str = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
批量设置仓库 Action Secrets
|
||
|
||
Args:
|
||
repo_name: 仓库名称
|
||
secrets: Secret 字典 {name: value}
|
||
owner: 仓库所有者
|
||
|
||
Returns:
|
||
设置结果
|
||
"""
|
||
import time
|
||
|
||
results = {"success": True, "secrets": []}
|
||
|
||
for name, value in secrets.items():
|
||
result = self.set_repo_secret(repo_name, name, value, owner)
|
||
results["secrets"].append({
|
||
"name": name,
|
||
"status": "success" if result["success"] else "failed",
|
||
"error": result.get("error")
|
||
})
|
||
|
||
if not result["success"]:
|
||
results["success"] = False
|
||
|
||
# 避免 API 限流
|
||
time.sleep(0.3)
|
||
|
||
return results
|
||
|
||
|
||
# 全局实例
|
||
gitee_manager = GiteeManager()
|