docker_compose.py

#!/usr/bin/env python
"""Run Docker Compose with port mappings rendered from JSON settings.

Docker Compose cannot read ``data/user/settings/system.json`` directly for
host port interpolation. This wrapper renders a tiny compose env file from the
JSON settings and then invokes ``docker compose --env-file``. It intentionally
does not read or migrate the project-root ``.env`` file.
"""

from __future__ import annotations

import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
from typing import Any

PROJECT_ROOT = Path(__file__).resolve().parent.parent
SETTINGS_DIR = PROJECT_ROOT / "data" / "user" / "settings"
DOCKER_ENV_PATH = SETTINGS_DIR / "docker.env"

DEFAULT_BACKEND_PORT = 8001
DEFAULT_FRONTEND_PORT = 3782
DEFAULT_POCKETBASE_PORT = 8090
DEFAULT_WORKSPACE_HOST = Path("data/user/workspace")


def _read_json_object(path: Path) -> dict[str, Any]:
    try:
        loaded = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return loaded if isinstance(loaded, dict) else {}


def _coerce_port(value: Any, default: int) -> int:
    try:
        port = int(str(value).strip())
    except (TypeError, ValueError):
        return default
    return port if 1 <= port <= 65535 else default


def render_docker_env(
    settings_dir: Path = SETTINGS_DIR,
    output_path: Path = DOCKER_ENV_PATH,
) -> dict[str, str]:
    """Render compose interpolation vars from JSON settings only."""
    system = _read_json_object(settings_dir / "system.json")
    integrations = _read_json_object(settings_dir / "integrations.json")
    values = {
        "DEEPTUTOR_DOCKER_BACKEND_PORT": str(
            _coerce_port(system.get("backend_port"), DEFAULT_BACKEND_PORT)
        ),
        "DEEPTUTOR_DOCKER_FRONTEND_PORT": str(
            _coerce_port(system.get("frontend_port"), DEFAULT_FRONTEND_PORT)
        ),
        "DEEPTUTOR_DOCKER_POCKETBASE_PORT": str(
            _coerce_port(integrations.get("pocketbase_port"), DEFAULT_POCKETBASE_PORT)
        ),
    }
    output_path.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        "# Auto-generated by scripts/docker_compose.py from data/user/settings/*.json.",
        "# Do not edit manually; update system.json/integrations.json instead.",
    ]
    lines.extend(f"{key}={value}" for key, value in values.items())
    output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return values


def _compose_command(args: list[str]) -> list[str]:
    docker = shutil.which("docker")
    if not docker:
        raise SystemExit("docker was not found on PATH")
    return [docker, "compose", "--env-file", str(DOCKER_ENV_PATH), *args]


def ensure_workspace_host(raw_path: str | None = None) -> Path:
    """Create the host workspace and nested writable outputs bind.

    Compose can create a missing bind source as root, which leaves the
    unprivileged app/runner unable to write it. Preparing both directories as
    the invoking user also makes the effective path explicit before startup.
    """
    raw = str(raw_path or os.environ.get("DEEPTUTOR_WORKSPACE_HOST") or "").strip()
    candidate = Path(raw).expanduser() if raw else DEFAULT_WORKSPACE_HOST
    if not candidate.is_absolute():
        candidate = PROJECT_ROOT / candidate
    candidate = candidate.resolve()
    candidate.mkdir(parents=True, exist_ok=True)
    (candidate / "outputs").mkdir(parents=True, exist_ok=True)
    return candidate


def main(argv: list[str] | None = None) -> int:
    args = list(argv if argv is not None else sys.argv[1:])
    if not args:
        args = ["up", "-d"]

    values = render_docker_env()
    workspace_host = ensure_workspace_host()
    print(
        "Docker settings: "
        f"backend={values['DEEPTUTOR_DOCKER_BACKEND_PORT']} "
        f"frontend={values['DEEPTUTOR_DOCKER_FRONTEND_PORT']} "
        f"pocketbase={values['DEEPTUTOR_DOCKER_POCKETBASE_PORT']} "
        f"workspace={workspace_host}",
        file=sys.stderr,
    )

    env = os.environ.copy()
    # Keep Docker execution detached from host process overrides.
    for key in (
        "BACKEND_PORT",
        "FRONTEND_PORT",
        "POCKETBASE_PORT",
        "AUTH_ENABLED",
        "POCKETBASE_URL",
        "NEXT_PUBLIC_API_BASE",
        "NEXT_PUBLIC_API_BASE_EXTERNAL",
    ):
        env.pop(key, None)

    result = subprocess.run(_compose_command(args), cwd=str(PROJECT_ROOT), env=env, check=False)
    return int(result.returncode)


if __name__ == "__main__":
    raise SystemExit(main())