happy-life-star/tools/download-server-log.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
服务器日志下载脚本

从生产服务器下载后端日志到本地 logs/server/ 目录

使用方法:
  python tools/download-server-log.py

Author: Peanut
Created: 2026-06-03
Purpose: 从服务器下载 emotion-single.log 到本地，方便排查问题
"""

import shutil
import subprocess
import sys
from pathlib import Path

# ============================================================================
# 配置
# ============================================================================
SERVER_IP = "101.200.208.45"
USERNAME = "root"
REMOTE_LOG = "/data/logs/emotion-museum/emotion-single.log"

# 本地路径（项目根目录下的 logs/server/）
PROJECT_DIR = Path(__file__).parent.parent.absolute()
LOCAL_LOG_DIR = PROJECT_DIR / "logs" / "server"
LOCAL_LOG = LOCAL_LOG_DIR / "emotion-single.log"
BACKUP_LOG = LOCAL_LOG_DIR / "emotion-single.log.bak"


# ============================================================================
# 日志输出
# ============================================================================
def log_info(msg):
    print(f"\033[32m[INFO]\033[0m {msg}")


def log_warn(msg):
    print(f"\033[33m[WARN]\033[0m {msg}")


def log_error(msg):
    print(f"\033[31m[ERROR]\033[0m {msg}")


# ============================================================================
# 核心逻辑
# ============================================================================
def check_ssh():
    """检查 SSH 连接是否正常"""
    try:
        result = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
             f"{USERNAME}@{SERVER_IP}", "echo ok"],
            capture_output=True, text=True, timeout=15
        )
        if result.returncode == 0:
            log_info("SSH 连接正常")
            return True
        else:
            log_error(f"SSH 连接失败: {result.stderr.strip()}")
            log_error(f"请确认已配置免密登录: ssh {USERNAME}@{SERVER_IP}")
            return False
    except subprocess.TimeoutExpired:
        log_error("SSH 连接超时")
        return False
    except FileNotFoundError:
        log_error("未找到 ssh 命令，请确认已安装 OpenSSH 客户端")
        return False


def get_remote_file_size():
    """获取远程日志文件大小（字节）"""
    try:
        result = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
             f"{USERNAME}@{SERVER_IP}", f"stat -c %s {REMOTE_LOG} 2>/dev/null || echo -1"],
            capture_output=True, text=True, timeout=15
        )
        size = int(result.stdout.strip())
        return size if size > 0 else -1
    except (subprocess.TimeoutExpired, ValueError, OSError) as e:
        log_error(f"获取远程文件大小失败: {e}")
        return -1


def format_size(size_bytes):
    """格式化文件大小"""
    if size_bytes < 0:
        return "未知"
    for unit in ["B", "KB", "MB", "GB"]:
        if size_bytes < 1024:
            return f"{size_bytes:.1f} {unit}"
        size_bytes /= 1024
    return f"{size_bytes:.1f} TB"


def backup_local_log():
    """备份本地旧日志文件"""
    if LOCAL_LOG.exists():
        shutil.copy2(LOCAL_LOG, BACKUP_LOG)
        log_info(f"已备份旧日志: {BACKUP_LOG}")
    else:
        log_info("本地无旧日志，跳过备份")


def download_log():
    """下载远程日志文件"""
    LOCAL_LOG_DIR.mkdir(parents=True, exist_ok=True)
    log_info(f"本地目录: {LOCAL_LOG_DIR}")

    backup_local_log()

    log_info(f"正在下载: {USERNAME}@{SERVER_IP}:{REMOTE_LOG}")
    log_info(f"        -> {LOCAL_LOG}")

    try:
        result = subprocess.run(
            ["scp", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
             f"{USERNAME}@{SERVER_IP}:{REMOTE_LOG}", str(LOCAL_LOG)],
            capture_output=True, text=True, timeout=300
        )
        if result.returncode == 0:
            local_size = LOCAL_LOG.stat().st_size
            log_info(f"下载成功: {format_size(local_size)}")
            return True
        else:
            log_error(f"下载失败: {result.stderr.strip()}")
            return False
    except subprocess.TimeoutExpired:
        log_error("下载超时（300 秒）")
        return False
    except Exception as e:
        log_error(f"下载异常: {e}")
        return False


def main():
    log_info("=== 情绪博物馆 - 日志下载工具 ===")
    log_info(f"服务器: {USERNAME}@{SERVER_IP}")
    log_info(f"远程日志: {REMOTE_LOG}")

    # 检查 SSH
    if not check_ssh():
        sys.exit(1)

    # 检查远程文件
    remote_size = get_remote_file_size()
    if remote_size < 0:
        log_error(f"远程日志文件不存在或无法访问: {REMOTE_LOG}")
        sys.exit(1)
    log_info(f"远程日志大小: {format_size(remote_size)}")

    # 下载
    if download_log():
        log_info("=== 下载完成 ===")
    else:
        log_error("下载失败")
        sys.exit(1)


if __name__ == "__main__":
    main()