Files
ccdi/deploy/remote-deploy-tongweb.py
2026-04-17 10:18:13 +08:00

137 lines
4.8 KiB
Python

import argparse
import posixpath
import shlex
import sys
from pathlib import Path
import paramiko
def parse_args():
parser = argparse.ArgumentParser(description="Upload backend war to NAS and restart TongWeb.")
parser.add_argument("--host", required=True)
parser.add_argument("--port", type=int, required=True)
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument("--local-war", required=True)
parser.add_argument("--remote-root", required=True)
parser.add_argument("--tongweb-home", required=True)
parser.add_argument("--app-name", required=True)
return parser.parse_args()
def run_command(ssh, command):
stdin, stdout, stderr = ssh.exec_command(command)
exit_code = stdout.channel.recv_exit_status()
output = stdout.read().decode("utf-8", errors="ignore")
error = stderr.read().decode("utf-8", errors="ignore")
return exit_code, output, error
def sudo_prefix(password):
return f"printf '%s\\n' {shlex.quote(password)} | sudo -S -p '' "
def detect_command_prefix(ssh, password, command):
plain_exit_code, _, _ = run_command(ssh, f"{command} >/dev/null 2>&1")
if plain_exit_code == 0:
return ""
sudo_probe = f"{sudo_prefix(password)}{command} >/dev/null 2>&1"
sudo_exit_code, _, _ = run_command(ssh, sudo_probe)
if sudo_exit_code == 0:
return sudo_prefix(password)
raise RuntimeError(f"Remote command is not accessible: {command}")
def ensure_remote_path(ssh, prefix, remote_path):
command = f"{prefix}mkdir -p {shlex.quote(remote_path)}"
exit_code, output, error = run_command(ssh, command)
if exit_code != 0:
raise RuntimeError(f"Failed to create remote directory {remote_path}:\n{output}\n{error}")
def upload_file(sftp, local_file, remote_file):
parent_dir = posixpath.dirname(remote_file)
try:
sftp.listdir(parent_dir)
except OSError:
raise RuntimeError(f"SFTP remote directory not found: {parent_dir}")
sftp.put(str(local_file), remote_file)
def build_deploy_command(args, prefix):
app_war_name = f"{args.app_name}.war"
remote_war_path = posixpath.join(args.remote_root.rstrip("/"), "backend", app_war_name)
autodeploy_dir = posixpath.join(args.tongweb_home.rstrip("/"), "autodeploy")
deployed_war_path = posixpath.join(autodeploy_dir, app_war_name)
deployed_dir_path = posixpath.join(autodeploy_dir, args.app_name)
stop_script = posixpath.join(args.tongweb_home.rstrip("/"), "bin", "stopserver.sh")
start_script = posixpath.join(args.tongweb_home.rstrip("/"), "bin", "startservernohup.sh")
return (
"set -e;"
f"test -d {shlex.quote(args.tongweb_home)};"
f"test -x {shlex.quote(stop_script)};"
f"test -x {shlex.quote(start_script)};"
f"{prefix}mkdir -p {shlex.quote(autodeploy_dir)};"
f"{prefix}sh {shlex.quote(stop_script)} >/dev/null 2>&1 || true;"
f"{prefix}rm -rf {shlex.quote(deployed_dir_path)};"
f"{prefix}rm -f {shlex.quote(deployed_war_path)};"
f"{prefix}cp {shlex.quote(remote_war_path)} {shlex.quote(deployed_war_path)};"
f"{prefix}sh {shlex.quote(start_script)};"
"sleep 5;"
f"ls -l {shlex.quote(autodeploy_dir)};"
)
def main():
args = parse_args()
local_war = Path(args.local_war).resolve()
if not local_war.exists():
raise FileNotFoundError(f"Local war does not exist: {local_war}")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(
hostname=args.host,
port=args.port,
username=args.username,
password=args.password,
timeout=20,
)
sftp = ssh.open_sftp()
try:
remote_root = args.remote_root.rstrip("/")
remote_backend_dir = posixpath.join(remote_root, "backend")
remote_war_path = posixpath.join(remote_backend_dir, f"{args.app_name}.war")
ensure_remote_path(ssh, "", remote_root)
ensure_remote_path(ssh, "", remote_backend_dir)
upload_file(sftp, local_war, remote_war_path)
command_prefix = detect_command_prefix(ssh, args.password, f"test -d {shlex.quote(args.tongweb_home)}")
deploy_command = build_deploy_command(args, command_prefix)
exit_code, output, error = run_command(ssh, deploy_command)
if exit_code != 0:
raise RuntimeError(f"Remote TongWeb deploy failed:\n{output}\n{error}")
print("=== DEPLOY OUTPUT ===")
print(output.strip())
if error.strip():
print("=== DEPLOY STDERR ===")
print(error.strip())
finally:
sftp.close()
ssh.close()
if __name__ == "__main__":
try:
main()
except Exception as exc:
print(str(exc), file=sys.stderr)
sys.exit(1)