Phase 1: ProjectEngine + PyEnvManager + project management WebUI
- New: services/project_engine.py (async subprocess manager) - New: services/pyenv_manager.py (Python version + venv + git clone) - New: services/web_panel/routes/projects.py (5 REST endpoints) - New: static/web_panel/pages/projects.html (WebUI) - Tests: 28/28 passing, API verified (GET /api/projects) - Docs: Phase1_Progress.md (local)
This commit is contained in:
@@ -0,0 +1,130 @@
|
||||
#!/usr/bin/env python3
|
||||
"""SenSu PyEnvManager — Python 版本管理 + venv + 依赖安装"""
|
||||
import os, sys, subprocess, logging, venv, 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:
|
||||
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
|
||||
Reference in New Issue
Block a user