fotohost/app/deploy_utils.py

import os
import re
import subprocess

REF_PATTERN = re.compile(r"^[a-zA-Z0-9._/-]+$")


def is_deploy_enabled():
    return os.getenv("ALLOW_GIT_DEPLOY", "false").lower() in ("1", "true", "yes")


def get_repo_path():
    return os.getenv("GIT_REPO_PATH", "/repo")


def get_git_remote():
    return os.getenv("GIT_REMOTE_URL", "").strip()


def _repo_ready():
    repo = get_repo_path()
    return os.path.isdir(repo) and os.path.isdir(os.path.join(repo, ".git"))


def run_git(args, timeout=120):
    if not _repo_ready():
        return False, f"Git-репозиторий не найден: {get_repo_path()}"

    repo = get_repo_path()
    result = subprocess.run(
        ["git", "-c", f"safe.directory={repo}", "-C", repo] + args,
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    if result.returncode != 0:
        return False, (result.stderr or result.stdout or "Git error").strip()
    return True, result.stdout.strip()


def run_ls_remote(extra_args=None, timeout=60):
    remote = get_git_remote()
    if not remote:
        return False, "GIT_REMOTE_URL не задан", []

    cmd = ["git", "ls-remote", remote]
    if extra_args:
        cmd.extend(extra_args)

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        return False, (result.stderr or result.stdout or "ls-remote error").strip(), []
    return True, "", result.stdout.splitlines()


def fetch_remote():
    remote = get_git_remote()
    if not remote:
        return run_git(["fetch", "--all", "--tags", "--prune"], timeout=180)

    # Fetch from URL directly — do not write remote.origin.url to .git/config
    return run_git(
        [
            "fetch",
            "--tags",
            "--prune",
            remote,
            "+refs/heads/*:refs/remotes/origin/*",
            "+refs/tags/*:refs/tags/*",
        ],
        timeout=180,
    )


def list_tags():
    ok, err, lines = run_ls_remote(["--tags"])
    if not ok:
        return [], err

    tags = []
    for line in lines:
        parts = line.split()
        if len(parts) < 2:
            continue
        ref = parts[1]
        if not ref.startswith("refs/tags/"):
            continue
        tag = ref.removeprefix("refs/tags/")
        if tag.endswith("^{}"):
            continue
        tags.append(tag)

    tags = sorted(set(tags), reverse=True)
    return tags, None


def list_branches():
    ok, err, lines = run_ls_remote(["--heads"])
    if not ok:
        return [], err

    branches = []
    for line in lines:
        parts = line.split()
        if len(parts) < 2:
            continue
        ref = parts[1]
        if ref.startswith("refs/heads/"):
            branches.append(ref.removeprefix("refs/heads/"))

    return sorted(set(branches)), None


def get_current_version():
    if not _repo_ready():
        return None, "Репозиторий недоступен"

    ok, tag = run_git(["describe", "--tags", "--always"])
    if ok and tag:
        return tag, None

    ok, branch = run_git(["rev-parse", "--abbrev-ref", "HEAD"])
    if ok:
        return branch, None
    return "unknown", None


def checkout_version(ref):
    if not ref or not REF_PATTERN.match(ref):
        return False, "Недопустимое имя версии"

    ok, msg = fetch_remote()
    if not ok:
        return False, msg

    ok, msg = run_git(["checkout", ref])
    if not ok:
        return False, msg
    return True, f"Переключено на {ref}"


def deploy_rebuild():
    if not is_deploy_enabled():
        return False, "Обновление через админку отключено (ALLOW_GIT_DEPLOY=false)"

    repo = get_repo_path()
    compose_file = os.path.join(repo, "docker-compose.yml")
    if not os.path.isfile(compose_file):
        return False, f"Не найден {compose_file}"

    commands = [
        ["docker", "compose", "-f", compose_file, "up", "-d", "--build"],
        ["docker-compose", "-f", compose_file, "up", "-d", "--build"],
    ]

    last_error = ""
    for cmd in commands:
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=600,
                cwd=repo,
            )
            if result.returncode == 0:
                return True, result.stdout or "Контейнеры пересобраны и запущены"
            last_error = result.stderr or result.stdout
        except FileNotFoundError:
            last_error = f"Команда не найдена: {cmd[0]}"
        except subprocess.TimeoutExpired:
            return False, "Превышено время ожидания пересборки (10 мин)"

    return False, last_error or "Не удалось выполнить docker compose"


def get_deploy_status():
    current, _ = get_current_version()
    tags, tags_err = list_tags()
    branches, branches_err = list_branches()
    return {
        "enabled": is_deploy_enabled(),
        "repo_path": get_repo_path(),
        "repo_ready": _repo_ready(),
        "remote_url": get_git_remote(),
        "current": current,
        "tags": tags[:30],
        "branches": branches[:30],
        "tags_error": tags_err,
        "branches_error": branches_err,
    }