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()}" result = subprocess.run( ["git", "-C", get_repo_path()] + 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 fetch_remote(): remote = get_git_remote() if remote: ok, msg = run_git(["remote", "set-url", "origin", remote]) if not ok: return False, msg return run_git(["fetch", "--all", "--tags", "--prune"], timeout=180) def list_tags(): ok, msg = fetch_remote() if not ok: return [], msg ok, out = run_git(["tag", "--sort=-version:refname"]) if not ok: return [], out return [line for line in out.splitlines() if line.strip()], None def list_branches(): ok, msg = fetch_remote() if not ok: return [], msg ok, out = run_git(["branch", "-a", "--format=%(refname:short)"]) if not ok: return [], out branches = [] for line in out.splitlines(): name = line.strip().replace("origin/", "") if name and name not in branches and "HEAD" not in name: branches.append(name) return 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, }