#!/usr/bin/env python3 from __future__ import annotations import argparse import os import posixpath import stat import sys from pathlib import Path from typing import Iterable DEFAULT_PASSWORD = "Nie@159357" SKIP_DIR_NAMES = {"__pycache__", ".pytest_cache", ".mypy_cache", ".ruff_cache", ".git"} SKIP_FILE_SUFFIXES = {".pyc", ".pyo", ".DS_Store"} SKIP_FILE_NAMES = {"music_server_deploy.tar", "music_server_deploy.zip"} def parse_args() -> argparse.Namespace: script_dir = Path(__file__).resolve().parent project_root = script_dir.parent default_template_dir = script_dir / "templates" parser = argparse.ArgumentParser( description="Upload Music_Server to NAS staging and trigger deploy_and_restart.sh" ) parser.add_argument("--host", default="192.168.5.43") parser.add_argument("--port", type=int, default=222) parser.add_argument("--user", default="xiaoming") parser.add_argument( "--password", default=os.environ.get("NAS_192168543_PASSWORD") or DEFAULT_PASSWORD, ) parser.add_argument("--remote-app-home", default="/volume4/Music_Cloud/Music_Server") parser.add_argument("--source-dir", default=str(project_root)) parser.add_argument("--template-dir", default=str(default_template_dir)) parser.add_argument("--skip-health-check", action="store_true") return parser.parse_args() def to_sftp_path(shell_path: str) -> str: normalized = shell_path.rstrip("/") if normalized.startswith("/volume4"): mapped = normalized[len("/volume4") :] return mapped or "/" return normalized or "/" def shell_quote(value: str) -> str: return "'" + value.replace("'", "'\"'\"'") + "'" def ensure_remote_dir(sftp, path: str) -> None: path = path.rstrip("/") if not path or path == "/": return parts: list[str] = [] current = path while current not in ("", "/"): parts.append(current) current = posixpath.dirname(current) for part in reversed(parts): try: sftp.stat(part) except OSError: sftp.mkdir(part) def exists_remote(sftp, path: str) -> bool: try: sftp.stat(path) return True except OSError: return False def remove_remote_tree(sftp, root: str) -> None: if not exists_remote(sftp, root): return for entry in sftp.listdir_attr(root): child = posixpath.join(root, entry.filename) if stat.S_ISDIR(entry.st_mode): remove_remote_tree(sftp, child) else: sftp.remove(child) sftp.rmdir(root) def iter_local_files(local_root: Path) -> Iterable[Path]: for current_root, dir_names, file_names in os.walk(local_root): dir_names[:] = [name for name in dir_names if name not in SKIP_DIR_NAMES] for file_name in file_names: path = Path(current_root) / file_name if path.suffix in SKIP_FILE_SUFFIXES: continue if path.name in SKIP_FILE_NAMES: continue yield path def upload_tree(sftp, local_root: Path, remote_root: str) -> int: ensure_remote_dir(sftp, remote_root) uploaded = 0 for local_file in iter_local_files(local_root): relative_path = local_file.relative_to(local_root).as_posix() remote_file = posixpath.join(remote_root, relative_path) remote_dir = posixpath.dirname(remote_file) ensure_remote_dir(sftp, remote_dir) sftp.put(str(local_file), remote_file) uploaded += 1 return uploaded def read_channel_text(channel_file) -> str: data = channel_file.read() if isinstance(data, bytes): return data.decode("utf-8", "replace") return str(data) def run_remote_command(client, command: str, sudo_password: str | None = None) -> int: remote_command = command if sudo_password: remote_command = f"sudo -S sh -lc {shell_quote(command)}" stdin, stdout, stderr = client.exec_command(remote_command) if sudo_password: stdin.write(sudo_password + "\n") stdin.flush() out_text = read_channel_text(stdout) err_text = read_channel_text(stderr) if out_text: print(out_text, end="") if err_text: print(err_text, file=sys.stderr, end="") return stdout.channel.recv_exit_status() def main() -> int: import paramiko args = parse_args() source_dir = Path(args.source_dir).resolve() template_dir = Path(args.template_dir).resolve() if not source_dir.exists(): print(f"Source dir not found: {source_dir}", file=sys.stderr) return 2 deploy_template = template_dir / "deploy_and_restart.sh" if not deploy_template.exists(): print(f"Missing template: {deploy_template}", file=sys.stderr) return 2 remote_app_home_shell = args.remote_app_home.rstrip("/") remote_app_home_sftp = to_sftp_path(remote_app_home_shell) remote_staging_shell = f"{remote_app_home_shell}/deploy/staging/music-server-app" remote_staging_sftp = to_sftp_path(remote_staging_shell) remote_bin_sftp = to_sftp_path(f"{remote_app_home_shell}/bin") client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) print( f"[deploy_to_nas] Connecting {args.user}@{args.host}:{args.port} ...", flush=True, ) client.connect( hostname=args.host, port=args.port, username=args.user, password=args.password, timeout=15, banner_timeout=15, auth_timeout=15, ) try: sftp = client.open_sftp() try: ensure_remote_dir(sftp, remote_app_home_sftp) ensure_remote_dir(sftp, remote_bin_sftp) if exists_remote(sftp, remote_staging_sftp): print( f"[deploy_to_nas] Clearing staging: {remote_staging_sftp}", flush=True, ) remove_remote_tree(sftp, remote_staging_sftp) ensure_remote_dir(sftp, remote_staging_sftp) uploaded_count = upload_tree(sftp, source_dir, remote_staging_sftp) print( f"[deploy_to_nas] Uploaded {uploaded_count} files to staging.", flush=True, ) sftp.put( str(deploy_template), posixpath.join(remote_bin_sftp, "deploy_and_restart.sh"), ) finally: sftp.close() chmod_cmd = ( f"chmod +x {shell_quote(remote_app_home_shell + '/bin/deploy_and_restart.sh')}" ) chmod_exit = run_remote_command(client, chmod_cmd) if chmod_exit != 0: print( f"[deploy_to_nas] chmod failed with exit code {chmod_exit}", file=sys.stderr, ) return chmod_exit deploy_cmd = ( f"bash {shell_quote(remote_app_home_shell + '/bin/deploy_and_restart.sh')} " f"--staging-dir {shell_quote(remote_staging_shell)}" ) if args.skip_health_check: deploy_cmd += " --skip-health-check" print("[deploy_to_nas] Running deploy command...", flush=True) deploy_exit = run_remote_command(client, deploy_cmd, sudo_password=args.password) if deploy_exit != 0: print( f"[deploy_to_nas] Deploy failed with exit code {deploy_exit}", file=sys.stderr, ) return deploy_exit print("[deploy_to_nas] Deploy completed successfully.", flush=True) return 0 finally: client.close() if __name__ == "__main__": raise SystemExit(main())