File size: 4,533 Bytes
b6db694 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """
网络工具
处理Git克隆、文件下载等
"""
import os
import tempfile
import logging
from pathlib import Path
from typing import Optional, Tuple, Callable
from urllib.parse import urlparse
import subprocess
logger = logging.getLogger(__name__)
class NetworkUtils:
"""网络工具类"""
@staticmethod
def clone_git_repository(repo_url: str, target_dir: Path, branch: str = "main") -> Tuple[bool, str]:
"""克隆Git仓库"""
try:
# 验证Git URL
if not NetworkUtils._is_valid_git_url(repo_url):
return False, "无效的Git仓库URL"
# 确保目标目录不存在或为空
if target_dir.exists():
# 如果目录存在且非空，先删除
if any(target_dir.iterdir()):
import shutil
shutil.rmtree(target_dir)
# 执行Git克隆命令
cmd = ["git", "clone", "--branch", branch, "--depth", "1", repo_url, str(target_dir)]
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300 # 5分钟超时
)
if result.returncode == 0:
logger.info(f"成功克隆Git仓库: {repo_url} -> {target_dir}")
return True, "克隆成功"
else:
error_msg = result.stderr.strip()
logger.error(f"克隆Git仓库失败: {error_msg}")
return False, f"克隆失败: {error_msg}"
except subprocess.TimeoutExpired:
logger.error("克隆Git仓库超时")
return False, "克隆超时，请检查网络连接"
except Exception as e:
logger.error(f"克隆Git仓库异常: {e}")
return False, f"克隆异常: {str(e)}"
@staticmethod
def _is_valid_git_url(url: str) -> bool:
"""验证Git URL格式"""
try:
parsed = urlparse(url)
if parsed.scheme in ('http', 'https', 'git'):
return True
# 支持SSH格式的URL
if url.startswith('git@') and '.git' in url:
return True
return False
except Exception:
return False
@staticmethod
def download_file(url: str, target_path: Path) -> Tuple[bool, str]:
"""下载文件"""
try:
import requests
response = requests.get(url, stream=True, timeout=30)
response.raise_for_status()
with open(target_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
logger.info(f"成功下载文件: {url} -> {target_path}")
return True, "下载成功"
except requests.RequestException as e:
logger.error(f"下载文件失败: {e}")
return False, f"下载失败: {str(e)}"
except Exception as e:
logger.error(f"下载文件异常: {e}")
return False, f"下载异常: {str(e)}"
@staticmethod
def get_git_info(repo_url: str) -> Tuple[bool, dict]:
"""获取Git仓库信息"""
try:
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# 克隆仓库（浅克隆以加快速度）
success, message = NetworkUtils.clone_git_repository(repo_url, temp_path)
if not success:
return False, {"error": message}
# 检查插件结构
from app.utils.file_utils import file_utils
valid, msg, metadata = file_utils.validate_plugin_structure(temp_path)
if valid:
return True, {
"name": metadata.get("name"),
"version": metadata.get("version"),
"description": metadata.get("description"),
"author": metadata.get("author")
}
else:
return False, {"error": msg}
except Exception as e:
logger.error(f"获取Git信息失败: {e}")
return False, {"error": str(e)}
# 创建全局实例
network_utils = NetworkUtils() |