diary-news/scripts/deploy_pull.py

"""部署:从 Gitea 拉取最新代码到远程服务器(免密)。

行为:
- 服务器项目目录不存在?git clone : git fetch + reset --hard
- 拉前记录当前 HEAD(用于失败回滚)
- 失败回滚:git reset --hard <之前 sha>
- 输出 commit hash / log / working tree 状态 / 跟本地 main 对比

用法:
    python scripts/deploy_pull.py
    python scripts/deploy_pull.py --dry-run
    python scripts/deploy_pull.py --repo-dir /opt/diary-news
    python scripts/deploy_pull.py --rollback <sha>     # 手动回退

环境变量(覆盖默认值):
    DEPLOY_HOST, DEPLOY_PORT, DEPLOY_USER,
    DEPLOY_REPO_DIR, DEPLOY_REPO_URL, DEPLOY_SSH_KEY
"""
from __future__ import annotations

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

import paramiko

# ===== 默认配置(适配当前项目)=====
DEFAULT_HOST     = "207.57.129.228"
DEFAULT_PORT     = 19717
DEFAULT_USER     = "root"
DEFAULT_REPO_DIR = "/root/diary-news"
DEFAULT_REPO_URL = "http://124.223.26.33:3000/xiaji/diary-news.git"
DEFAULT_SSH_KEY  = os.path.expanduser("~/.ssh/id_ed25519")


def _run(c: paramiko.SSHClient, cmd: str, timeout: int = 60) -> tuple[int, str, str]:
    si, so, se = c.exec_command(cmd, timeout=timeout)
    out = so.read().decode(errors="replace")
    err = se.read().decode(errors="replace")
    rc = so.channel.recv_exit_status()
    return rc, out, err


def _connect(host: str, port: int, user: str, ssh_key: str) -> paramiko.SSHClient:
    # 依次尝试 RSA / Ed25519 / ECDSA(paramiko 5 没有统一入口)
    pkey: Any = None
    last_err: Exception | None = None
    for loader in (paramiko.RSAKey, paramiko.Ed25519Key, paramiko.ECDSAKey):
        try:
            pkey = loader.from_private_key_file(ssh_key)
            break
        except Exception as e:
            last_err = e
    if pkey is None:
        raise RuntimeError(f"无法解析 SSH key {ssh_key}: {last_err}")
    c = paramiko.SSHClient()
    c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    c.connect(host, port=port, username=user, pkey=pkey, timeout=30,
              allow_agent=False, look_for_keys=False)
    return c


def _local_head() -> str:
    """本地 HEAD sha(用作同步参考)。"""
    try:
        r = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            capture_output=True, text=True, cwd=str(Path(__file__).resolve().parents[1]),
            check=True,
        )
        return r.stdout.strip()
    except Exception as e:
        return f"unknown:{e}"


