80 lines
2.2 KiB
Python
Executable File
80 lines
2.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
在远程节点上执行 docker compose 操作。
|
|
|
|
用法:
|
|
python remote.py <服务目录> <命令>
|
|
|
|
命令:
|
|
up 启动/更新容器（不拉镜像）
|
|
recreate 强制重建容器（up -d --force-recreate，改配置后用）
|
|
restart 重启 compose 内所有服务
|
|
upgrade 拉取镜像并重建容器
|
|
ps 查看容器状态
|
|
logs 查看最近日志（非 follow）
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
from lib import project_root, service_info, ssh_cmd
|
|
|
|
REMOTE_COMMANDS = {
|
|
"up": "docker compose up -d",
|
|
"recreate": "docker compose up -d --force-recreate",
|
|
"restart": "docker compose restart",
|
|
"upgrade": "docker compose pull && docker compose up -d",
|
|
"ps": "docker compose ps",
|
|
"logs": "docker compose logs --tail=100",
|
|
}
|
|
|
|
|
|
def run_ssh(info: dict, compose_command: str) -> int:
|
|
remote_dir = info["remote_dir"]
|
|
remote_shell = f"cd {remote_dir} && {compose_command}"
|
|
cmd = ssh_cmd(info, remote_shell)
|
|
endpoint = info["node"]
|
|
if info.get("port") is not None:
|
|
endpoint = f"{endpoint}:{info['port']}"
|
|
print(f"远程执行: {endpoint}:{remote_dir}")
|
|
print(f"$ {compose_command}")
|
|
print("-" * 60)
|
|
try:
|
|
return subprocess.run(cmd, check=False).returncode
|
|
except FileNotFoundError:
|
|
print("错误: ssh 命令未找到")
|
|
return 1
|
|
|
|
|
|
def run_for_service(service_dir: str, action: str) -> int:
|
|
info = service_info(service_dir)
|
|
compose_command = REMOTE_COMMANDS[action]
|
|
endpoint = info["node"]
|
|
if info.get("port") is not None:
|
|
endpoint = f"{endpoint}:{info['port']}"
|
|
print(f"服务: {service_dir} -> {endpoint}:{info['remote_dir']}")
|
|
return run_ssh(info, compose_command)
|
|
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="在远程节点执行 docker compose 操作")
|
|
parser.add_argument("target", help="服务目录，例如 hosts/web1/myapp")
|
|
parser.add_argument(
|
|
"action",
|
|
choices=sorted(REMOTE_COMMANDS),
|
|
help="远程操作",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
os.chdir(project_root())
|
|
return run_for_service(args.target, args.action)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|