server-core/scripts/fix_prod_db.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""直接在生产数据库创建 compute_series 和 compute_icons 表，并更新 alembic 版本。"""
import sys

# 生产数据库连接配置
PROD_HOST = "124.221.143.140"
PORT = 8091
USER = "opc"
PASSWORD = "jjjsgysyujkwjwgb"
DB_NAME = "opc"

PROD_URL = f"mysql+pymysql://{USER}:{PASSWORD}@{PROD_HOST}:{PORT}/{DB_NAME}?charset=utf8mb4"

from sqlalchemy import create_engine, text, inspect

print(f"连接生产数据库: {PROD_HOST}:{PORT}/{DB_NAME}")
print()

engine = create_engine(PROD_URL, pool_pre_ping=True, connect_args={'connect_timeout': 10})

try:
    # 1. 创建 compute_series 表
    print("1. 创建 compute_series 表...")
    with engine.begin() as conn:
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS compute_series (
                id VARCHAR(64) PRIMARY KEY,
                name VARCHAR(128) NOT NULL,
                display_name VARCHAR(256) DEFAULT '',
                supplier VARCHAR(128) DEFAULT '',
                icon VARCHAR(512) DEFAULT '',
                icon_file VARCHAR(512) DEFAULT '',
                description TEXT,
                sort_order INTEGER DEFAULT 0,
                status VARCHAR(32) DEFAULT 'active',
                created_at VARCHAR(32) DEFAULT '',
                updated_at VARCHAR(32) DEFAULT '',
                UNIQUE KEY uq_compute_series_name (name),
                KEY idx_compute_series_name (name)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """))
    print("   ✓ compute_series 表创建成功")

    # 2. 创建 compute_icons 表
    print("2. 创建 compute_icons 表...")
    with engine.begin() as conn:
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS compute_icons (
                id VARCHAR(64) PRIMARY KEY,
                name VARCHAR(128) NOT NULL,
                category VARCHAR(32) DEFAULT 'model',
                url VARCHAR(512) DEFAULT '',
                file VARCHAR(512) DEFAULT '',
                file_size INTEGER DEFAULT 0,
                mime_type VARCHAR(64) DEFAULT '',
                width INTEGER DEFAULT 0,
                height INTEGER DEFAULT 0,
                description TEXT,
                usage_count INTEGER DEFAULT 0,
                status VARCHAR(32) DEFAULT 'active',
                created_at VARCHAR(32) DEFAULT '',
                updated_at VARCHAR(32) DEFAULT '',
                KEY idx_compute_icons_name (name),
                KEY idx_compute_icons_category (category)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """))
    print("   ✓ compute_icons 表创建成功")

    # 3. 更新 alembic_version 表
    print("3. 更新 alembic_version 到 0078...")
    with engine.begin() as conn:
        # 检查 alembic_version 表是否存在
        result = conn.execute(text("SHOW TABLES LIKE 'alembic_version'"))
        if result.fetchone():
            # 更新版本号
            conn.execute(text("UPDATE alembic_version SET version_num = '0078'"))
            print("   ✓ alembic_version 已更新到 0078")
        else:
            # 创建表并插入版本号
            conn.execute(text("""
                CREATE TABLE alembic_version (
                    version_num VARCHAR(32) NOT NULL,
                    PRIMARY KEY (version_num)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """))
            conn.execute(text("INSERT INTO alembic_version (version_num) VALUES ('0078')"))
            print("   ✓ alembic_version 表已创建并设置为 0078")

    # 4. 验证表是否创建成功
    print()
    print("4. 验证表结构...")
    inspector = inspect(engine)
    tables = inspector.get_table_names()
    
    for table_name in ['compute_series', 'compute_icons']:
        if table_name in tables:
            cols = inspector.get_columns(table_name)
            print(f"   ✓ {table_name}: {len(cols)} 列")
            for col in cols:
                print(f"     - {col['name']}: {col['type']}")
        else:
            print(f"   ✗ {table_name}: 不存在！")

    # 5. 验证 alembic 版本
    print()
    print("5. 验证 alembic 版本...")
    with engine.connect() as conn:
        result = conn.execute(text("SELECT version_num FROM alembic_version"))
        version = result.fetchone()
        if version:
            print(f"   ✓ 当前 alembic 版本: {version[0]}")
        else:
            print("   ✗ alembic_version 表为空")

    print()
    print("=" * 60)
    print("✅ 生产数据库修复完成！")
    print("=" * 60)

except Exception as e:
    print(f"❌ 错误: {e}")
    import traceback
    traceback.print_exc()
    sys.exit(1)
finally:
    engine.dispose()