aitsc/check_tencent_db.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
腾讯云数据库连接检查脚本
验证数据库连接和配置是否正确
"""

import pymysql
import sys
from datetime import datetime

# 腾讯云数据库配置
DB_CONFIG = {
    'host': 'gz-cynosdbmysql-grp-d26pzce5.sql.tencentcdb.com',
    'port': 24936,
    'user': 'root',
    'password': '!Rjb12191',
    'database': 'pro_db',
    'charset': 'utf8mb4'
}

def check_connection():
    """检查数据库连接"""
    print("🔍 检查腾讯云数据库连接...")
    print(f"主机: {DB_CONFIG['host']}")
    print(f"端口: {DB_CONFIG['port']}")
    print(f"数据库: {DB_CONFIG['database']}")
    print(f"用户: {DB_CONFIG['user']}")
    print("-" * 40)
    
    try:
        connection = pymysql.connect(**DB_CONFIG)
        print("✅ 数据库连接成功")
        
        # 获取数据库信息
        cursor = connection.cursor()
        cursor.execute("SELECT VERSION()")
        version = cursor.fetchone()[0]
        print(f"✅ MySQL版本: {version}")
        
        cursor.execute("SELECT DATABASE()")
        current_db = cursor.fetchone()[0]
        print(f"✅ 当前数据库: {current_db}")
        
        cursor.execute("SELECT USER()")
        current_user = cursor.fetchone()[0]
        print(f"✅ 当前用户: {current_user}")
        
        cursor.close()
        connection.close()
        return True
        
    except Exception as e:
        print(f"❌ 数据库连接失败: {str(e)}")
        return False

def check_existing_tables():
    """检查现有表"""
    print("\n📋 检查现有表结构...")
    
    try:
        connection = pymysql.connect(**DB_CONFIG)
        cursor = connection.cursor()
        
        # 获取所有表
        cursor.execute("SHOW TABLES")
        tables = [table[0] for table in cursor.fetchall()]
        
        print(f"✅ 数据库中共有 {len(tables)} 个表:")
        for table in tables:
            print(f"   - {table}")
        
        # 检查历史相关表
        history_tables = ['prompt_history', 'history_tags', 'user_statistics']
        existing_history_tables = [table for table in history_tables if table in tables]
        
        if existing_history_tables:
            print(f"\n✅ 发现历史相关表: {', '.join(existing_history_tables)}")
            
            # 检查表结构
            for table in existing_history_tables:
                cursor.execute(f"DESCRIBE {table}")
                columns = cursor.fetchall()
                print(f"\n📊 {table} 表结构 ({len(columns)} 列):")
                for col in columns:
                    print(f"   - {col[0]}: {col[1]}")
        else:
            print("\n⚠️  未发现历史相关表，需要创建")
        
        cursor.close()
        connection.close()
        return True
        
    except Exception as e:
        print(f"❌ 检查表结构失败: {str(e)}")
        return False

def check_permissions():
    """检查数据库权限"""
    print("\n🔐 检查数据库权限...")
    
    try:
        connection = pymysql.connect(**DB_CONFIG)
        cursor = connection.cursor()
        
        # 检查创建表权限
        cursor.execute("SHOW GRANTS")
        grants = cursor.fetchall()
        
        print("✅ 用户权限:")
        for grant in grants:
            print(f"   - {grant[0]}")
        
        # 检查是否有CREATE权限
        has_create = any('CREATE' in grant[0] for grant in grants)
        if has_create:
            print("✅ 具有CREATE权限")
        else:
            print("⚠️  可能没有CREATE权限")
        
        cursor.close()
        connection.close()
        return True
        
    except Exception as e:
        print(f"❌ 检查权限失败: {str(e)}")
        return False

def test_basic_operations():
    """测试基本操作"""
    print("\n🧪 测试基本数据库操作...")
    
    try:
        connection = pymysql.connect(**DB_CONFIG)
        cursor = connection.cursor()
        
        # 测试创建临时表
        cursor.execute("""
        CREATE TEMPORARY TABLE test_table (
            id INT PRIMARY KEY,
            name VARCHAR(50)
        )
        """)
        print("✅ 创建临时表成功")
        
        # 测试插入数据
        cursor.execute("INSERT INTO test_table (id, name) VALUES (1, 'test')")
        print("✅ 插入数据成功")
        
        # 测试查询数据
        cursor.execute("SELECT * FROM test_table")
        result = cursor.fetchone()
        print(f"✅ 查询数据成功: {result}")
        
        # 测试删除数据
        cursor.execute("DELETE FROM test_table WHERE id = 1")
        print("✅ 删除数据成功")
        
        cursor.close()
        connection.close()
        return True
        
    except Exception as e:
        print(f"❌ 基本操作测试失败: {str(e)}")
        return False

def main():
    """主函数"""
    print("🔍 腾讯云数据库连接检查")
    print("=" * 50)
    print(f"检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 50)
    
    # 执行检查
    checks = [
        ("数据库连接", check_connection),
        ("现有表结构", check_existing_tables),
        ("数据库权限", check_permissions),
        ("基本操作", test_basic_operations)
    ]
    
    results = []
    for name, check_func in checks:
        print(f"\n{'='*20} {name} {'='*20}")
        result = check_func()
        results.append((name, result))
    
    # 总结
    print("\n" + "=" * 50)
    print("📊 检查结果总结")
    print("=" * 50)
    
    all_passed = True
    for name, result in results:
        status = "✅ 通过" if result else "❌ 失败"
        print(f"{name}: {status}")
        if not result:
            all_passed = False
    
    print("\n" + "=" * 50)
    if all_passed:
        print("🎉 所有检查通过！可以开始部署优化历史功能")
        print("\n📋 下一步操作:")
        print("   1. 运行部署脚本: python3 deploy_tencent_simple.py")
        print("   2. 或运行一键部署: ./quick_deploy_history.sh")
    else:
        print("⚠️  部分检查失败，请解决相关问题后再部署")
        print("\n🔧 常见问题解决:")
        print("   1. 检查网络连接")
        print("   2. 验证数据库配置")
        print("   3. 确认用户权限")
        print("   4. 检查防火墙设置")
    
    print("=" * 50)

if __name__ == "__main__":
    main()