SIDEL_ScriptsManager/app/services/conda_service.py

import subprocess
import json
import os
import platform
from pathlib import Path
from typing import List, Dict, Optional
from app.models import CondaEnvironment
from app.config.database import db


class CondaService:
    """Service for managing conda environments."""

    def __init__(self):
        self.conda_executable = self.find_conda_executable()
        self.system = platform.system().lower()

    def find_conda_executable(self) -> Optional[str]:
        """Find conda executable on Windows/Linux."""
        possible_paths = [
            "conda",
            "/opt/conda/bin/conda",
            "/usr/local/bin/conda",
            os.path.expanduser("~/miniconda3/bin/conda"),
            os.path.expanduser("~/anaconda3/bin/conda"),
        ]

        # Windows specific paths
        if platform.system().lower() == "windows":
            possible_paths.extend(
                [
                    r"C:\ProgramData\Miniconda3\Scripts\conda.exe",
                    r"C:\ProgramData\Anaconda3\Scripts\conda.exe",
                    os.path.expanduser(r"~\Miniconda3\Scripts\conda.exe"),
                    os.path.expanduser(r"~\Anaconda3\Scripts\conda.exe"),
                    r"C:\tools\miniconda3\Scripts\conda.exe",
                    r"C:\Miniconda3\Scripts\conda.exe",
                    r"C:\Anaconda3\Scripts\conda.exe",
                ]
            )

        for path in possible_paths:
            try:
                result = subprocess.run(
                    [path, "--version"], capture_output=True, text=True, timeout=10
                )
                if result.returncode == 0:
                    return path
            except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
                continue

        return None

    def is_available(self) -> bool:
        """Check if conda is available."""
        return self.conda_executable is not None

    def list_environments(self) -> List[Dict]:
        """List all available conda environments."""
        if not self.conda_executable:
            return []

        try:
            result = subprocess.run(
                [self.conda_executable, "env", "list", "--json"],
                capture_output=True,
                text=True,
                timeout=30,
            )

            if result.returncode == 0:
                env_data = json.loads(result.stdout)
                environments = []

                for env_path in env_data.get("envs", []):
                    env_name = Path(env_path).name
                    if env_path == env_data.get("conda_default_env", ""):
                        env_name = "base"

                    python_version = self.get_python_version(env_path)

                    environments.append(
                        {
                            "name": env_name,
                            "path": env_path,
                            "python_version": python_version,
                        }
                    )

                return environments

        except Exception as e:
            print(f"Error listing conda environments: {e}")
            return []

    def get_python_version(self, env_path: str) -> Optional[str]:
        """Get Python version for an environment."""
        try:
            python_exe = os.path.join(
                env_path, "python.exe" if self.system == "windows" else "bin/python"
            )

            if os.path.exists(python_exe):
                result = subprocess.run(
                    [python_exe, "--version"],
                    capture_output=True,
                    text=True,
                    timeout=10,
                )

                if result.returncode == 0:
                    version_output = result.stdout.strip()
                    if version_output.startswith("Python "):
                        return version_output.split()[1]

        except Exception:
            pass

        return None

    def refresh_environments(self) -> List[CondaEnvironment]:
        """Refresh conda environments in database."""
        environments = self.list_environments()
        db_environments = []

        for env_data in environments:
            # Check if environment already exists
            existing = CondaEnvironment.query.filter_by(name=env_data["name"]).first()

            if existing:
                # Update existing environment
                existing.path = env_data["path"]
                existing.python_version = env_data["python_version"]
                existing.is_available = True
                existing.last_verified = db.func.now()
                db_environments.append(existing)
            else:
                # Create new environment
                new_env = CondaEnvironment(
                    name=env_data["name"],
                    path=env_data["path"],
                    python_version=env_data["python_version"],
                    is_available=True,
                )
                db.session.add(new_env)
                db_environments.append(new_env)

        # Mark missing environments as unavailable
        all_db_envs = CondaEnvironment.query.all()
        current_names = [env["name"] for env in environments]

        for db_env in all_db_envs:
            if db_env.name not in current_names:
                db_env.is_available = False

        db.session.commit()
        return db_environments

    def get_environment_info(self, env_name: str) -> Optional[Dict]:
        """Get detailed information about an environment."""
        env = CondaEnvironment.query.filter_by(name=env_name).first()
        if env and env.is_available:
            return env.to_dict()
        return None

    def build_conda_command(self, env_name: str, command: List[str]) -> List[str]:
        """Build conda command for executing in specific environment."""
        if not self.conda_executable:
            raise RuntimeError("Conda executable not found")

        # Check if we're already in the target environment
        current_env = os.environ.get("CONDA_DEFAULT_ENV", "")
        print(f"[CONDA_SERVICE] Current environment: {current_env}")
        print(f"[CONDA_SERVICE] Target environment: {env_name}")

        if current_env == env_name:
            print(
                f"[CONDA_SERVICE] Already in target environment {env_name}, "
                "using direct command"
            )
            return command

        print("[CONDA_SERVICE] Different environment, using conda run")

        if self.system == "windows":
            # Windows conda activation
            return [
                self.conda_executable,
                "run",
                "-n",
                env_name,
                "--no-capture-output",
            ] + command
        else:
            # Linux conda activation
            return [self.conda_executable, "run", "-n", env_name] + command

    def validate_environment(self, env_name: str) -> bool:
        """Validate that an environment exists and is functional."""
        try:
            command = self.build_conda_command(env_name, ["python", "--version"])
            result = subprocess.run(command, capture_output=True, text=True, timeout=15)
            return result.returncode == 0
        except Exception:
            return False