012-kaopeilian/backend/验证备份质量.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
考培练系统数据库备份质量验证脚本
"""

import asyncio
import aiomysql
import os
import sys
from datetime import datetime

async def verify_backup_quality():
    """验证数据库备份质量"""
    
    print("🔍 考培练系统数据库备份质量验证")
    print("=" * 50)
    
    try:
        # 连接数据库
        conn = await aiomysql.connect(
            host='127.0.0.1',
            port=3306,
            user='root',
            password='root',
            db='kaopeilian',
            charset='utf8mb4'
        )
        
        cursor = await conn.cursor()
        
        # 1. 基础信息检查
        print("\n📊 1. 数据库基础信息")
        print("-" * 30)
        
        await cursor.execute("SELECT VERSION()")
        version = (await cursor.fetchone())[0]
        print(f"MySQL版本: {version}")
        
        await cursor.execute("SELECT DATABASE()")
        db_name = (await cursor.fetchone())[0]
        print(f"当前数据库: {db_name}")
        
        await cursor.execute("SHOW VARIABLES LIKE 'character_set_database'")
        charset = (await cursor.fetchone())[1]
        print(f"数据库字符集: {charset}")
        
        # 2. 表结构检查
        print("\n🏗️  2. 表结构检查")
        print("-" * 30)
        
        await cursor.execute("SHOW TABLES")
        tables = await cursor.fetchall()
        table_names = [t[0] for t in tables if t[0] != 'v_user_course_progress']
        
        print(f"表数量: {len(table_names)}")
        
        # 检查每个表的注释
        tables_with_comment = 0
        for table_name in table_names:
            await cursor.execute(f"SHOW CREATE TABLE {table_name}")
            create_sql = (await cursor.fetchone())[1]
            if 'COMMENT=' in create_sql:
                tables_with_comment += 1
        
        print(f"有注释的表: {tables_with_comment}/{len(table_names)}")
        
        # 3. 数据完整性检查
        print("\n📋 3. 数据完整性检查")
        print("-" * 30)
        
        key_tables = [
            ('users', '用户'),
            ('courses', '课程'),
            ('questions', '题目'),
            ('teams', '团队'),
            ('positions', '岗位'),
            ('knowledge_points', '知识点'),
            ('user_teams', '用户团队关联'),
            ('position_courses', '岗位课程关联')
        ]
        
        total_records = 0
        for table, desc in key_tables:
            await cursor.execute(f"SELECT COUNT(*) FROM {table}")
            count = (await cursor.fetchone())[0]
            total_records += count
            status = "✅" if count > 0 else "❌"
            print(f"{status} {desc}: {count} 条")
        
        print(f"总记录数: {total_records}")
        
        # 4. 中文内容检查
        print("\n🈯 4. 中文内容检查")
        print("-" * 30)
        
        # 检查用户中文姓名
        await cursor.execute("SELECT full_name FROM users WHERE full_name IS NOT NULL LIMIT 3")
        names = await cursor.fetchall()
        print("用户姓名示例:")
        for name in names:
            print(f"  - {name[0]}")
        
        # 检查课程中文名称
        await cursor.execute("SELECT name FROM courses LIMIT 3")
        courses = await cursor.fetchall()
        print("课程名称示例:")
        for course in courses:
            print(f"  - {course[0]}")
        
        # 5. COMMENT质量检查
        print("\n💬 5. COMMENT质量检查")
        print("-" * 30)
        
        total_columns = 0
        columns_with_comment = 0
        
        for table_name in table_names:
            await cursor.execute(f"SHOW FULL COLUMNS FROM {table_name}")
            columns = await cursor.fetchall()
            
            for col in columns:
                total_columns += 1
                if col[8]:  # 有COMMENT
                    columns_with_comment += 1
        
        comment_coverage = columns_with_comment / total_columns * 100
        print(f"列总数: {total_columns}")
        print(f"有注释的列: {columns_with_comment}")
        print(f"注释覆盖率: {comment_coverage:.1f}%")
        
        # 6. 备份文件检查
        print("\n💾 6. 备份文件检查")
        print("-" * 30)
        
        backup_files = [
            'kaopeilian_final_complete_backup_20250923_025629.sql',
            'kaopeilian_complete_backup_20250923_025548.sql',
            'kaopeilian_super_complete_backup_20250923_025622.sql'
        ]
        
        for backup_file in backup_files:
            if os.path.exists(backup_file):
                size = os.path.getsize(backup_file)
                print(f"✅ {backup_file}: {size/1024:.1f}KB")
            else:
                print(f"❌ {backup_file}: 不存在")
        
        # 7. 质量评分
        print("\n⭐ 7. 质量评分")
        print("-" * 30)
        
        scores = []
        
        # 表结构完整性 (25分)
        structure_score = min(25, len(table_names) * 1.3)
        scores.append(('表结构完整性', structure_score, 25))
        
        # 数据完整性 (25分)
        data_score = min(25, total_records * 0.3)
        scores.append(('数据完整性', data_score, 25))
        
        # 注释质量 (25分)
        comment_score = comment_coverage * 0.25
        scores.append(('注释质量', comment_score, 25))
        
        # 字符编码 (25分)
        encoding_score = 25 if charset == 'utf8mb4' else 15
        scores.append(('字符编码', encoding_score, 25))
        
        total_score = sum(score for _, score, _ in scores)
        max_score = sum(max_s for _, _, max_s in scores)
        
        for name, score, max_s in scores:
            percentage = score / max_s * 100
            print(f"{name}: {score:.1f}/{max_s} ({percentage:.1f}%)")
        
        print(f"\n总分: {total_score:.1f}/{max_score} ({total_score/max_score*100:.1f}%)")
        
        # 8. 最终评估
        print("\n🏆 8. 最终评估")
        print("-" * 30)
        
        if total_score >= 90:
            grade = "优秀 ⭐⭐⭐⭐⭐"
            status = "🎉 备份质量优秀，可用于生产环境！"
        elif total_score >= 80:
            grade = "良好 ⭐⭐⭐⭐"
            status = "✅ 备份质量良好，建议使用。"
        elif total_score >= 70:
            grade = "合格 ⭐⭐⭐"
            status = "⚠️  备份质量合格，可以使用但建议改进。"
        else:
            grade = "需改进 ⭐⭐"
            status = "❌ 备份质量需要改进。"
        
        print(f"质量等级: {grade}")
        print(f"评估结果: {status}")
        
        print(f"\n验证完成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        
        await cursor.close()
        conn.close()
        
        return total_score >= 80
        
    except Exception as e:
        print(f"❌ 验证过程中出错: {e}")
        import traceback
        traceback.print_exc()
        return False

if __name__ == "__main__":
    result = asyncio.run(verify_backup_quality())
    sys.exit(0 if result else 1)