#!/usr/bin/env python3 """ 数据库迁移脚本 连接到远程 MySQL 数据库并执行 SQL 建表语句 """ import mysql.connector from mysql.connector import Error # 数据库配置 DB_CONFIG = { 'host': '8.148.151.56', 'port': 3306, 'user': 'root', 'password': 'reading_platform_pwd', 'database': 'reading_platform', 'charset': 'utf8mb4', 'autocommit': True } # SQL 建表语句 SQL_STATEMENTS = [ # 1. 课程套餐表 """ CREATE TABLE IF NOT EXISTS course_package ( id BIGINT NOT NULL AUTO_INCREMENT, name VARCHAR(255) NOT NULL COMMENT '套餐名称', description TEXT COMMENT '套餐描述', price BIGINT NOT NULL COMMENT '价格（分）', discount_price BIGINT COMMENT '折后价格（分）', discount_type VARCHAR(50) COMMENT '折扣类型', grade_levels VARCHAR(500) COMMENT '适用年级', course_count INT NOT NULL DEFAULT 0 COMMENT '课程数量', status VARCHAR(50) NOT NULL DEFAULT 'DRAFT' COMMENT '状态', submitted_at DATETIME COMMENT '提交时间', submitted_by BIGINT COMMENT '提交人ID', reviewed_at DATETIME COMMENT '审核时间', reviewed_by BIGINT COMMENT '审核人ID', review_comment TEXT COMMENT '审核意见', published_at DATETIME COMMENT '发布时间', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_status (status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='课程套餐表' """, # 2. 套餐课程关联表 """ CREATE TABLE IF NOT EXISTS course_package_course ( id BIGINT NOT NULL AUTO_INCREMENT, package_id BIGINT NOT NULL COMMENT '套餐ID', course_id BIGINT NOT NULL COMMENT '课程ID', grade_level VARCHAR(50) COMMENT '适用年级', sort_order INT NOT NULL DEFAULT 0 COMMENT '排序号', PRIMARY KEY (id), UNIQUE KEY uk_package_course (package_id, course_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='套餐课程关联表' """, # 3. 租户套餐关联表 """ CREATE TABLE IF NOT EXISTS tenant_package ( id BIGINT NOT NULL AUTO_INCREMENT, tenant_id BIGINT NOT NULL COMMENT '租户ID', package_id BIGINT NOT NULL COMMENT '套餐ID', start_date DATE NOT NULL COMMENT '开始日期', end_date DATE NOT NULL COMMENT '结束日期', price_paid BIGINT NOT NULL DEFAULT 0 COMMENT '实付价格', status VARCHAR(50) NOT NULL DEFAULT 'ACTIVE' COMMENT '状态', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_tenant_id (tenant_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='租户套餐关联表' """, # 4. 课程环节表 """ CREATE TABLE IF NOT EXISTS course_lesson ( id BIGINT NOT NULL AUTO_INCREMENT, course_id BIGINT NOT NULL COMMENT '课程ID', lesson_type VARCHAR(50) NOT NULL COMMENT '课程类型', name VARCHAR(255) NOT NULL COMMENT '课程名称', description TEXT COMMENT '课程描述', duration INT COMMENT '时长（分钟）', video_path VARCHAR(500) COMMENT '视频路径', video_name VARCHAR(255) COMMENT '视频名称', ppt_path VARCHAR(500) COMMENT 'PPT路径', ppt_name VARCHAR(255) COMMENT 'PPT名称', pdf_path VARCHAR(500) COMMENT 'PDF路径', pdf_name VARCHAR(255) COMMENT 'PDF名称', objectives TEXT COMMENT '教学目标', preparation TEXT COMMENT '教学准备', extension TEXT COMMENT '教学延伸', reflection TEXT COMMENT '教学反思', assessment_data TEXT COMMENT '评测数据', use_template TINYINT(1) DEFAULT 0 COMMENT '是否使用模板', sort_order INT NOT NULL DEFAULT 0 COMMENT '排序号', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_course_lesson_type (course_id, lesson_type) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='课程环节表' """, # 5. 教学环节表 """ CREATE TABLE IF NOT EXISTS lesson_step ( id BIGINT NOT NULL AUTO_INCREMENT, lesson_id BIGINT NOT NULL COMMENT '课程环节ID', name VARCHAR(255) NOT NULL COMMENT '环节名称', content TEXT COMMENT '环节内容', duration INT NOT NULL DEFAULT 5 COMMENT '时长（分钟）', objective TEXT COMMENT '教学目标', resource_ids TEXT COMMENT '资源ID列表', sort_order INT NOT NULL DEFAULT 0 COMMENT '排序号', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_lesson_id (lesson_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='教学环节表' """, # 6. 环节资源关联表 """ CREATE TABLE IF NOT EXISTS lesson_step_resource ( id BIGINT NOT NULL AUTO_INCREMENT, step_id BIGINT NOT NULL COMMENT '环节ID', resource_id BIGINT NOT NULL COMMENT '资源ID', sort_order INT NOT NULL DEFAULT 0 COMMENT '排序号', PRIMARY KEY (id), KEY idx_step_id (step_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='环节资源关联表' """, # 7. 主题字典表 """ CREATE TABLE IF NOT EXISTS theme ( id BIGINT NOT NULL AUTO_INCREMENT, name VARCHAR(255) NOT NULL COMMENT '主题名称', description TEXT COMMENT '主题描述', sort_order INT NOT NULL DEFAULT 0 COMMENT '排序号', status VARCHAR(50) NOT NULL DEFAULT 'ACTIVE' COMMENT '状态', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_status (status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='主题字典表' """ ] def execute_migration(): """执行数据库迁移""" print("=" * 50) print("开始执行数据库迁移...") print("=" * 50) print(f"数据库: {DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}") print("-" * 50) # 表名列表 table_names = ['course_package', 'course_package_course', 'tenant_package', 'course_lesson', 'lesson_step', 'lesson_step_resource', 'theme'] try: # 连接数据库 print("正在连接数据库...") conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() print("✓ 数据库连接成功\n") # 执行 SQL 语句 success_count = 0 skip_count = 0 error_count = 0 for i, sql in enumerate(SQL_STATEMENTS, 1): table_name = table_names[i - 1] print(f"[{i}/{len(SQL_STATEMENTS)}] 创建表: {table_name}") try: cursor.execute(sql) print(f" ✓ 成功\n") success_count += 1 except Error as e: if "already exists" in str(e): print(f" ⚠ 表已存在，跳过\n") skip_count += 1 else: print(f" ✗ 失败: {e}\n") error_count += 1 # 验证表创建 print("-" * 50) print("验证表创建...") cursor.execute("SHOW TABLES") tables = cursor.fetchall() print(f"✓ 数据库中共有 {len(tables)} 张表") # 检查新创建的表 existing_tables = [t[0] for t in tables] print("\n新创建的表:") for table in table_names: if table in existing_tables: print(f" ✓ {table}") else: print(f" ✗ {table} (未找到)") # 关闭连接 cursor.close() conn.close() # 总结 print("\n" + "=" * 50) print("数据库迁移完成！") print("=" * 50) print(f"成功: {success_count}, 跳过: {skip_count}, 失败: {error_count}") print(f"总计: {len(SQL_STATEMENTS)} 张表需要创建") return success_count > 0 except Error as e: print(f"\n✗ 数据库连接失败: {e}") return False except Exception as e: print(f"\n✗ 执行失败: {e}") import traceback traceback.print_exc() return False if __name__ == "__main__": try: success = execute_migration() exit(0 if success else 1) except KeyboardInterrupt: print("\n\n用户中断") exit(1)