""" gvsdsdk User 表迁移脚本 将 user_main 数据迁移到 gvsdsdk.User + user_sensitive_data 执行前确保已全量备份数据库！ 分阶段执行，每阶段验证后才继续下一阶段。 """ import uuid import sys import MySQLdb # ==================== 配置 ==================== DB_CONFIG = { 'host': 'gvsds.com', 'port': 50030, 'user': 'root', 'passwd': 'sajksh.sdfGH3YUge.wjkd+', 'db': 'xaio_cheng_xu', 'charset': 'utf8mb4', 'connect_timeout': 10, } # UUID 命名空间（固定值，确保可重现） UUID_NAMESPACE = uuid.UUID('6e6d6163-0000-0000-0000-000000000000') # 旧 RBAC 表 → 重命名映射 RBAC_RENAME_MAP = { 'role': 'legacy_role', 'user_role': 'legacy_user_role', 'permission': 'legacy_permission', 'role_permission': 'legacy_role_permission', 'account_permission': 'legacy_account_permission', } # 扩展表 + admin_profile（有 FK 约束指向 user_main.id） EXT_TABLES_WITH_FK = [ 'user_boss', 'user_dashou', 'user_guanshi', 'user_kefu', 'user_shangjia', 'user_shenheguan', 'user_zuzhang', 'admin_profile', ] def get_conn(): return MySQLdb.connect(**DB_CONFIG) def table_exists(cur, table_name): cur.execute(""" SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s """, (DB_CONFIG['db'], table_name)) return cur.fetchone()[0] > 0 def fk_constraint_exists(cur, table_name, ref_table): cur.execute(""" SELECT CONSTRAINT_NAME FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND REFERENCED_TABLE_NAME = %s """, (DB_CONFIG['db'], table_name, ref_table)) return [r[0] for r in cur.fetchall()] def phase1_rename_rbac_tables(): """Phase 1: 重命名旧 RBAC 表，避免与 gvsdsdk PascalCase 表名冲突""" print('\n' + '='*60) print('Phase 1: 重命名旧 RBAC 表') print('='*60) conn = get_conn() cur = conn.cursor() for old_name, new_name in RBAC_RENAME_MAP.items(): if table_exists(cur, old_name) and not table_exists(cur, new_name): sql = f'RENAME TABLE `{old_name}` TO `{new_name}`' print(f' {old_name} -> {new_name}') cur.execute(sql) elif table_exists(cur, new_name): print(f' {new_name} 已存在，跳过') else: print(f' {old_name} 不存在，跳过') conn.commit() cur.close() conn.close() print('Phase 1 完成') def phase2_create_tables(): """Phase 2: 创建 User 表和 user_sensitive_data 表""" print('\n' + '='*60) print('Phase 2: 创建 User 表 + user_sensitive_data 表') print('='*60) conn = get_conn() cur = conn.cursor() # 2.1 创建 User 表 if table_exists(cur, 'User'): print(' User 表已存在，跳过创建') else: print(' 创建 User 表...') cur.execute(""" CREATE TABLE `User` ( `UserUUID` BINARY(16) NOT NULL PRIMARY KEY, `UserName` varchar(256) NOT NULL UNIQUE, `UserPassword` varchar(200) NOT NULL DEFAULT '', `UserPositionStatus` smallint NOT NULL DEFAULT 0, `UserPersonalSignature` varchar(1024) NOT NULL DEFAULT '', `UserEmail` varchar(256) NOT NULL DEFAULT '', `UserEmailVerify` smallint NOT NULL DEFAULT 0, `UserAccountLicense` smallint NOT NULL DEFAULT 1, `UserGender` smallint NOT NULL DEFAULT -1, `UserTele` varchar(16) NULL, `UserCreateTime` datetime(6) NOT NULL, `UserLastLoginDate` datetime(6) NOT NULL, `IsStaff` tinyint(1) NOT NULL DEFAULT 0, `IsSuperuser` tinyint(1) NOT NULL DEFAULT 0, `YonghuID` varchar(7) NULL UNIQUE, `UserAvatar` varchar(500) NULL DEFAULT 'a_long/morentouxiang.jpg', `UserType` varchar(10) NOT NULL DEFAULT 'normal', INDEX `User_UserUUID_0a9d8a_idx` (`UserUUID`), INDEX `User_YonghuID_idx` (`YonghuID`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci """) print(' User 表创建完成') # 2.2 创建 user_sensitive_data 表 if table_exists(cur, 'user_sensitive_data'): print(' user_sensitive_data 表已存在，跳过创建') else: print(' 创建 user_sensitive_data 表...') cur.execute(""" CREATE TABLE `user_sensitive_data` ( `id` bigint NOT NULL AUTO_INCREMENT PRIMARY KEY, `UserUUID` BINARY(16) NOT NULL UNIQUE, `OpenID` varchar(64) NULL UNIQUE, `UnionID` varchar(64) NULL, `Password` varchar(200) NULL, `PhoneVerified` tinyint(1) NOT NULL DEFAULT 0, `Zhifu` varchar(500) NULL, `SKZhanghao` varchar(11) NULL, `IP` varchar(50) NULL, `create_time` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), `update_time` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6), INDEX `user_sensitive_data_OpenID_idx` (`OpenID`), INDEX `user_sensitive_data_UnionID_idx` (`UnionID`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci """) print(' user_sensitive_data 表创建完成') conn.commit() cur.close() conn.close() print('Phase 2 完成') def phase3_migrate_user_data(): """Phase 3: 迁移 user_main 数据到 User + user_sensitive_data""" print('\n' + '='*60) print('Phase 3: 迁移 user_main 数据') print('='*60) conn = get_conn() cur = conn.cursor() # 检查 User 表是否已有数据 cur.execute('SELECT COUNT(*) FROM `User`') existing = cur.fetchone()[0] if existing > 0: print(f' User 表已有 {existing} 行数据，跳过迁移') cur.close() conn.close() return # 获取所有 user_main 数据 cur.execute(""" SELECT id, yonghuid, openid, avatar, phone, password, zhifu, skzhanghao, ip, last_login_time, user_type, create_time, update_time, unionid, shoujihao_renzheng, last_login FROM user_main """) rows = cur.fetchall() print(f' 从 user_main 读取 {len(rows)} 行') # 生成 UUID 并批量插入 user_data = [] sensitive_data = [] for row in rows: (id_, yonghuid, openid, avatar, phone, password, zhifu, skzhanghao, ip, last_login_time, user_type, create_time, update_time, unionid, shoujihao_renzheng, last_login) = row # 生成确定性 UUID user_uuid = uuid.uuid5(UUID_NAMESPACE, yonghuid).bytes # User 表数据 is_superuser = 1 if user_type == 'admin' else 0 is_staff = 1 if user_type in ('admin', 'kefu') else 0 user_name = yonghuid # 用 yonghuid 作为 UserName avatar_val = avatar if avatar else 'a_long/morentouxiang.jpg' last_login = last_login_time or last_login or create_time user_data.append(( user_uuid, user_name, '', # UserPassword 留空 0, # UserPositionStatus '', # UserPersonalSignature '', # UserEmail 0, # UserEmailVerify 1, # UserAccountLicense -1, # UserGender phone, # UserTele create_time, # UserCreateTime last_login, # UserLastLoginDate is_staff, is_superuser, yonghuid, # YonghuID avatar_val, # UserAvatar user_type, # UserType )) # user_sensitive_data 表数据 sensitive_data.append(( user_uuid, openid if openid else None, unionid if unionid else None, password if password else None, 1 if shoujihao_renzheng else 0, zhifu if zhifu else None, skzhanghao if skzhanghao else None, ip if ip else None, )) # 批量插入 User 表 print(f' 插入 {len(user_data)} 行到 User 表...') cur.executemany(""" INSERT INTO `User` ( UserUUID, UserName, UserPassword, UserPositionStatus, UserPersonalSignature, UserEmail, UserEmailVerify, UserAccountLicense, UserGender, UserTele, UserCreateTime, UserLastLoginDate, IsStaff, IsSuperuser, YonghuID, UserAvatar, UserType ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, user_data) # 批量插入 user_sensitive_data 表 print(f' 插入 {len(sensitive_data)} 行到 user_sensitive_data 表...') cur.executemany(""" INSERT INTO user_sensitive_data ( UserUUID, OpenID, UnionID, Password, PhoneVerified, Zhifu, SKZhanghao, IP ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """, sensitive_data) conn.commit() # 验证 cur.execute('SELECT COUNT(*) FROM `User`') user_count = cur.fetchone()[0] cur.execute('SELECT COUNT(*) FROM user_sensitive_data') sd_count = cur.fetchone()[0] print(f' 验证: User={user_count}, UserSensitiveData={sd_count}') if user_count != len(rows): print(f' 警告: User 表行数({user_count}) != user_main 行数({len(rows)})') cur.close() conn.close() print('Phase 3 完成') def phase4_alter_extension_tables(): """Phase 4: 删除 FK 约束，修改扩展表 user_id 列类型 int → BINARY(16)""" print('\n' + '='*60) print('Phase 4: 修改扩展表 FK 列类型') print('='*60) conn = get_conn() cur = conn.cursor() for table in EXT_TABLES_WITH_FK: print(f'\n 处理 {table}...') # 4.1 删除 FK 约束 fk_names = fk_constraint_exists(cur, table, 'user_main') for fk_name in fk_names: print(f' 删除 FK: {fk_name}') cur.execute(f'ALTER TABLE `{table}` DROP FOREIGN KEY `{fk_name}`') # 4.2 检查是否已有 new_user_id 列 cur.execute(""" SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND COLUMN_NAME = 'new_user_id' """, (DB_CONFIG['db'], table)) has_new = cur.fetchone()[0] > 0 if not has_new: # 4.3 添加新 BINARY(16) 列 print(f' 添加 new_user_id BINARY(16) 列') cur.execute(f'ALTER TABLE `{table}` ADD COLUMN `new_user_id` BINARY(16) NULL') # 4.4 填充新列：通过 user_main.id → user_main.yonghuid → User.UserUUID print(f' 填充 new_user_id...') cur.execute(f""" UPDATE `{table}` ext JOIN user_main um ON ext.user_id = um.id JOIN `User` u ON u.YonghuID = um.yonghuid COLLATE utf8mb4_general_ci SET ext.new_user_id = u.UserUUID """) affected = cur.rowcount print(f' 更新 {affected} 行') # 4.5 删除旧 user_id 列 print(f' 删除旧 user_id 列') cur.execute(f'ALTER TABLE `{table}` DROP COLUMN `user_id`') # 4.6 重命名 new_user_id → user_id print(f' 重命名 new_user_id → user_id') cur.execute(f'ALTER TABLE `{table}` CHANGE `new_user_id` `user_id` BINARY(16) NULL') # 4.7 添加索引 print(f' 添加索引') cur.execute(f'ALTER TABLE `{table}` ADD INDEX `{table}_user_id_idx` (`user_id`)') conn.commit() # 4.8 处理 django_admin_log（空表，直接改列类型） print(f'\n 处理 django_admin_log...') fk_names = fk_constraint_exists(cur, 'django_admin_log', 'user_main') for fk_name in fk_names: print(f' 删除 FK: {fk_name}') cur.execute(f'ALTER TABLE `django_admin_log` DROP FOREIGN KEY `{fk_name}`') # 检查并修改 user_id 列类型 cur.execute(""" SELECT DATA_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'django_admin_log' AND COLUMN_NAME = 'user_id' """, (DB_CONFIG['db'],)) col_type = cur.fetchone() if col_type and col_type[0] != 'binary': print(f' 修改 user_id 列类型: {col_type[0]} → BINARY(16)') cur.execute('ALTER TABLE `django_admin_log` MODIFY COLUMN `user_id` BINARY(16) NULL') else: print(f' user_id 已是 BINARY 类型，跳过') conn.commit() cur.close() conn.close() print('\nPhase 4 完成') def phase5_update_migrations(): """Phase 5: 更新 django_migrations 记录""" print('\n' + '='*60) print('Phase 5: 更新 django_migrations 记录') print('='*60) conn = get_conn() cur = conn.cursor() # 删除旧迁移记录 apps = ['gvsdsdk', 'users', 'shop', 'backend', 'rank', 'orders', 'config', 'products'] for app in apps: cur.execute('DELETE FROM django_migrations WHERE app = %s', (app,)) if cur.rowcount > 0: print(f' 删除 {app} 旧迁移记录: {cur.rowcount} 条') # 插入新迁移记录 from datetime import datetime now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') new_migrations = [ ('gvsdsdk', '0001_initial'), ('users', '0001_initial'), ('shop', '0001_initial'), ('backend', '0001_initial'), ('rank', '0001_initial'), ('orders', '0001_initial'), ('config', '0001_initial'), ('products', '0001_initial'), ('products', '0002_initial'), ] for app, name in new_migrations: cur.execute( 'INSERT INTO django_migrations (app, name, applied) VALUES (%s, %s, %s)', (app, name, now) ) print(f' 插入 {app}.{name}') conn.commit() cur.close() conn.close() print('Phase 5 完成') def phase6_verify(): """Phase 6: 验证迁移结果""" print('\n' + '='*60) print('Phase 6: 验证迁移结果') print('='*60) conn = get_conn() cur = conn.cursor() # 1. User 表数据量 cur.execute('SELECT COUNT(*) FROM `User`') user_count = cur.fetchone()[0] print(f' User 表: {user_count} 行') # 2. user_sensitive_data 数据量 cur.execute('SELECT COUNT(*) FROM user_sensitive_data') sd_count = cur.fetchone()[0] print(f' user_sensitive_data: {sd_count} 行') # 3. user_main 对比 cur.execute('SELECT COUNT(*) FROM user_main') um_count = cur.fetchone()[0] print(f' user_main: {um_count} 行 (原表保留)') # 4. 验证 User 表字段 cur.execute('DESCRIBE `User`') print(f'\n User 表结构:') for c in cur.fetchall(): print(f' {c[0]} | {c[1]}') # 5. 验证扩展表 user_id 列类型 print(f'\n 扩展表 user_id 列类型:') for table in EXT_TABLES_WITH_FK: cur.execute(""" SELECT DATA_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND COLUMN_NAME = 'user_id' """, (DB_CONFIG['db'], table)) result = cur.fetchone() print(f' {table}.user_id: {result[0] if result else "不存在"}') # 6. 验证扩展表数据完整性 print(f'\n 扩展表数据完整性:') for table in EXT_TABLES_WITH_FK: cur.execute(f'SELECT COUNT(*) FROM `{table}`') total = cur.fetchone()[0] cur.execute(f'SELECT COUNT(*) FROM `{table}` WHERE user_id IS NULL') null_count = cur.fetchone()[0] print(f' {table}: total={total}, null_user_id={null_count}') # 7. 验证 RBAC 表重命名 print(f'\n RBAC 表重命名验证:') for old, new in RBAC_RENAME_MAP.items(): cur.execute(""" SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s """, (DB_CONFIG['db'], new)) exists = cur.fetchone()[0] > 0 print(f' {new}: {"存在" if exists else "不存在"}') # 8. 验证 django_migrations print(f'\n django_migrations:') cur.execute("SELECT app, name FROM django_migrations WHERE app IN ('gvsdsdk','users','shop') ORDER BY app") for r in cur.fetchall(): print(f' {r[0]}.{r[1]}') # 9. 抽样验证数据一致性 print(f'\n 抽样验证 (前5行):') cur.execute(""" SELECT u.YonghuID, u.UserType, u.UserTele, u.UserAvatar, sd.OpenID, sd.Password, sd.PhoneVerified FROM `User` u LEFT JOIN user_sensitive_data sd ON u.UserUUID = sd.UserUUID LIMIT 5 """) for r in cur.fetchall(): print(f' YonghuID={r[0]}, Type={r[1]}, Tele={r[2]}, OpenID={r[4]}, Verified={r[6]}') cur.close() conn.close() print('\nPhase 6 验证完成') def main(): print('='*60) print('gvsdsdk User 表迁移脚本') print('='*60) print(f'目标数据库: {DB_CONFIG["db"]} @ {DB_CONFIG["host"]}:{DB_CONFIG["port"]}') # 确认 if '--yes' not in sys.argv: print('\n警告: 此脚本将修改生产数据库！') print('请确保已全量备份数据库。') print('使用 --yes 参数跳过确认。') response = input('输入 YES 确认执行: ') if response != 'YES': print('已取消') return # 执行各阶段 phase1_rename_rbac_tables() phase2_create_tables() phase3_migrate_user_data() phase4_alter_extension_tables() phase5_update_migrations() phase6_verify() print('\n' + '='*60) print('迁移完成！') print('='*60) print('\n后续步骤:') print('1. 在服务器上部署新代码 (git pull)') print('2. 重启 Django 服务') print('3. 测试登录和核心功能') print('4. 确认无误后 DROP TABLE user_main') if __name__ == '__main__': main()