.pouch/skills/deployer/scripts/deploy/sync.py

#!/usr/bin/env python3
"""
目录同步脚本
将指定目录同步到远程机器

用法:
    python sync.py <目录路径>
    例如: python sync.py hosts/web1/myapp
"""

from __future__ import annotations

import argparse
import os
import shlex
import shutil
import subprocess
import sys
from pathlib import Path

_SCRIPT_DIR = Path(__file__).resolve().parent
if str(_SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(_SCRIPT_DIR))

from lib import DEFAULT_BASE_PATH, rsync_ssh_args, scp_base_args, service_info, ssh_cmd  # noqa: E402


def sync_directory(info: dict) -> int:
    if shutil.which("rsync"):
        return _sync_rsync(info)
    print("本机没有 rsync，改用 scp（不会删除远程多余文件）")
    return _sync_scp(info)


def _sync_rsync(info: dict) -> int:
    source_dir = info["service_dir"]
    node = info["node"]
    remote_path = f"{node}:{info['remote_dir']}"

    rsync_cmd = [
        "rsync",
        "-avz",
        "--delete",
        *rsync_ssh_args(info),
    ]
    for item in info["sync_exclude"]:
        rsync_cmd.append(f"--exclude={item}/")

    rsync_cmd.extend([f"{source_dir}/", remote_path])

    print(f"正在同步 {source_dir} 到 {remote_path}")
    if info["sync_exclude"]:
        print(f"排除目录: {', '.join(info['sync_exclude'])}")
    print(f"执行命令: {' '.join(rsync_cmd)}")
    print("-" * 60)

    try:
        subprocess.run(rsync_cmd, check=True)
        print("-" * 60)
        print("同步完成!")
        return 0
    except subprocess.CalledProcessError as e:
        print(f"错误: rsync 执行失败，退出码: {e.returncode}")
        return 1
    except FileNotFoundError:
        print("本机没有 rsync，改用 scp（不会删除远程多余文件）")
        return _sync_scp(info)


def collect_sync_files(source_dir: str, excludes: list[str]) -> list[Path]:
    """列出待同步文件；排除规则与 rsync `--exclude=name/` 一致（任意层级的同名目录）。"""
    source = Path(source_dir)
    exclude_set = set(excludes)
    files: list[Path] = []
    for root, dirs, names in os.walk(source):
        dirs[:] = [d for d in dirs if d not in exclude_set]
        for name in names:
            path = Path(root) / name
            if path.is_file():
                files.append(path)
    return files


def _sync_scp(info: dict) -> int:
    source_dir = info["service_dir"]
    remote_dir = info["remote_dir"]
    node = info["node"]
    source = Path(source_dir)
    files = collect_sync_files(source_dir, info["sync_exclude"])

    print(f"正在同步 {source_dir} 到 {node}:{remote_dir}")
    if info["sync_exclude"]:
        print(f"排除目录: {', '.join(info['sync_exclude'])}")
    print("-" * 60)

    remote_dirs = {remote_dir}
    grouped: dict[str, list[Path]] = {}
    for path in files:
        parent = path.relative_to(source).parent.as_posix()
        grouped.setdefault(parent, []).append(path)
        if parent != ".":
            remote_dirs.add(f"{remote_dir}/{parent}")

    mkdir_remote = "mkdir -p " + " ".join(shlex.quote(d) for d in sorted(remote_dirs))
    mkdir_cmd = ssh_cmd(info, mkdir_remote)
    print(f"执行命令: {' '.join(mkdir_cmd)}")
    try:
        completed = subprocess.run(mkdir_cmd, check=False)
    except FileNotFoundError:
        print("错误: ssh 命令未找到")
        return 1
    if completed.returncode:
        print(f"错误: 远程 mkdir 失败，退出码: {completed.returncode}")
        return completed.returncode

    if not files:
        print("没有需要同步的文件")
        print("-" * 60)
        print("同步完成!")
        return 0

    try:
        for parent, group in sorted(grouped.items()):
            dest = remote_dir if parent == "." else f"{remote_dir}/{parent}"
            scp_cmd = [
                *scp_base_args(info),
                "-p",
                "--",
                *[str(p) for p in group],
                f"{node}:{dest}/",
            ]
            print(f"执行命令: {' '.join(scp_cmd)}")
            subprocess.run(scp_cmd, check=True)
    except subprocess.CalledProcessError as e:
        print(f"错误: scp 执行失败，退出码: {e.returncode}")
        return 1
    except FileNotFoundError:
        print("错误: scp 命令未找到，请确保已安装 openssh-client")
        return 1

    print("-" * 60)
    print("同步完成!")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(
        description="将指定目录同步到远程机器",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  python sync.py hosts/web1/myapp
  python sync.py infra/traefik --base-path /opt/app
        """,
    )

    parser.add_argument(
        "directory",
        help="要同步的目录路径（相对项目根，例如: hosts/web1/myapp）",
    )

    parser.add_argument(
        "--base-path",
        default=DEFAULT_BASE_PATH,
        help=f"远程基础路径（默认: {DEFAULT_BASE_PATH}）",
    )

    args = parser.parse_args()

    info = service_info(args.directory)
    if args.base_path != DEFAULT_BASE_PATH:
        name = info["name"]
        info["base_path"] = args.base_path
        info["remote_dir"] = f"{args.base_path}/{name}"

    print(f"目标节点: {info['node']}")
    if info.get("port") is not None:
        print(f"SSH 端口: {info['port']}")
    if info.get("identity_file"):
        print(f"SSH 密钥: {info['identity_file']}")
    print(f"源目录: {args.directory}")
    print(f"远程路径: {info['remote_dir']}")
    print()

    return sync_directory(info)


if __name__ == "__main__":
    sys.exit(main())