app/utils/network_utils.py

"""
网络工具
处理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()