zhanggangyong
090f4a04b3
- 更新 AZ_CLIENT_ID 为新的 SP - 更新 AZ_CLIENT_SECRET - 更新 AZ_SUBSCRIPTION_ID 为新订阅
419 lines
14 KiB
Python
419 lines
14 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
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
创建 Gitee 仓库
|
||
|
||
Args:
|
||
repo_name: 仓库名称
|
||
description: 仓库描述
|
||
private: 是否私有
|
||
|
||
Returns:
|
||
仓库信息
|
||
"""
|
||
url = f"{self.gitee_api_url}/user/repos"
|
||
|
||
data = {
|
||
"name": repo_name,
|
||
"description": description,
|
||
"private": private,
|
||
"auto_init": True, # 自动初始化
|
||
"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}
|
||
|
||
except Exception as e:
|
||
logger.error(f"创建仓库异常: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
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]:
|
||
"""
|
||
推送文件到仓库
|
||
|
||
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}
|
||
|
||
for file_path, content in files.items():
|
||
try:
|
||
# 使用 Contents API 创建/更新文件
|
||
url = f"{self.gitee_api_url}/repos/{repo_owner}/{repo_name}/contents/{file_path}"
|
||
|
||
# 检查文件是否存在
|
||
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,
|
||
"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.5)
|
||
|
||
if response.status_code in [200, 201]:
|
||
results["files"].append({
|
||
"path": file_path,
|
||
"status": "success"
|
||
})
|
||
logger.info(f"✅ 文件推送成功: {file_path}")
|
||
else:
|
||
results["files"].append({
|
||
"path": file_path,
|
||
"status": "failed",
|
||
"error": response.text
|
||
})
|
||
results["success"] = False
|
||
|
||
except Exception as e:
|
||
results["files"].append({
|
||
"path": file_path,
|
||
"status": "failed",
|
||
"error": str(e)
|
||
})
|
||
results["success"] = False
|
||
|
||
return results
|
||
|
||
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()
|