Spaces:
Paused
Paused
| import os | |
| import subprocess | |
| import venv | |
| import shutil | |
| from git import Repo | |
| import threading | |
| import time | |
| class Deployer: | |
| def __init__(self, base_path="deployments"): | |
| self.base_path = os.path.abspath(base_path) | |
| if not os.path.exists(self.base_path): | |
| os.makedirs(self.base_path) | |
| self.active_processes = {} | |
| def deploy(self, repo_url, project_name): | |
| project_dir = os.path.join(self.base_path, project_name) | |
| # 1. Clone or Update | |
| if repo_url: | |
| if os.path.exists(project_dir): | |
| try: | |
| repo = Repo(project_dir) | |
| repo.remotes.origin.pull() | |
| except: | |
| shutil.rmtree(project_dir) | |
| Repo.clone_from(repo_url, project_dir) | |
| else: | |
| Repo.clone_from(repo_url, project_dir) | |
| else: | |
| # Local file handling - ensure directory exists | |
| os.makedirs(project_dir, exist_ok=True) | |
| # 2. Venv Management | |
| venv_dir = os.path.join(project_dir, "venv") | |
| if not os.path.exists(venv_dir): | |
| venv.create(venv_dir, with_pip=True) | |
| # 3. Automatic Library Installation | |
| pip_path = os.path.join(venv_dir, "Scripts", "pip") if os.name == "nt" else os.path.join(venv_dir, "bin", "pip") | |
| req_path = os.path.join(project_dir, "requirements.txt") | |
| if os.path.exists(req_path): | |
| # Check for changes in requirements before installing? For now just install. | |
| subprocess.run([pip_path, "install", "--upgrade", "pip"], capture_output=True) | |
| subprocess.run([pip_path, "install", "-r", req_path], check=True, capture_output=True) | |
| return project_dir | |
| def run_script(self, project_name, script_name=None, log_callback=None): | |
| project_dir = os.path.join(self.base_path, project_name) | |
| venv_dir = os.path.join(project_dir, "venv") | |
| # Ensure venv and script exist | |
| python_path = os.path.join(venv_dir, "Scripts", "python") if os.name == "nt" else os.path.join(venv_dir, "bin", "python") | |
| if script_name: | |
| script_path = os.path.join(project_dir, script_name) | |
| # If script name is provided but doesn't exist in project dir, check deployments root | |
| if not os.path.exists(script_path): | |
| if os.path.exists(os.path.join(self.base_path, script_name)): | |
| script_path = os.path.join(self.base_path, script_name) | |
| else: | |
| # Auto-detect entry point | |
| script_path = None | |
| for alt in ["main.py", "app.py", "run.py", "index.py"]: | |
| if os.path.exists(os.path.join(project_dir, alt)): | |
| script_path = os.path.join(project_dir, alt) | |
| break | |
| if not script_path: | |
| # Last resort: check if project_name itself is a script | |
| if project_name.endswith('.py') and os.path.exists(os.path.join(self.base_path, project_name)): | |
| script_path = os.path.join(self.base_path, project_name) | |
| else: | |
| raise Exception(f"COULD NOT DETECT ENTRY POINT FOR {project_name}") | |
| if not os.path.exists(python_path): | |
| raise Exception("VIRTUAL ENVIRONMENT CORRUPTED OR MISSING") | |
| # Dynamic Dependency Discovery | |
| self._install_dynamic_deps(script_path, os.path.join(venv_dir, "Scripts", "pip") if os.name == "nt" else os.path.join(venv_dir, "bin", "pip"), log_callback) | |
| # Kill existing process if any | |
| if project_name in self.active_processes: | |
| self.stop_project(project_name) | |
| process = subprocess.Popen( | |
| [python_path, script_path], | |
| stdin=subprocess.PIPE, | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.STDOUT, | |
| text=True, | |
| bufsize=1, | |
| cwd=project_dir | |
| ) | |
| self.active_processes[project_name] = process | |
| if log_callback: | |
| threading.Thread(target=self._stream_logs, args=(project_name, process, log_callback), daemon=True).start() | |
| return process.pid | |
| def _install_dynamic_deps(self, script_path, pip_path, log_callback): | |
| import re | |
| # Mapping of import names to package names | |
| PKG_MAP = { | |
| "PIL": "Pillow", | |
| "telebot": "pyTelegramBotAPI", | |
| "cv2": "opencv-python", | |
| "sklearn": "scikit-learn", | |
| "sqlite3": None, # Standard lib | |
| "telegram": "python-telegram-bot" | |
| } | |
| try: | |
| with open(script_path, "r", encoding="utf-8") as f: | |
| content = f.read() | |
| imports = re.findall(r"^(?:import|from)\s+([a-zA-Z0-9_]+)", content, re.MULTILINE) | |
| unique_imports = list(set(imports)) | |
| std_libs = {"os", "sys", "time", "asyncio", "re", "json", "math", "random", "shutil", "threading", "datetime", "logging", "string", "tempfile", "pathlib"} | |
| to_install = [] | |
| for i in unique_imports: | |
| if i in std_libs: continue | |
| pkg = PKG_MAP.get(i, i) | |
| if pkg: to_install.append(pkg) | |
| if to_install: | |
| if log_callback: log_callback("INSTALLER", f"Optimizing dependencies: {', '.join(to_install)}") | |
| # Bulk install for high speed | |
| try: | |
| subprocess.run([pip_path, "install", "--quiet"] + to_install, capture_output=True) | |
| if log_callback: log_callback("SYSTEM", "All dependencies synchronized.") | |
| except Exception as e: | |
| if log_callback: log_callback("ERROR", f"Installation failed: {e}") | |
| except: pass | |
| def stop_project(self, project_name): | |
| if project_name in self.active_processes: | |
| proc = self.active_processes[project_name] | |
| proc.terminate() | |
| try: | |
| proc.wait(timeout=5) | |
| except: | |
| proc.kill() | |
| del self.active_processes[project_name] | |
| def send_input(self, project_name, text): | |
| if project_name in self.active_processes: | |
| proc = self.active_processes[project_name] | |
| if proc.stdin: | |
| proc.stdin.write(text + "\n") | |
| proc.stdin.flush() | |
| def _stream_logs(self, project_name, process, callback): | |
| log_file_path = os.path.join(self.base_path, project_name, "execution.log") | |
| with open(log_file_path, "a", encoding="utf-8") as log_file: | |
| for line in iter(process.stdout.readline, ''): | |
| if line: | |
| timestamp = time.strftime("[%H:%M:%S]") | |
| full_line = f"{timestamp} {line.strip()}" | |
| log_file.write(full_line + "\n") | |
| log_file.flush() | |
| callback(project_name, full_line) | |
| process.stdout.close() | |
| process.wait() | |
| if project_name in self.active_processes: | |
| del self.active_processes[project_name] | |
| def kill_all(self): | |
| for name in list(self.active_processes.keys()): | |
| self.stop_project(name) | |
| self.active_processes = {} | |