qinglong
132 lines
5.0 KiB
Python
132 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""SenSu PyEnvManager — Python 版本管理 + venv + 依赖安装"""
|
|
import os, sys, subprocess, logging, shutil
|
|
from pathlib import Path
|
|
from typing import Optional, List
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class PyEnvManager:
|
|
def __init__(self, workspace_dir: str = None):
|
|
self.workspace = Path(workspace_dir or os.getcwd())
|
|
self.venvs_dir = self.workspace / "venvs"
|
|
self.venvs_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def detect_versions(self) -> List[str]:
|
|
versions = set()
|
|
# Current Python
|
|
v = f"{sys.version_info.major}.{sys.version_info.minor}"
|
|
versions.add(v)
|
|
|
|
# Check Termux pkg
|
|
try:
|
|
result = subprocess.run(["pkg", "list-installed"], capture_output=True, text=True, timeout=10)
|
|
for line in result.stdout.split("\n"):
|
|
if line.startswith("python-3.") or line.startswith("python3-"):
|
|
versions.add(line.split("/")[0].replace("python-", "").replace("python3-", ""))
|
|
except:
|
|
pass
|
|
|
|
# Check pyenv
|
|
pyenv = shutil.which("pyenv")
|
|
if pyenv:
|
|
try:
|
|
result = subprocess.run([pyenv, "versions", "--bare"], capture_output=True, text=True, timeout=10)
|
|
for line in result.stdout.split("\n"):
|
|
line = line.strip()
|
|
if line and line[0].isdigit():
|
|
versions.add(line.split("/")[0])
|
|
except:
|
|
pass
|
|
|
|
return sorted(versions)
|
|
|
|
def ensure_version(self, version: str) -> Optional[str]:
|
|
"""确保指定 Python 版本可用，返回解释器路径"""
|
|
available = self.detect_versions()
|
|
if version in available:
|
|
return self._find_python(version)
|
|
|
|
# Try to install via Termux
|
|
if shutil.which("pkg"):
|
|
pkg_name = f"python-{version}"
|
|
logger.info(f"尝试安装 {pkg_name} ...")
|
|
try:
|
|
subprocess.run(["pkg", "install", "-y", pkg_name], check=True, timeout=120)
|
|
return self._find_python(version)
|
|
except:
|
|
pass
|
|
|
|
logger.warning(f"无法获取 Python {version}，使用当前版本")
|
|
return sys.executable
|
|
|
|
def _find_python(self, version: str) -> Optional[str]:
|
|
for name in [f"python{version}", f"python{version[:3]}", "python3"]:
|
|
path = shutil.which(name)
|
|
if path: return path
|
|
return sys.executable
|
|
|
|
def create_venv(self, name: str, python_version: str = None) -> Optional[Path]:
|
|
venv_path = self.venvs_dir / name
|
|
if venv_path.exists():
|
|
logger.info(f"venv 已存在: {venv_path}")
|
|
return venv_path
|
|
|
|
python_exe = self.ensure_version(python_version) if python_version else sys.executable
|
|
logger.info(f"创建 venv: {venv_path} (Python {python_version or 'default'})")
|
|
|
|
try:
|
|
import venv as _venv
|
|
_venv.create(str(venv_path), with_pip=True, clear=True)
|
|
# Install/upgrade pip
|
|
pip = str(venv_path / "bin" / "pip")
|
|
subprocess.run([pip, "install", "--upgrade", "pip"], capture_output=True, timeout=60)
|
|
return venv_path
|
|
except Exception as e:
|
|
logger.error(f"创建 venv 失败: {e}")
|
|
# Fallback: use virtualenv
|
|
try:
|
|
subprocess.run([sys.executable, "-m", "virtualenv", str(venv_path)], check=True, timeout=120)
|
|
return venv_path
|
|
except:
|
|
return None
|
|
|
|
def install_deps(self, venv_path: Path, requirements: List[str]) -> bool:
|
|
pip = str(venv_path / "bin" / "pip")
|
|
for req_file in requirements:
|
|
req_path = Path(req_file)
|
|
if not req_path.is_absolute():
|
|
# Relative to workspace
|
|
pass
|
|
if Path(req_file).exists():
|
|
logger.info(f"安装依赖: {req_file}")
|
|
try:
|
|
subprocess.run([pip, "install", "-r", req_file], check=True, timeout=300)
|
|
except subprocess.CalledProcessError as e:
|
|
logger.warning(f"依赖安装部分失败: {e}")
|
|
return False
|
|
return True
|
|
|
|
def clone_git(self, url: str, target_dir: Path, branch: str = None) -> bool:
|
|
if target_dir.exists():
|
|
logger.info(f"目录已存在: {target_dir}")
|
|
# Try git pull instead
|
|
try:
|
|
subprocess.run(["git", "-C", str(target_dir), "pull"], check=True, timeout=60)
|
|
return True
|
|
except:
|
|
pass
|
|
|
|
cmd = ["git", "clone"]
|
|
if branch:
|
|
cmd += ["-b", branch]
|
|
cmd += [url, str(target_dir)]
|
|
|
|
try:
|
|
subprocess.run(cmd, check=True, timeout=300)
|
|
logger.info(f"Git clone 完成: {url} → {target_dir}")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f"Git clone 失败: {e}")
|
|
return False
|