headquarter/apps/api/src/services/docker.py

"""Docker service for managing tool instances."""

import os
import subprocess
from pathlib import Path
from typing import Any


def render_compose_template(template: str, variables: dict[str, Any]) -> str:
    """Render a Docker Compose template with variable substitution.

    Args:
        template: The compose template string
        variables: Dictionary of variable names to values

    Returns:
        Rendered compose file content
    """
    result = template
    for key, value in variables.items():
        placeholder = f"{{{{{key}}}}}"
        result = result.replace(placeholder, str(value))
    return result


def ensure_instance_directory(instance_id: str, base_path: str | None = None) -> str:
    """Create and return the instance directory path.

    Args:
        instance_id: Unique instance identifier
        base_path: Base directory for all instances (defaults to Settings.instance_base_path)

    Returns:
        Absolute path to instance directory
    """
    if base_path is None:
        from src.config import Settings
        base_path = Settings().instance_base_path
    instance_dir = Path(base_path) / instance_id
    instance_dir.mkdir(parents=True, exist_ok=True)
    return str(instance_dir.absolute())


def write_compose_file(instance_dir: str, content: str) -> str:
    """Write the rendered compose file to the instance directory.

    Args:
        instance_dir: Path to instance directory
        content: Rendered compose content

    Returns:
        Path to the compose file
    """
    compose_path = Path(instance_dir) / "docker-compose.yml"
    compose_path.write_text(content)
    return str(compose_path)


def execute_compose_command(
    compose_path: str, action: str, timeout: int = 60
) -> tuple[int, str, str]:
    """Execute a docker compose command.

    Args:
        compose_path: Path to docker-compose.yml
        action: The compose action (up, down, start, stop, restart)
        timeout: Command timeout in seconds

    Returns:
        Tuple of (returncode, stdout, stderr)
    """
    instance_dir = Path(compose_path).parent

    cmd = ["docker", "compose", "-f", compose_path]

    if action == "up":
        cmd.extend(["up", "-d"])
    elif action == "down":
        cmd.extend(["down", "-v"])
    elif action in ("start", "stop", "restart"):
        cmd.append(action)
    else:
        raise ValueError(f"Unknown compose action: {action}")

    result = subprocess.run(
        cmd,
        cwd=str(instance_dir),
        capture_output=True,
        text=True,
        timeout=timeout,
    )

    return result.returncode, result.stdout, result.stderr


def get_container_id(instance_name: str) -> str | None:
    """Get the container ID for a compose service.

    Args:
        instance_name: The service name in compose

    Returns:
        Container ID or None if not found
    """
    result = subprocess.run(
        ["docker", "ps", "-q", "--filter", f"name={instance_name}"],
        capture_output=True,
        text=True,
    )

    if result.returncode == 0 and result.stdout.strip():
        return result.stdout.strip().split("\n")[0]
    return None


def get_container_status(container_id: str) -> str:
    """Get the status of a Docker container.

    Args:
        container_id: Docker container ID

    Returns:
        Container status string (running, exited, etc.)
    """
    result = subprocess.run(
        ["docker", "inspect", "-f", "{{.State.Status}}", container_id],
        capture_output=True,
        text=True,
    )

    if result.returncode == 0:
        return result.stdout.strip()
    return "unknown"


def get_container_logs(container_id: str, tail: int = 100) -> str:
    """Get the logs of a Docker container.

    Args:
        container_id: Docker container ID
        tail: Number of lines to return

    Returns:
        Container logs
    """
    result = subprocess.run(
        ["docker", "logs", "--tail", str(tail), container_id],
        capture_output=True,
        text=True,
    )

    if result.returncode == 0:
        return result.stdout
    return f"Failed to get logs: {result.stderr}"


def find_free_port(start: int = 10000, end: int = 20000) -> int:
    """Find a free TCP port in the given range.

    Args:
        start: Start of port range
        end: End of port range

    Returns:
        Free port number
    """
    import socket

    for port in range(start, end):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            if s.connect_ex(("localhost", port)) != 0:
                return port

    raise RuntimeError(f"No free port found in range {start}-{end}")