def deploy(
    host: str = DEFAULT_HOST,
    port: int = DEFAULT_PORT,
    user: str = DEFAULT_USER,
    repo_dir: str = DEFAULT_REPO_DIR,
    repo_url: str = DEFAULT_REPO_URL,
    ssh_key: str = DEFAULT_SSH_KEY,
    branch: str = "main",
    dry_run: bool = False,
    rollback: str | None = None,
) -> int:
    if not Path(ssh_key).exists():
        print(f"ERROR: SSH key 不存在: {ssh_key}", file=sys.stderr)
        return 2

    print(f"=== 连接 {user}@{host}:{port} ===")
    c = _connect(host, port, user, ssh_key)
    print(f"✓ 已免密登录")

    try:
        # 1) 手动回退
        if rollback:
            print(f"\n=== 手动回退到 {rollback[:12]} ===")
            rc, out, err = _run(c, f"cd {repo_dir} && git reset --hard {rollback}", timeout=30)
            if rc != 0:
                print(f"✗ 回退失败: {err}")
                return 1
            print("✓ 已回退")
            rc, head, _ = _run(c, f"cd {repo_dir} && git rev-parse HEAD", timeout=10)
            print(f"现在 HEAD: {head.strip()[:12]}")
            return 0

        # 2) 探查:目录在不在?
        rc, exists, _ = _run(c, f"test -d {repo_dir}/.git && echo EXISTS || echo MISSING", timeout=10)
        action = "pull" if "EXISTS" in exists else "clone"

        # 3) 拉前 HEAD(用于回滚)
        before_sha: str | None = None
        if action == "pull":
            rc, h, _ = _run(c, f"cd {repo_dir} && git rev-parse HEAD", timeout=10)
            before_sha = h.strip()
            print(f"拉前 HEAD: {before_sha[:12]}")

        if dry_run:
            print(f"DRY-RUN: 将要 {action} from {repo_url} → {repo_dir}")
            return 0

        # 4) 拉
        print(f"\n=== {action.upper()} ===")
        if action == "clone":
            cmd = f"git clone --depth 50 {repo_url} {repo_dir}"
        else:
            cmd = (
                f"cd {repo_dir} && "
                f"git fetch origin {branch} && "
                f"git reset --hard origin/{branch}"
            )
        rc, out, err = _run(c, cmd, timeout=180)

        if rc != 0:
            print(f"✗ {action} 失败 (rc={rc})")
            if out.strip(): print(f"stdout:\n{out}")
            if err.strip(): print(f"stderr:\n{err}")
            # 回滚(只在 pull 时)
            if before_sha and action == "pull":
                print(f"\n=== 回退到 {before_sha[:12]} ===")
                rrc, _, rerr = _run(c, f"cd {repo_dir} && git reset --hard {before_sha}", timeout=30)
                if rrc == 0:
                    print(f"✓ 已回退到 {before_sha[:12]}")
                else:
                    print(f"✗ 回退失败: {rerr}")
            return 1

        print(f"✓ {action} 成功")
        if out.strip(): print(out)

        # 5) 拉后状态
        rc, head_sha, _ = _run(c, f"cd {repo_dir} && git rev-parse HEAD", timeout=10)
        head_sha = head_sha.strip()
        print(f"\n拉后 HEAD: {head_sha[:12]}")

        rc, log, _ = _run(c, f"cd {repo_dir} && git log --oneline -5", timeout=10)
        print("--- 最近的 commit ---")
        print(log)

        rc, status, _ = _run(c, f"cd {repo_dir} && git status --porcelain", timeout=10)
        if status.strip():
            print(f"⚠️ working tree 不干净:\n{status}")
        else:
            print("✓ working tree 干净")

        # 6) 跟本地 main 对比
        local = _local_head()
        print(f"\n=== 同步对比 ===")
        print(f"本地 HEAD:     {local[:12]}")
        print(f"服务器 HEAD:   {head_sha[:12]}")
        if local.startswith(head_sha):
            print("✓ 服务器 == 本地")
        else:
            print("ℹ️ 服务器跟本地不完全一致(可能本地未推送,或服务器在另一条 commit 上)")

        # 7) 总结
        print(f"\n=== 部署报告 ===")
        print(f"服务器:     {user}@{host}:{port}")
        print(f"项目目录:   {repo_dir}")
        print(f"动作:       {action}")
        print(f"最终 HEAD:  {head_sha[:12]}")
        return 0
    finally:
        c.close()


def main() -> None:
    p = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument("--host",     default=os.environ.get("DEPLOY_HOST", DEFAULT_HOST))
    p.add_argument("--port",     type=int, default=int(os.environ.get("DEPLOY_PORT", str(DEFAULT_PORT))))
    p.add_argument("--user",     default=os.environ.get("DEPLOY_USER", DEFAULT_USER))
    p.add_argument("--repo-dir", default=os.environ.get("DEPLOY_REPO_DIR", DEFAULT_REPO_DIR))
    p.add_argument("--repo-url", default=os.environ.get("DEPLOY_REPO_URL", DEFAULT_REPO_URL))
    p.add_argument("--ssh-key",  default=os.environ.get("DEPLOY_SSH_KEY", DEFAULT_SSH_KEY))
    p.add_argument("--branch",   default="main")
    p.add_argument("--dry-run",  action="store_true")
    p.add_argument("--rollback", metavar="SHA", help="手动回退到指定 commit")
    args = p.parse_args()
    sys.exit(deploy(
        host=args.host, port=args.port, user=args.user,
        repo_dir=args.repo_dir, repo_url=args.repo_url,
        ssh_key=args.ssh_key, branch=args.branch,
        dry_run=args.dry_run, rollback=args.rollback,
    ))


if __name__ == "__main__":
    main()