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 = {}