diagnose_update_issue.py

# -*- coding: utf-8 -*-
"""
诊断更新推送问题

检查整个更新流程的每个环节
"""

import os
import sys
import json
import requests
from pathlib import Path


def print_section(title):
    """打印分节标题"""
    print("\n" + "="*60)
    print(f"  {title}")
    print("="*60)


def check_server_running(base_url):
    """检查服务端是否运行"""
    print_section("1. 检查服务端状态")
    
    try:
        response = requests.get(f"{base_url}/health", timeout=5)
        if response.status_code == 200:
            print("✅ 服务端正常运行")
            return True
        else:
            print(f"❌ 服务端响应异常: {response.status_code}")
            return False
    except Exception as e:
        print(f"❌ 无法连接到服务端: {e}")
        return False


def check_updates_directory(server_path=None):
    """检查服务端updates目录"""
    print_section("2. 检查服务端updates目录")
    
    if server_path:
        updates_dir = Path(server_path) / "updates"
    else:
        updates_dir = Path("server/updates")
    
    if not updates_dir.exists():
        print(f"❌ updates目录不存在: {updates_dir}")
        return False
    
    print(f"✅ updates目录存在: {updates_dir}")
    
    # 检查metadata.json
    metadata_file = updates_dir / "metadata.json"
    if not metadata_file.exists():
        print("❌ metadata.json不存在")
        return False
    
    print("✅ metadata.json存在")
    
    # 读取metadata
    try:
        with open(metadata_file, 'r', encoding='utf-8') as f:
            metadata = json.load(f)
        
        updates = metadata.get("updates", [])
        print(f"\n📦 共有 {len(updates)} 个更新包:")
        
        for update in updates:
            version = update.get("version")
            filename = update.get("filename")
            size = update.get("size", 0) / 1024 / 1024
            
            # 检查文件是否存在
            file_path = updates_dir / filename
            exists = "✅" if file_path.exists() else "❌ 文件不存在"
            
            print(f"  - {version}: {filename} ({size:.2f} MB) {exists}")
        
        return len(updates) > 0
        
    except Exception as e:
        print(f"❌ 读取metadata.json失败: {e}")
        return False


def check_update_api(base_url, token=None):
    """检查更新API"""
    print_section("3. 检查更新API")
    
    # 测试检查更新接口
    print("\n测试 GET /api/updates/check ...")
    
    headers = {}
    if token:
        headers['Authorization'] = f'Bearer {token}'
    
    try:
        response = requests.get(
            f"{base_url}/api/updates/check",
            params={
                'client_id': 'test-client-id',
                'current_version': '1.0.0'
            },
            headers=headers,
            timeout=5
        )
        
        print(f"状态码: {response.status_code}")
        
        if response.status_code == 200:
            data = response.json()
            print("✅ API响应正常")
            print(f"响应内容: {json.dumps(data, indent=2, ensure_ascii=False)}")
            return True
        elif response.status_code == 401:
            print("❌ 认证失败 - 需要客户端token")
            return False
        else:
            print(f"❌ API响应异常: {response.text}")
            return False
            
    except Exception as e:
        print(f"❌ API请求失败: {e}")
        return False


def check_websocket_connection(base_url):
    """检查WebSocket连接"""
    print_section("4. 检查WebSocket连接")
    
    ws_url = base_url.replace('http://', 'ws://').replace('https://', 'wss://')
    print(f"WebSocket URL: {ws_url}/ws/test-client-id")
    print("⚠️  需要手动测试WebSocket连接")
    print("   可以使用客户端日志查看连接状态")


def check_client_logs():
    """检查客户端日志"""
    print_section("5. 检查客户端日志")
    
    print("请检查客户端日志中是否有以下信息：")
    print("  1. '📢 收到服务器推送的更新通知'")
    print("  2. '开始自动更新流程...'")
    print("  3. '[1/3] 检查更新...'")
    print("  4. '[2/3] 下载更新包...'")
    print("  5. '下载进度: XX%'")
    print("\n如果没有这些日志，说明：")
    print("  - 客户端未收到推送通知（WebSocket问题）")
    print("  - 或自动更新功能未启用")


def test_upload_api(base_url, username, password):
    """测试上传API"""
    print_section("6. 测试上传API")
    
    # 登录获取token
    print("\n登录获取管理员token...")
    try:
        response = requests.post(
            f"{base_url}/api/auth/login",
            json={"username": username, "password": password},
            timeout=5
        )
        
        if response.status_code != 200:
            print(f"❌ 登录失败: {response.text}")
            return False
        
        token = response.json().get("access_token")
        print("✅ 登录成功")
        
    except Exception as e:
        print(f"❌ 登录失败: {e}")
        return False
    
    # 测试上传接口是否存在
    print("\n测试上传接口...")
    print("⚠️  实际上传需要在GUI中操作")
    print(f"   上传URL: {base_url}/api/updates/upload")
    print(f"   需要管理员token: {token[:20]}...")
    
    return True


def main():
    """主函数"""
    print("="*60)
    print("  更新推送问题诊断工具")
    print("="*60)
    
    # 配置
    base_url = input("\n请输入服务端URL (默认: http://localhost:8000): ").strip()
    if not base_url:
        base_url = "http://localhost:8000"
    
    base_url = base_url.rstrip('/')
    
    # 1. 检查服务端
    if not check_server_running(base_url):
        print("\n❌ 服务端未运行，请先启动服务端")
        return
    
    # 2. 检查updates目录
    server_path = input("\n请输入服务端目录路径 (默认: ./server): ").strip()
    if not server_path:
        server_path = "./server"
    
    has_updates = check_updates_directory(server_path)
    
    # 3. 检查API
    print("\n是否测试更新API? (需要客户端token)")
    test_api = input("输入 y 继续，其他跳过: ").strip().lower()
    
    if test_api == 'y':
        token = input("请输入客户端token: ").strip()
        check_update_api(base_url, token)
    
    # 4. WebSocket
    check_websocket_connection(base_url)
    
    # 5. 客户端日志
    check_client_logs()
    
    # 6. 测试上传
    print("\n是否测试上传API?")
    test_upload = input("输入 y 继续，其他跳过: ").strip().lower()
    
    if test_upload == 'y':
        username = input("管理员用户名 (默认: admin): ").strip() or "admin"
        password = input("管理员密码 (默认: admin123): ").strip() or "admin123"
        test_upload_api(base_url, username, password)
    
    # 总结
    print_section("诊断总结")
    
    print("\n常见问题和解决方案：")
    print("\n1. 如果updates目录中没有更新包文件：")
    print("   → 在服务端GUI中点击'添加更新包'")
    print("   → 选择ZIP文件并填写版本信息")
    print("   → 等待上传完成（会显示进度）")
    print("   → 确认收到'更新包已成功上传到服务端'的提示")
    
    print("\n2. 如果客户端没有收到推送：")
    print("   → 检查客户端是否在线（服务端GUI中查看）")
    print("   → 检查WebSocket连接是否正常")
    print("   → 查看客户端日志是否有连接错误")
    
    print("\n3. 如果客户端收到推送但不下载：")
    print("   → 检查客户端.env中AUTO_UPDATE_ENABLED=True")
    print("   → 检查客户端日志中的错误信息")
    print("   → 确认客户端token有效")
    
    print("\n4. 如果下载失败返回404：")
    print("   → 确认更新包文件在服务端updates目录中")
    print("   → 检查metadata.json中的文件名是否正确")
    print("   → 尝试直接访问下载URL测试")
    
    print("\n" + "="*60)
    print("  诊断完成")
    print("="*60)


if __name__ == "__main__":
    main()