#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
soulApi（soul-api 后端）Go 项目一键部署到宝塔（正式环境）
默认全程走宝塔面板 API（无需 SSH）：探活 → 本机编译打包 → 面板「上传文件」→「解压」→「删包」→ Site 启停重启。

1.（默认）宝塔 API 探活：/system?action=GetDiskInfo
2. 本地交叉编译 → tar.gz（包内 soul-api 强制 755，避免 Windows 打包丢执行位）
3. /files?action=upload 分片上传到 DEPLOY_PROJECT_PATH
4. /files?action=UnZip 解压到同目录；/files?action=DeleteFile 删除 tar
5. SiteStop/SiteStart 重启 Go 站点；auto 模式下自动执行第二轮启停（无 SSH 时尽量换新进程）

命令行：
  python3 master.py                          # 默认 --deploy-method btapi
  python3 master.py --deploy-method ssh      # 仅当服务器允许密码/密钥 SSH 时使用旧逻辑
  python3 master.py --skip-bt-ping
  python3 master.py --verify
  python3 master.py --restart-method btapi

环境变量：BT_PANEL_URL、BT_API_KEY、BT_GO_PROJECT_NAME、BT_GO_SITE_ID；可选 BT_UPLOAD_CHUNK_MB（默认 4）
依赖：requests 必选；paramiko 仅在 --deploy-method ssh 时需要
"""

from __future__ import print_function

import hashlib
import json
import os
import sys
import tempfile
import argparse
import subprocess
import shutil
import tarfile
import time
import threading
import shlex

try:
    import paramiko
except ImportError:
    paramiko = None

try:
    import requests
    try:
        import urllib3
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    except Exception:
        pass
except ImportError:
    requests = None

# ==================== 配置 ====================

DEPLOY_PROJECT_PATH = "/www/wwwroot/self/soul-api"
DEFAULT_SSH_PORT = int(os.environ.get("DEPLOY_SSH_PORT", "22022"))


# 宝塔 API 密钥（写死，用于部署后重启 Go 项目）
BT_API_KEY_DEFAULT = "qcWubCdlfFjS2b2DMT1lzPFaDfmv1cBT"


def get_cfg():
    host = os.environ.get("DEPLOY_HOST", "43.139.27.93")
    bt_url = (os.environ.get("BT_PANEL_URL") or "").strip().rstrip("/")
    if not bt_url:
        bt_url = "https://%s:9988" % host
    return {
        "host": host,
        "user": os.environ.get("DEPLOY_USER", "root"),
        "password": os.environ.get("DEPLOY_PASSWORD", "Zhiqun1984"),
        "ssh_key": os.environ.get("DEPLOY_SSH_KEY", ""),
        "project_path": os.environ.get("DEPLOY_PROJECT_PATH", DEPLOY_PROJECT_PATH),
        "bt_panel_url": bt_url,
        "bt_api_key": os.environ.get("BT_API_KEY", BT_API_KEY_DEFAULT),
        "bt_go_project_name": os.environ.get("BT_GO_PROJECT_NAME", "soulApi"),
    }


# ==================== 本地构建 ====================


def run_build(root):
    """交叉编译 Go 二进制（Linux amd64）"""
    print("[1/4] 本地交叉编译 Go 二进制 ...")
    env = os.environ.copy()
    env["GOOS"] = "linux"
    env["GOARCH"] = "amd64"
    env["CGO_ENABLED"] = "0"
    # 必须 shell=False，否则 Windows 下 -ldflags 等参数会被当成包路径导致 "malformed import path"
    cmd = ["go", "build", "-o", "soul-api", "./cmd/server"]
    try:
        r = subprocess.run(
            cmd,
            cwd=root,
            env=env,
            shell=False,
            timeout=120,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
        if r.returncode != 0:
            print("  [失败] go build 失败，退出码:", r.returncode)
            if r.stderr:
                for line in (r.stderr or "").strip().split("\n")[-10:]:
                    print("    " + line)
            return None
        out_path = os.path.join(root, "soul-api")
        if not os.path.isfile(out_path):
            print("  [失败] 未找到编译产物 soul-api")
            return None
        print("  [成功] 编译完成: %s (%.2f MB)" % (out_path, os.path.getsize(out_path) / 1024 / 1024))
        return out_path
    except subprocess.TimeoutExpired:
        print("  [失败] 编译超时")
        return None
    except FileNotFoundError:
        print("  [失败] 未找到 go 命令，请安装 Go")
        return None
    except Exception as e:
        print("  [失败] 编译异常:", str(e))
        return None


# ==================== 打包 ====================

# 正式环境 Nginx 一般反代 8080；可用环境变量覆盖：DEPLOY_API_PORT=9090
DEPLOY_PORT = int(os.environ.get("DEPLOY_API_PORT", "8080"))


def set_env_port(env_path, port=DEPLOY_PORT):
    """将 .env 文件中的 PORT 设为指定值（用于部署包）"""
    if not os.path.isfile(env_path):
        return
    with open(env_path, "r", encoding="utf-8", errors="replace") as f:
        lines = f.readlines()
    found = False
    new_lines = []
    for line in lines:
        s = line.strip()
        if "=" in s and s.split("=", 1)[0].strip() == "PORT":
            new_lines.append("PORT=%s\n" % port)
            found = True
        else:
            new_lines.append(line)
    if not found:
        new_lines.append("PORT=%s\n" % port)
    with open(env_path, "w", encoding="utf-8", newline="\n") as f:
        f.writelines(new_lines)


def set_env_mini_program_state(env_path, state):
    """将 .env 中的 WECHAT_MINI_PROGRAM_STATE 设为 developer/formal（打包前按环境覆盖）"""
    if not os.path.isfile(env_path):
        return
    key = "WECHAT_MINI_PROGRAM_STATE"
    with open(env_path, "r", encoding="utf-8", errors="replace") as f:
        lines = f.readlines()
    found = False
    new_lines = []
    for line in lines:
        s = line.strip()
        if "=" in s and s.split("=", 1)[0].strip() == key:
            new_lines.append("%s=%s\n" % (key, state))
            found = True
        else:
            new_lines.append(line)
    if not found:
        new_lines.append("%s=%s\n" % (key, state))
    with open(env_path, "w", encoding="utf-8", newline="\n") as f:
        f.writelines(new_lines)


def pack_deploy(root, binary_path, include_env=True):
    """打包二进制和 .env 为 tar.gz"""
    print("[2/4] 打包部署文件 ...")
    staging = tempfile.mkdtemp(prefix="soul_api_deploy_")
    try:
        shutil.copy2(binary_path, os.path.join(staging, "soul-api"))
        env_src = os.path.join(root, ".env.production")
        staging_env = os.path.join(staging, ".env")
        if include_env and os.path.isfile(env_src):
            shutil.copy2(env_src, staging_env)
            print("  [已包含] .env.production -> .env")
        else:
            env_example = os.path.join(root, ".env.example")
            if os.path.isfile(env_example):
                shutil.copy2(env_example, staging_env)
                print("  [已包含] .env.example -> .env (请服务器上检查配置)")
        if os.path.isfile(staging_env):
            set_env_port(staging_env, DEPLOY_PORT)
            set_env_mini_program_state(staging_env, "formal")
            print("  [已设置] PORT=%s（部署用）, WECHAT_MINI_PROGRAM_STATE=formal（正式环境）" % DEPLOY_PORT)
        tarball = os.path.join(tempfile.gettempdir(), "soul_api_deploy.tar.gz")
        with tarfile.open(tarball, "w:gz") as tf:
            for name in os.listdir(staging):
                full = os.path.join(staging, name)
                if name == "soul-api":
                    ti = tf.gettarinfo(full, arcname=name)
                    ti.mode = 0o755
                    with open(full, "rb") as bf:
                        tf.addfile(ti, bf)
                else:
                    tf.add(full, arcname=name)
        print("  [成功] 打包完成: %s (%.2f MB)" % (tarball, os.path.getsize(tarball) / 1024 / 1024))
        return tarball
    except Exception as e:
        print("  [失败] 打包异常:", str(e))
        return None
    finally:
        shutil.rmtree(staging, ignore_errors=True)


# ==================== 宝塔 API 重启 ====================


def _bt_signed_post(base_url, key, path, extra_data, timeout=20):
    """单次宝塔签名 POST（每请求独立 request_time/token）。"""
    req_time = int(time.time())
    sk_md5 = hashlib.md5(key.encode()).hexdigest()
    req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
    data = {"request_time": req_time, "request_token": req_token}
    data.update(extra_data or {})
    return requests.post(base_url + path, data=data, timeout=timeout, verify=False)


def _bt_parse_json_response(r):
    """解析面板 JSON（部分响应 Content-Type 不准）。"""
    if r is None or r.status_code != 200:
        return None
    ct = (r.headers.get("content-type") or "").lower()
    if "json" in ct:
        try:
            return r.json()
        except Exception:
            pass
    t = (r.text or "").lstrip()
    if t.startswith("{"):
        try:
            return json.loads(r.text)
        except Exception:
            pass
    return None


def _bt_discover_go_site(base, key, want_name):
    """从「网站」列表匹配 project_type=Go 的站点（新版面板 go_project 插件表常为空）。"""
    want = (want_name or "").strip()
    want_l = want.lower()
    if not want:
        return None, None

    def _scan_rows(rows):
        for row in rows:
            if not isinstance(row, dict):
                continue
            if row.get("project_type") != "Go":
                continue
            n = row.get("name") or row.get("ps") or ""
            path = (row.get("path") or "").lower()
            rid = row.get("id")
            if rid is None:
                continue
            if n == want or (isinstance(n, str) and n.strip().lower() == want_l):
                return str(rid), (n or want)
            if "soul-api" in path and (want_l in path or want_l in (n or "").lower()):
                return str(rid), (n or want)
        return None, None

    for search in (want, ""):
        r = _bt_signed_post(
            base,
            key,
            "/data?action=getData&table=sites",
            {"p": "1", "limit": "500", "search": search, "type": "-1"},
        )
        j = _bt_parse_json_response(r)
        rows = j.get("data") if isinstance(j, dict) else None
        if isinstance(rows, list):
            hit = _scan_rows(rows)
            if hit[0]:
                return hit
    return None, None


def _bt_restart_go_via_site_api(base, key, site_id, site_name):
    """/site?action=SiteStop / SiteStart（与面板「网站」一致）。"""
    print("  [宝塔API] site SiteStop/SiteStart (id=%s) …" % site_id)
    j = _bt_parse_json_response(
        _bt_signed_post(
            base,
            key,
            "/site?action=SiteStop",
            {"id": str(site_id), "name": site_name},
        )
    )
    if not isinstance(j, dict) or j.get("status") is not True:
        if isinstance(j, dict) and j.get("msg"):
            print("  [宝塔API] SiteStop: %s" % j.get("msg"))
        return False
    time.sleep(2)
    j2 = _bt_parse_json_response(
        _bt_signed_post(
            base,
            key,
            "/site?action=SiteStart",
            {"id": str(site_id), "name": site_name},
        )
    )
    if isinstance(j2, dict) and j2.get("status") is True:
        print("  [成功] 已通过宝塔 API 重启 Go 站点: %s" % site_name)
        return True
    if isinstance(j2, dict) and j2.get("msg"):
        print("  [宝塔API] SiteStart: %s" % j2.get("msg"))
    return False


def restart_via_bt_api(cfg):
    """通过宝塔 API 重启：优先网站型 Go（SiteStop/SiteStart），失败再试 go_project 插件。"""
    url = cfg.get("bt_panel_url") or ""
    key = cfg.get("bt_api_key") or ""
    name = cfg.get("bt_go_project_name", "soulApi")
    if not url or not key:
        return False
    if not requests:
        print("  [提示] 未安装 requests，无法使用宝塔 API。pip install requests")
        return False
    try:
        base = url.rstrip("/")
        site_id_env = (os.environ.get("BT_GO_SITE_ID") or "").strip()
        if site_id_env:
            sid, snm = site_id_env, name
        else:
            sid, snm = _bt_discover_go_site(base, key, name)
        if sid and _bt_restart_go_via_site_api(base, key, sid, snm or name):
            return True

        # 兜底：go_project 插件（部分旧面板）
        for action in ("stop_go_project", "start_go_project"):
            j = _bt_parse_json_response(
                _bt_signed_post(
                    base,
                    key,
                    "/plugin?name=go_project",
                    {"action": action, "project_name": name, "name": name},
                )
            )
            if action == "stop_go_project":
                time.sleep(2)
            if isinstance(j, dict) and j.get("status") is False and j.get("msg"):
                print("  [宝塔API] %s: %s" % (action, j.get("msg", "")))
        j = _bt_parse_json_response(
            _bt_signed_post(
                base,
                key,
                "/plugin?name=go_project",
                {"action": "start_go_project", "project_name": name, "name": name},
            )
        )
        if isinstance(j, dict) and j.get("status") is True:
            print("  [成功] 已通过宝塔 API 重启 Go 项目(插件): %s" % name)
            return True
        return False
    except Exception as e:
        print("  [宝塔API 失败] %s" % str(e))
        return False


def bt_panel_ping(cfg):
    """
    与 scripts/deploy_kr_btapi_verify.py 一致：GetDiskInfo 校验面板 API 密钥与可达性。
    返回 True=通过；无 url/key 或未装 requests 时返回 True（不阻断，由后续重启阶段再报错）。
    """
    if not requests:
        return True
    url = (cfg.get("bt_panel_url") or "").rstrip("/")
    key = cfg.get("bt_api_key") or ""
    if not url or not key:
        return True
    req_time = int(time.time())
    sk_md5 = hashlib.md5(key.encode()).hexdigest()
    req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
    try:
        r = requests.post(
            url + "/system?action=GetDiskInfo",
            data={"request_time": req_time, "request_token": req_token},
            timeout=20,
            verify=False,
        )
        if r.status_code != 200:
            print("[宝塔API 探活] HTTP %s" % r.status_code)
            return False
        ct = (r.headers.get("content-type") or "").lower()
        j = {}
        if "json" in ct:
            try:
                j = r.json()
            except Exception:
                j = {}
        else:
            t = (r.text or "").lstrip()
            if t.startswith("{"):
                try:
                    j = json.loads(r.text)
                except Exception:
                    j = {}
        if isinstance(j, dict) and j.get("status") is False:
            print("[宝塔API 探活] %s" % (j.get("msg") or j))
            return False
        print("[宝塔API 探活] 成功（GetDiskInfo）", flush=True)
        return True
    except Exception as e:
        print("[宝塔API 探活] 异常: %s" % e)
        return False


def verify_public_api_only(base_api=None):
    """公网只读冒烟（仅 soul-api），与 deploy_kr_btapi_verify 中 API 三项一致。"""
    if not requests:
        print("  [跳过] --verify 需要 requests")
        return False
    base = (base_api or os.environ.get("VERIFY_API_BASE", "https://soulapi.quwanzhi.com")).rstrip("/")
    checks = [
        ("GET %s/health" % base, "%s/health" % base, lambda r: r.status_code == 200 and '"status"' in r.text),
        ("GET %s/api/book/parts" % base, "%s/api/book/parts" % base, lambda r: r.status_code == 200),
        ("GET %s/api/config" % base, "%s/api/config" % base, lambda r: r.status_code == 200),
    ]
    print("")
    print("=" * 60)
    print("  线上冒烟验证（仅 API）")
    print("=" * 60)
    all_ok = True
    for name, url, pred in checks:
        try:
            r = requests.get(url, timeout=20, verify=True)
            ok = pred(r)
            print("  [%s] %s" % ("OK" if ok else "FAIL", name))
            if not ok:
                print("      status=%s len=%s" % (r.status_code, len(r.text or "")))
                all_ok = False
        except Exception as ex:
            print("  [FAIL] %s — %s" % (name, ex))
            all_ok = False
    return all_ok


# ==================== 宝塔 API：上传 / 解压 / 删文件（无 SSH）====================


def _bt_upload_headers():
    return {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"}


def _bt_parse_upload_response(r):
    """解析 /files?action=upload 的返回：成功 dict、进行中为纯数字偏移、失败 dict。"""
    if r is None or r.status_code != 200:
        return "err", None, "HTTP %s" % (r.status_code if r else "?")
    raw = (r.text or "").strip()
    if raw.isdigit():
        return "progress", int(raw), None
    j = _bt_parse_json_response(r)
    if isinstance(j, dict):
        if j.get("status") is True:
            return "ok", None, None
        return "err", None, j.get("msg") or str(j)
    return "err", None, raw[:300] if raw else "empty body"


def bt_upload_file_resumable(cfg, local_path, remote_dir, remote_name):
    """
    宝塔 /files?action=upload（multipart：blob + f_path/f_name/f_size/f_start + 签名）。
    大文件按块续传（与面板逻辑一致：f_start 须等于服务器上 .upload.tmp 当前大小）。
    """
    if not requests:
        print("  [失败] 需要 requests：pip install requests")
        return False
    url = (cfg.get("bt_panel_url") or "").rstrip("/")
    key = cfg.get("bt_api_key") or ""
    if not url or not key:
        print("  [失败] 缺少 BT_PANEL_URL 或 BT_API_KEY")
        return False
    total = os.path.getsize(local_path)
    chunk_mb = int(os.environ.get("BT_UPLOAD_CHUNK_MB", "4"))
    chunk_size = max(1, chunk_mb) * 1024 * 1024
    f_start = 0
    last_start = -1
    upload_url = url + "/files?action=upload"
    headers = _bt_upload_headers()
    print("  [宝塔API] 上传 %s → %s/%s（共 %.2f MB，每块 %d MB）…" % (
        os.path.basename(local_path), remote_dir, remote_name, total / 1024 / 1024, chunk_mb))
    try:
        with open(local_path, "rb") as fp:
            while f_start < total:
                if f_start == last_start:
                    print("  [失败] 上传停滞 offset=%s（请删服务器上同名 .upload.tmp 后重试）" % f_start)
                    return False
                last_start = f_start
                fp.seek(f_start)
                buf = fp.read(min(chunk_size, total - f_start))
                if not buf:
                    print("  [失败] 读取本地文件失败 offset=%s" % f_start)
                    return False
                req_time = int(time.time())
                sk_md5 = hashlib.md5(key.encode()).hexdigest()
                req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
                data = {
                    "request_time": str(req_time),
                    "request_token": req_token,
                    "f_path": remote_dir,
                    "f_name": remote_name,
                    "f_size": str(total),
                    "f_start": str(f_start),
                }
                files = {"blob": ("blob", buf, "application/octet-stream")}
                r = requests.post(
                    upload_url,
                    data=data,
                    files=files,
                    headers=headers,
                    timeout=600,
                    verify=False,
                )
                kind, prog, err = _bt_parse_upload_response(r)
                if kind == "ok":
                    print("  [成功] 宝塔上传完成")
                    return True
                if kind == "progress":
                    if prog <= f_start:
                        print("  [失败] 上传偏移未前进: server=%s local=%s" % (prog, f_start))
                        return False
                    f_start = prog
                    continue
                print("  [失败] 上传接口: %s" % (err or kind))
                return False
        print("  [失败] 上传未正常结束（offset=%s size=%s）" % (f_start, total))
        return False
    except Exception as e:
        print("  [失败] 上传异常: %s" % e)
        return False


def bt_delete_file(cfg, remote_path):
    """宝塔 /files?action=DeleteFile"""
    url = (cfg.get("bt_panel_url") or "").rstrip("/")
    key = cfg.get("bt_api_key") or ""
    if not url or not key or not remote_path:
        return False
    j = _bt_parse_json_response(
        _bt_signed_post(url, key, "/files?action=DeleteFile", {"path": remote_path})
    )
    if isinstance(j, dict) and j.get("status") is True:
        return True
    if isinstance(j, dict) and j.get("msg"):
        print("  [宝塔API] DeleteFile: %s" % j.get("msg"))
    return False


def bt_unzip_remote(cfg, sfile, dfile):
    """宝塔 /files?action=UnZip（sfile=压缩包绝对路径，dfile=解压目标目录）。"""
    url = (cfg.get("bt_panel_url") or "").rstrip("/")
    key = cfg.get("bt_api_key") or ""
    coding = "UTF-8"
    for type1 in ("tar.gz", "tgz", "tar"):
        j = _bt_parse_json_response(
            _bt_signed_post(
                url,
                key,
                "/files?action=UnZip",
                {"sfile": sfile, "dfile": dfile, "type1": type1, "coding": coding},
                timeout=180,
            )
        )
        if isinstance(j, dict) and j.get("status") is True:
            print("  [成功] 宝塔解压完成 (type1=%s)" % type1)
            return True
        if isinstance(j, dict) and j.get("msg"):
            print("  [宝塔API] UnZip type1=%s: %s" % (type1, j.get("msg")))
    return False


def upload_and_extract_via_btapi(cfg, tarball_path, no_restart=False, restart_method="auto"):
    """仅宝塔 API：上传 tar → 解压到项目目录 → 删 tar → Site 启停。"""
    print("[3/4] 宝塔 API 上传并解压 …")
    if not requests:
        print("  [失败] 需要 requests：pip install requests")
        return False
    project_path = cfg["project_path"].rstrip("/")
    tar_name = "soul_api_deploy.tar.gz"
    remote_tar = project_path + "/" + tar_name

    if not bt_upload_file_resumable(cfg, tarball_path, project_path, tar_name):
        return False

    ddir = project_path.rstrip("/") + "/"
    print("  [宝塔API] 解压 %s → %s …" % (remote_tar, ddir))
    if not bt_unzip_remote(cfg, remote_tar, ddir):
        print("  [提示] 解压失败时可检查面板「文件」中是否存在同名目录占用或 type 不支持")
        return False

    if bt_delete_file(cfg, remote_tar):
        print("  [已清理] 远程 %s" % remote_tar)
    else:
        print("  [提示] 未能删除远程压缩包，可在面板手动删: %s" % remote_tar)

    if not no_restart:
        print("[4/4] 重启 soulApi（宝塔 Site API）…")
        if restart_method == "ssh":
            print("  [提示] --deploy-method btapi 时不支持仅 SSH 重启；已改为宝塔 API")
            restart_method = "btapi"
        ok = False
        if restart_method in ("auto", "btapi"):
            ok = restart_via_bt_api(cfg)
            if ok and restart_method == "auto":
                print("  [宝塔API] 第二轮 SiteStop/SiteStart（无 SSH 时提高换新进程概率）…")
                time.sleep(4)
                ok = restart_via_bt_api(cfg) or ok
        if not ok:
            print("  [失败] 宝塔重启未成功，请到面板「网站」手动启停 Go 站点或核对 BT_GO_SITE_ID / API 白名单")
            return False
    else:
        print("[4/4] 跳过重启 (--no-restart)")

    print("  [成功] 已解压到: %s" % project_path)
    return True


# ==================== SSH 上传（可选，仅 --deploy-method ssh）====================


def _connect_ssh(cfg):
    """建立 SSH 连接，启用 keepalive 防大文件上传时断连"""
    if paramiko is None:
        raise RuntimeError("未安装 paramiko，无法使用 SSH 部署：pip install paramiko")
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    if cfg.get("ssh_key") and os.path.isfile(cfg["ssh_key"]):
        client.connect(
            cfg["host"], port=DEFAULT_SSH_PORT,
            username=cfg["user"], key_filename=cfg["ssh_key"],
            timeout=15,
        )
    else:
        client.connect(
            cfg["host"], port=DEFAULT_SSH_PORT,
            username=cfg["user"], password=cfg["password"],
            timeout=15,
        )
    transport = client.get_transport()
    if transport:
        transport.set_keepalive(15)
    return client


def _get_port_pids(client, port):
    """读取监听端口的进程 PID 集合（用于判断是否真正重启）。"""
    try:
        cmd = "ss -lntp 2>/dev/null | grep ':%d ' || true" % port
        stdin, stdout, stderr = client.exec_command(cmd, timeout=15)
        out = stdout.read().decode("utf-8", errors="replace")
        pids = set()
        for part in out.split("pid=")[1:]:
            num = ""
            for ch in part:
                if ch.isdigit():
                    num += ch
                else:
                    break
            if num:
                pids.add(num)
        return pids
    except Exception:
        return set()


def upload_and_extract_via_ssh(cfg, tarball_path, no_restart=False, restart_method="auto"):
    """SSH/SFTP 上传 tar 并远程解压（仅当服务器允许密码或公钥登录时使用）。"""
    print("[3/4] SSH 上传并解压 ...")
    if not cfg.get("password") and not cfg.get("ssh_key"):
        print("  [失败] 请设置 DEPLOY_PASSWORD 或 DEPLOY_SSH_KEY")
        return False
    remote_tar = "/tmp/soul_api_deploy.tar.gz"
    project_path = cfg["project_path"]
    client = None
    try:
        # SFTP 上传易因网络抖动 EOF，失败时重连并重试最多 3 次
        for attempt in range(1, 4):
            try:
                if client:
                    try:
                        client.close()
                    except Exception:
                        pass
                client = _connect_ssh(cfg)
                sftp = client.open_sftp()
                sftp.put(tarball_path, remote_tar)
                sftp.close()
                break
            except (EOFError, ConnectionResetError, OSError) as e:
                if attempt < 3:
                    print("  [重试 %d/3] 上传中断: %s，5 秒后重连 ..." % (attempt, e))
                    time.sleep(5)
                else:
                    raise

        cmd = (
            "mkdir -p %s && cd %s && tar -xzf %s && "
            "chmod +x soul-api && rm -f %s && echo OK"
        ) % (project_path, project_path, remote_tar, remote_tar)
        stdin, stdout, stderr = client.exec_command(cmd, timeout=120)
        ex_err = []

        def _drain_tar_stderr():
            try:
                ex_err.append(stderr.read().decode("utf-8", errors="replace"))
            except Exception:
                ex_err.append("")

        t_tar = threading.Thread(target=_drain_tar_stderr)
        t_tar.daemon = True
        t_tar.start()
        out = stdout.read().decode("utf-8", errors="replace").strip()
        t_tar.join(timeout=10)
        exit_status = stdout.channel.recv_exit_status()
        if exit_status != 0 or "OK" not in out:
            print("  [失败] 解压失败，退出码:", exit_status)
            return False
        print("  [成功] 已解压到: %s" % project_path)

        if not no_restart:
            print("[4/4] 重启 soulApi 服务 ...")
            ok = False
            pids_before = _get_port_pids(client, DEPLOY_PORT)
            if restart_method in ("auto", "btapi") and (cfg.get("bt_panel_url") and cfg.get("bt_api_key")):
                ok = restart_via_bt_api(cfg)
                # 宝塔接口有时返回成功但进程未真正重启：若 PID 未变化，视为 btapi 未生效
                if ok:
                    time.sleep(2)
                    pids_after = _get_port_pids(client, DEPLOY_PORT)
                    if pids_before and pids_after and pids_before == pids_after:
                        print("  [宝塔API] 检测到监听 %d 的 PID 未变化（%s），判定 btapi 未真正重启，转 SSH 兜底。" % (DEPLOY_PORT, ",".join(sorted(pids_after))))
                        ok = False
            if not ok and restart_method in ("auto", "ssh"):
                # SSH：正式环境固定监听 DEPLOY_PORT（默认 8080）。用 fuser 释放端口，避免宝塔守护
                # 启动的进程 cwd 与项目目录不一致导致 pgrep+cwd 校验永远失败。
                # 拆成两次 exec：先短命令起进程，本机 sleep 后再 curl，避免单条远程命令+管道偶发拖死 Paramiko。
                start_cmd = (
                    "cd %s && (fuser -k %d/tcp 2>/dev/null || true) && sleep 2 && "
                    "( setsid nohup ./soul-api >> soul-api.log 2>&1 </dev/null & ) && "
                    "sleep 1 && echo START_OK"
                ) % (project_path, DEPLOY_PORT)
                stdin, stdout, stderr = client.exec_command(
                    "timeout 45 bash -c " + shlex.quote(start_cmd),
                    timeout=55,
                    get_pty=True,
                )
                start_out = stdout.read().decode("utf-8", errors="replace").strip()
                if "START_OK" not in start_out:
                    print("  [stderr] 起进程输出: %s" % start_out[:300])
                time.sleep(12)
                health_cmd = (
                    "curl -sf --connect-timeout 5 --max-time 15 "
                    "http://127.0.0.1:%d/health 2>/dev/null | grep -q '\"status\"' "
                    "&& echo RESTART_OK || echo RESTART_FAIL"
                ) % DEPLOY_PORT
                stdin, stdout, stderr = client.exec_command(
                    "timeout 25 bash -c " + shlex.quote(health_cmd),
                    timeout=35,
                    get_pty=True,
                )
                out = stdout.read().decode("utf-8", errors="replace").strip()
                ok = "RESTART_OK" in out
                if ok:
                    print("  [成功] soulApi 已通过 SSH 重启")
                else:
                    print("  [警告] SSH 重启状态未知，请到宝塔 Go 项目里手动点击启动，或执行: cd %s && ./soul-api" % project_path)
            if restart_method == "btapi" and not ok:
                print("  [失败] 已指定 --restart-method btapi，但宝塔 API 重启未成功（请核对 API 白名单含本机出口 IP、BT_GO_PROJECT_NAME/BT_GO_SITE_ID）")
                return False
        else:
            print("[4/4] 跳过重启 (--no-restart)")

        return True
    except Exception as e:
        err_msg = str(e) or repr(e) or type(e).__name__
        print("  [失败] SSH 错误:", err_msg)
        import traceback
        traceback.print_exc()
        return False
    finally:
        if client:
            try:
                client.close()
            except Exception:
                pass


def upload_and_extract(cfg, tarball_path, no_restart=False, restart_method="auto", deploy_method="btapi"):
    """
    deploy_method:
      btapi — 仅宝塔 API（上传/解压/删包/启停），不要求 SSH。
      ssh   — 旧逻辑：SFTP + 远程 tar（需 paramiko 且服务器允许对应认证方式）。
    """
    dm = (deploy_method or "btapi").strip().lower()
    if dm == "ssh":
        return upload_and_extract_via_ssh(cfg, tarball_path, no_restart=no_restart, restart_method=restart_method)
    if dm != "btapi":
        print("  [失败] 未知 --deploy-method: %s（仅支持 btapi / ssh）" % deploy_method)
        return False
    return upload_and_extract_via_btapi(cfg, tarball_path, no_restart=no_restart, restart_method=restart_method)


# ==================== 主函数 ====================


def main():
    parser = argparse.ArgumentParser(
        description="soulApi（soul-api 后端）Go 项目一键部署到宝塔",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--no-build", action="store_true", help="跳过本地编译（使用已有 soul-api 二进制）")
    parser.add_argument("--no-env", action="store_true", help="不打包 .env（保留服务器现有 .env）")
    parser.add_argument("--no-restart", action="store_true", help="上传后不重启服务")
    parser.add_argument(
        "--restart-method",
        choices=("auto", "btapi", "ssh"),
        default="auto",
        help="重启方式: auto=先试宝塔API再SSH, btapi=仅宝塔API, ssh=仅SSH (默认 auto)",
    )
    parser.add_argument(
        "--skip-bt-ping",
        action="store_true",
        help="跳过宝塔面板 API 探活（与 deploy_kr_btapi_verify.py --skip-bt-ping 一致）",
    )
    parser.add_argument(
        "--verify",
        action="store_true",
        help="部署成功后对 VERIFY_API_BASE（默认 https://soulapi.quwanzhi.com）做 health/book/config 冒烟",
    )
    _deploy_default = (os.environ.get("DEPLOY_METHOD") or "btapi").strip().lower()
    if _deploy_default not in ("btapi", "ssh"):
        _deploy_default = "btapi"
    parser.add_argument(
        "--deploy-method",
        choices=("btapi", "ssh"),
        default=_deploy_default,
        help="btapi=仅宝塔API上传/解压/重启（默认）；ssh=SFTP+远程命令（需 SSH 可用）",
    )
    args = parser.parse_args()

    script_dir = os.path.dirname(os.path.abspath(__file__))
    root = script_dir

    cfg = get_cfg()
    print("=" * 60)
    print("  soulApi（soul-api）一键部署到宝塔")
    print("=" * 60)
    print("  部署方式: %s" % args.deploy_method)
    if args.deploy_method == "ssh":
        print("  服务器: %s@%s:%s" % (cfg["user"], cfg["host"], DEFAULT_SSH_PORT))
    print("  目标目录: %s" % cfg["project_path"])
    site_hint = (os.environ.get("BT_GO_SITE_ID") or "").strip()
    if site_hint:
        print("  宝塔站点: BT_GO_SITE_ID=%s, BT_GO_PROJECT_NAME=%s" % (site_hint, cfg.get("bt_go_project_name", "")))
    else:
        print("  宝塔站点: 自动匹配 Go 站点名 %s（可设 BT_GO_SITE_ID）" % cfg.get("bt_go_project_name", ""))
    print("=" * 60)

    if not args.skip_bt_ping:
        if not bt_panel_ping(cfg):
            print("[失败] 宝塔 API 探活未通过，已中止（可用 --skip-bt-ping 跳过）")
            return 1

    binary_path = os.path.join(root, "soul-api")
    if not args.no_build:
        p = run_build(root)
        if not p:
            return 1
    else:
        if not os.path.isfile(binary_path):
            print("[错误] 未找到 soul-api 二进制，请先编译或去掉 --no-build")
            return 1
        print("[1/4] 跳过编译，使用现有 soul-api")

    tarball = pack_deploy(root, binary_path, include_env=not args.no_env)
    if not tarball:
        return 1

    if not upload_and_extract(
        cfg,
        tarball,
        no_restart=args.no_restart,
        restart_method=args.restart_method,
        deploy_method=args.deploy_method,
    ):
        return 1

    try:
        os.remove(tarball)
    except Exception:
        pass

    print("")
    print("  部署完成！目录: %s" % cfg["project_path"])

    if args.verify:
        time.sleep(2)
        if not verify_public_api_only():
            print("")
            print("[失败] 线上 API 冒烟存在未通过项（部署已执行，请排查 Nginx/证书/业务）")
            return 1
        print("")
        print("  冒烟通过（仅 API）。")

    return 0


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