ai_fwt_pro/main/xiaozhi-server/deploy/tencent/deploy_tencent.py

import argparse
import getpass
import os
import shlex
import sys
import tarfile
import tempfile
import time
from pathlib import Path

import paramiko


EXCLUDED_DIRS = {
    ".git",
    ".venv",
    "__pycache__",
    "node_modules",
    "tmp",
}

EXCLUDED_FILES = {
    "data/.config.yaml",
}

EXCLUDED_PREFIXES = (
    "data/bin/",
)

EXCLUDED_SUFFIXES = (
    ".pyc",
    ".pyo",
    ".pyd",
)


def should_exclude(rel_path: str) -> bool:
    normalized = rel_path.replace("\\", "/").strip("./")
    if not normalized:
        return False
    if normalized in EXCLUDED_FILES:
        return True
    if normalized.endswith(EXCLUDED_SUFFIXES):
        return True
    for prefix in EXCLUDED_PREFIXES:
        if normalized.startswith(prefix):
            return True
    parts = normalized.split("/")
    if any(part in EXCLUDED_DIRS for part in parts):
        return True
    if normalized.startswith("data/") and normalized.endswith(".sqlite3"):
        return True
    return False


def create_archive(source_dir: Path) -> str:
    temp_file = tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False)
    temp_file.close()
    archive_path = temp_file.name
    with tarfile.open(archive_path, "w:gz") as tar:
        for path in source_dir.rglob("*"):
            if not path.is_file():
                continue
            rel_path = path.relative_to(source_dir).as_posix()
            if should_exclude(rel_path):
                continue
            tar.add(path, arcname=rel_path)
    return archive_path


def run_remote_command(client: paramiko.SSHClient, command: str) -> None:
    stdin, stdout, stderr = client.exec_command(command, get_pty=True)
    exit_status = stdout.channel.recv_exit_status()
    out = stdout.read().decode("utf-8", "ignore")
    err = stderr.read().decode("utf-8", "ignore")

    def _safe_write(stream, text: str) -> None:
        if not text:
            return
        if not text.endswith("\n"):
            text += "\n"
        encoding = getattr(stream, "encoding", None) or "utf-8"
        stream.buffer.write(text.encode(encoding, errors="replace"))
        stream.flush()

    if out:
        _safe_write(sys.stdout, out)
    if err:
        _safe_write(sys.stderr, err)
    if exit_status != 0:
        raise RuntimeError(f"Remote command failed with exit code {exit_status}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Deploy xingxing-server to Tencent Cloud over SSH.")
    parser.add_argument("--host", required=True, help="Remote host or IP.")
    parser.add_argument("--user", required=True, help="SSH username.")
    parser.add_argument("--password", help="SSH password. If omitted, prompt securely.")
    parser.add_argument(
        "--remote-dir",
        default="/home/ubuntu/xingxing-server",
        help="Remote project directory.",
    )
    parser.add_argument(
        "--source-dir",
        default=str(Path(__file__).resolve().parents[2]),
        help="Local xingxing-server source directory.",
    )
    parser.add_argument(
        "--compose-file",
        default="docker-compose.tencent.yml",
        help="Compose file to use on the remote host.",
    )
    args = parser.parse_args()

    source_dir = Path(args.source_dir).resolve()
    if not source_dir.exists():
        raise FileNotFoundError(f"Source directory does not exist: {source_dir}")

    password = args.password or getpass.getpass("SSH password: ")
    archive_path = create_archive(source_dir)
    timestamp = time.strftime("%Y%m%d%H%M%S")
    remote_archive = f"/tmp/xingxing-server-{timestamp}.tar.gz"
    remote_stage = f"{args.remote_dir}.stage.{timestamp}"
    remote_backup = f"{args.remote_dir}.bak.{timestamp}"

    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        print(f"Connecting to {args.user}@{args.host} ...")
        client.connect(
            hostname=args.host,
            username=args.user,
            password=password,
            timeout=20,
            banner_timeout=20,
            auth_timeout=20,
        )

        print(f"Uploading archive to {remote_archive} ...")
        sftp = client.open_sftp()
        sftp.put(archive_path, remote_archive)
        sftp.close()

        remote_dir_q = shlex.quote(args.remote_dir)
        remote_stage_q = shlex.quote(remote_stage)
        remote_backup_q = shlex.quote(remote_backup)
        remote_archive_q = shlex.quote(remote_archive)
        compose_file_q = shlex.quote(args.compose_file)
        sudo_password_q = shlex.quote(password)

        command = f"""
set -e
REMOTE_DIR={remote_dir_q}
STAGE_DIR={remote_stage_q}
BACKUP_DIR={remote_backup_q}
ARCHIVE={remote_archive_q}
SUDO_PASSWORD={sudo_password_q}
rm -rf "$STAGE_DIR"
mkdir -p "$STAGE_DIR"
tar -xzf "$ARCHIVE" -C "$STAGE_DIR"
if [ -d "$REMOTE_DIR/data" ]; then
  mkdir -p "$STAGE_DIR/data"
  cp -a "$REMOTE_DIR/data/." "$STAGE_DIR/data/"
fi
if [ -d "$REMOTE_DIR" ]; then
  rm -rf "$BACKUP_DIR"
  mv "$REMOTE_DIR" "$BACKUP_DIR"
fi
mv "$STAGE_DIR" "$REMOTE_DIR"
rm -f "$ARCHIVE"
cd "$REMOTE_DIR"
printf '%s\\n' "$SUDO_PASSWORD" | sudo -S -p '' docker compose -f {compose_file_q} up -d --build
printf '%s\\n' "$SUDO_PASSWORD" | sudo -S -p '' docker compose -f {compose_file_q} ps
"""
        print("Rebuilding and restarting remote service ...")
        run_remote_command(client, command)
        print(f"Deploy completed. Backup kept at {remote_backup}")
        return 0
    finally:
        client.close()
        if os.path.exists(archive_path):
            os.remove(archive_path)


if __name__ == "__main__":
    raise SystemExit(main())