Files
Django/migrate_data.py
T

464 lines
18 KiB
Python

#!/usr/bin/env python3
"""
migrate_data.py — 从 SQL 备份文件直接迁移数据到新的 gvsdsdk User + RBAC 体系
v3: 正确的字段映射 + 基于扩展表存在性分配角色 + bulk_create + 实时进度
"""
import os, sys, re, uuid, time, datetime, MySQLdb
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'a_long_dianjing.settings')
import django; django.setup()
from app_secrets import DATABASE_HOST, DATABASE_PORT, DATABASE_USER, DATABASE_PASSWORD, DATABASE_NAME
TENANT_UUID = uuid.UUID('a0000000-0000-0000-0000-000000000001').bytes
COMPANY_UUID = uuid.UUID('b0000000-0000-0000-0000-000000000001').bytes
BATCH_SIZE = 500
ROLE_MAP = {
'normal': '消费者', 'dashou': '打手', 'shop': '商家',
'guanshi': '负责人', 'admin': '管理员', 'kefu': '客服',
'zuzhang': '组长', 'shenheguan': '审核官',
}
# 角色优先级(高→低): 管理员 > 客服 > 负责人 > 组长 > 审核官 > 商家 > 打手 > 消费者
ROLE_PRIORITY = ['admin', 'kefu', 'guanshi', 'zuzhang', 'shenheguan', 'shop', 'dashou', 'normal']
# 扩展表 → 角色类型 映射
EXT_ROLE_MAP = {
'admin_profile': 'admin',
'user_kefu': 'kefu',
'user_guanshi': 'guanshi',
'user_zuzhang': 'zuzhang',
'user_shenheguan': 'shenheguan',
'user_shangjia': 'shop',
'user_dashou': 'dashou',
}
SQL_FILE = os.path.join(os.path.dirname(__file__), 'xaio_cheng_xu_backup.sql')
class Progress:
def __init__(self, total, width=35):
self.total = total; self.width = width; self.start = time.time(); self.done = 0
def update(self, n=1):
self.done += n; pct = self.done / self.total if self.total else 0
filled = int(self.width * pct); bar = '█' * filled + '░' * (self.width - filled)
elapsed = time.time() - self.start
eta = (elapsed / self.done * (self.total - self.done)) if self.done else 0
sys.stdout.write(f'\r [{bar}] {pct:5.1%} {self.done}/{self.total} {elapsed:.0f}s ETA{eta:.0f}s')
sys.stdout.flush()
def finish(self):
elapsed = time.time() - self.start
sys.stdout.write(f'\r [{"█"*self.width}] 100% {self.done}/{self.total} {elapsed:.0f}s\n')
sys.stdout.flush()
def parse_insert_values(table_name, sql_content):
pattern = re.compile(r"INSERT\s+INTO\s+`" + re.escape(table_name) + r"`\s+VALUES\s*\((.+?)\)\s*;", re.IGNORECASE | re.DOTALL)
return [m.group(1) for m in pattern.finditer(sql_content)]
def parse_sql_value(val_str):
val_str = val_str.strip()
if val_str.upper() == 'NULL': return None
if val_str.startswith("'") and val_str.endswith("'"):
return val_str[1:-1].replace("\\'", "'").replace("\\\\", "\\")
try: return int(val_str)
except ValueError:
try: return float(val_str)
except ValueError: return val_str
def split_sql_values(values_str):
values, current, in_quote, escape = [], [], False, False
for ch in values_str:
if escape: current.append(ch); escape = False; continue
if ch == '\\': current.append(ch); escape = True; continue
if ch == "'": in_quote = not in_quote; current.append(ch); continue
if ch == ',' and not in_quote: values.append(''.join(current).strip()); current = []; continue
current.append(ch)
if current: values.append(''.join(current).strip())
return [parse_sql_value(v) for v in values]
def parse_dt(v):
if isinstance(v, str) and v:
try: return datetime.datetime.fromisoformat(v.replace(' ', 'T'))
except: pass
return datetime.datetime.utcnow()
def get_raw_conn(db=None):
kwargs = dict(host=DATABASE_HOST, port=int(DATABASE_PORT), user=DATABASE_USER,
password=DATABASE_PASSWORD, charset='utf8mb4', connect_timeout=10,
read_timeout=60, write_timeout=60)
if db: kwargs['database'] = db
return MySQLdb.connect(**kwargs)
def run_migration():
from gvsdsdk.models import User, Role, UserRole
from users.models import (
UserBoss, UserDashou, UserShangjia, UserGuanshi,
UserZuzhang, UserShenheguan, AdminProfile, UserKefu,
UserSensitiveData
)
print("=" * 60)
print("开始从 SQL 备份迁移数据到 gvsdsdk.User + RBAC")
print("=" * 60)
# [0] 读取 SQL
print("\n[0/7] 读取 SQL 备份文件...", end=' ')
with open(SQL_FILE, 'r', encoding='utf-8') as f:
sql_content = f.read()
print(f"{len(sql_content)/1024/1024:.1f} MB")
# [1] 解析所有扩展表的 user_id → 建立角色映射
print("\n[1/7] 解析扩展表建立角色映射...")
# 正确的 user_id 索引位置
ext_user_id_idx = {
'user_dashou': 22,
'user_guanshi': 11,
'user_kefu': 9,
'user_shangjia': 14,
'user_boss': 6,
'admin_profile': 4,
'user_zuzhang': 17,
'user_shenheguan': 9,
}
# user_main.id → 角色集合
user_roles = {} # old_user_id -> set of role_types
for table_name, role_type in EXT_ROLE_MAP.items():
inserts = parse_insert_values(table_name, sql_content)
uid_idx = ext_user_id_idx[table_name]
count = 0
for values_str in inserts:
vals = split_sql_values(values_str)
if uid_idx < len(vals) and vals[uid_idx] is not None:
old_uid = vals[uid_idx]
if old_uid not in user_roles:
user_roles[old_uid] = set()
user_roles[old_uid].add(role_type)
count += 1
print(f" {table_name}: {count} 条有 user_id")
# [2] 解析 user_main
print("\n[2/7] 解析 user_main...")
user_main_inserts = parse_insert_values('user_main', sql_content)
total = len(user_main_inserts)
print(f" 共 {total} 条记录")
uid_map = {} # yonghuid -> UserUUID (bytes)
user_id_map = {} # old id -> yonghuid
phone_to_yonghuid = {}
# 先解析所有 user_main 记录
user_data = [] # (old_id, yonghuid, fields_dict, role_type)
prog = Progress(total)
for i, values_str in enumerate(user_main_inserts):
vals = split_sql_values(values_str)
if len(vals) < 16:
prog.update(); continue
old_id = vals[1]
yonghuid = str(vals[2]) if vals[2] else None
if not yonghuid:
prog.update(); continue
# 确定角色:基于扩展表存在性 + user_type
roles = user_roles.get(old_id, set())
user_type = str(vals[11]) if vals[11] else 'normal'
# user_type 也可能指定角色
if user_type == 'admin':
roles.add('admin')
elif user_type == 'kefu':
roles.add('kefu')
# 如果没有任何角色,默认为 normal
if not roles:
roles.add('normal')
# 选择优先级最高的角色
primary_role = 'normal'
for r in ROLE_PRIORITY:
if r in roles:
primary_role = r
break
user_data.append((old_id, yonghuid, vals, primary_role, roles))
user_id_map[old_id] = yonghuid
phone = str(vals[5]) if vals[5] else ''
if phone: phone_to_yonghuid[phone] = yonghuid
prog.update()
prog.finish()
print(f" 解析完成: {len(user_data)} 条有效记录")
# [3] 创建角色
print("\n[3/7] 创建角色...")
created_roles = {}
for user_type, role_name in ROLE_MAP.items():
role, created = Role.objects.get_or_create(
RoleName=role_name, TenantUUID=TENANT_UUID,
defaults={'RoleUUID': uuid.uuid4().bytes, 'RoleDesc': f'{role_name}角色', 'RoleStatus': 1, 'AssignScope': 0})
created_roles[user_type] = role
print(f" {role_name}: {'NEW' if created else 'EXISTS'}")
# [4] 批量创建 User + UserRole + UserSensitiveData
print("\n[4/7] 批量创建 User + UserRole + UserSensitiveData...")
user_batch, ur_batch, usd_batch = [], [], []
errors = 0
prog = Progress(len(user_data))
for old_id, yonghuid, vals, primary_role, all_roles in user_data:
openid = str(vals[3]) if vals[3] else ''
avatar = str(vals[4]) if vals[4] else 'a_long/morentouxiang.jpg'
phone = str(vals[5]) if vals[5] else ''
password = str(vals[6]) if vals[6] else ''
zhifu = str(vals[7]) if vals[7] else ''
skzhanghao = str(vals[8]) if vals[8] else ''
ip = str(vals[9]) if vals[9] else ''
last_login_time = vals[10]
create_time = vals[12]
unionid = str(vals[14]) if vals[14] else ''
shoujihao_renzheng = bool(vals[15]) if vals[15] is not None else False
user_uuid = uuid.uuid4().bytes
ct, lt = parse_dt(create_time), parse_dt(last_login_time)
user_batch.append(User(
UserUUID=user_uuid, UserName=yonghuid, UserUID=yonghuid,
OpenID=openid, UnionID=unionid, Avatar=avatar, Phone=phone,
PhoneVerified=shoujihao_renzheng, IP=ip, UserAccountLicense=1,
UserPositionStatus=0, UserCreateTime=ct, UserLastLoginDate=lt,
IsStaff=('admin' in all_roles), IsSuperuser=('admin' in all_roles)))
# 为每个角色创建 UserRole(多对多)
for role_type in all_roles:
role = created_roles.get(role_type)
if role:
ur_batch.append(UserRole(
UserRoleUUID=uuid.uuid4().bytes, UserUUID=user_uuid,
RoleUUID=role.RoleUUID, CompanyUUID=COMPANY_UUID))
usd_batch.append(UserSensitiveData(
user_uuid_id=user_uuid, password=password, zhifu=zhifu,
skzhanghao=skzhanghao, shoujihao_renzheng=shoujihao_renzheng))
uid_map[yonghuid] = user_uuid
if len(user_batch) >= BATCH_SIZE:
try:
User.objects.bulk_create(user_batch, ignore_conflicts=True)
UserRole.objects.bulk_create(ur_batch, ignore_conflicts=True)
UserSensitiveData.objects.bulk_create(usd_batch, ignore_conflicts=True)
except Exception: errors += len(user_batch)
user_batch, ur_batch, usd_batch = [], [], []
prog.update()
if user_batch:
try:
User.objects.bulk_create(user_batch, ignore_conflicts=True)
UserRole.objects.bulk_create(ur_batch, ignore_conflicts=True)
UserSensitiveData.objects.bulk_create(usd_batch, ignore_conflicts=True)
except Exception: errors += len(user_batch)
prog.finish()
print(f" 成功: {len(uid_map)}, 错误: {errors}")
# [5] 迁移扩展表 — 用 ORM bulk_create(正确字段映射)
print("\n[5/7] 迁移扩展表...")
# 正确的字段映射: (sql_index, model_field_name)
# None = 跳过, '_user_id' = FK 到新 User
ext_configs = [
('user_dashou', UserDashou, [
(0, None), (1, 'nicheng'), (2, 'chenghao'), (3, 'zhuangtai'), (4, 'zaixianzhuangtai'),
(5, 'zhanghaozhuangtai'), (6, 'jieshao'), (7, 'jiedanzongliang'), (8, 'chengjiaozongliang'),
(9, 'tuikuanliang'), (10, 'yue'), (11, 'zonge'), (12, 'dianhua'), (13, 'wechat'),
(14, 'yaoqingren'), (15, 'jinrijiedan'), (16, 'jinrishouyi'), (17, 'jinyuejiedan'),
(18, 'jinyueshouyi'), (19, 'jifen'), (20, 'yajin'), (21, None), # create_time
(22, '_user_id'), # user_id FK
(23, 'ewai_xiane'), (24, 'jinritixian_jine'), (25, 'kaioi_ewai_xiane'),
(26, 'last_tixian_date'),
]),
('user_guanshi', UserGuanshi, [
(0, None), (1, 'yaoqingma'), (2, 'dianhua'), (3, 'wechat'),
(4, 'yaogingshuliang'), (5, 'zhuangtai'), (6, 'jinrichongzhi'),
(7, 'jinyuechongzhi'), (8, 'chongzhifenrun'), (9, 'yue'), (10, None), # create_time
(11, '_user_id'),
(12, 'fenghong_erci'), (13, 'fenghong_erci_enabled'), (14, 'fenhong_yici'),
(15, 'ewai_xiane'), (16, 'jinritixian_jine'), (17, 'kaioi_ewai_xiane'),
(18, 'last_tixian_date'), (19, 'yaoqingren'), (20, 'invite_qrcode_url'),
]),
('user_kefu', UserKefu, [
(0, None), (1, 'nicheng'), (2, 'erjimima'), (3, 'zhuangtai'),
(4, 'jinrichuli'), (5, 'jinyuechuli'), (6, 'zongchuli'),
(7, None), (8, None), (9, '_user_id'),
]),
('user_shangjia', UserShangjia, [
(0, None), (1, 'nicheng'), (2, 'zhuangtai'), (3, 'dianhua'), (4, 'wechat'),
(5, 'fabu'), (6, 'tuikuan'), (7, 'yue'), (8, 'chengjiao'),
(9, 'jinridingdan'), (10, 'jinriliushui'), (11, 'jinyuedingdan'),
(12, 'jinyueliushui'), (13, None), (14, '_user_id'),
]),
('user_boss', UserBoss, [
(0, None), (1, 'nickname'), (2, 'zonge'), (3, 'alldingdan'),
(4, 'alltui'), (5, None), (6, '_user_id'),
]),
('admin_profile', AdminProfile, [
(0, None), (1, 'password'), (2, 'wechat'), (3, None),
(4, '_user_id'), (5, 'yaoqingma'),
]),
('user_zuzhang', UserZuzhang, [
(0, None), (1, 'yaoqingma'), (2, 'fenyong_zonge'), (3, 'ketixian_jine'),
(4, 'yaoqing_zongshu'), (5, 'jinri_fenyong'), (6, 'jinyue_fenyong'),
(7, 'zhuangtai'), (8, 'haibao_url'), (9, 'jinri_tixian'), (10, 'last_tixian_time'),
(11, 'ewai_tixian_xiane'), (12, 'kaioi_ewai_tixian'), (13, 'kaioi_ewai_fenhong'),
(14, 'ewai_fenhong_jine'), (15, None), (16, None), (17, '_user_id'),
]),
('user_shenheguan', UserShenheguan, [
(0, None), (1, 'zhuangtai'), (2, 'shenhe_zhuangtai'),
(3, 'shenhe_zongshu'), (4, 'tongguo_zongshu'), (5, 'yue'), (6, 'zonge'),
(7, None), (8, None), (9, '_user_id'), (10, 'is_renzheng'),
]),
]
for table_name, model, field_map in ext_configs:
inserts = parse_insert_values(table_name, sql_content)
if not inserts:
print(f" {table_name}: 0 条"); continue
user_id_sql_idx = next((idx for idx, fn in field_map if fn == '_user_id'), None)
if user_id_sql_idx is None:
print(f" {table_name}: 无法定位 user_id"); continue
batch = []
success = 0
fail = 0
prog = Progress(len(inserts))
for values_str in inserts:
vals = split_sql_values(values_str)
if not vals or user_id_sql_idx >= len(vals) or vals[user_id_sql_idx] is None:
fail += 1; prog.update(); continue
old_user_id = vals[user_id_sql_idx]
yonghuid = user_id_map.get(old_user_id)
if not yonghuid:
fail += 1; prog.update(); continue
new_uuid = uid_map.get(yonghuid)
if not new_uuid:
fail += 1; prog.update(); continue
try:
obj = model(user_id=new_uuid)
for sql_idx, field_name in field_map:
if field_name is None or field_name == '_user_id':
continue
if sql_idx < len(vals) and vals[sql_idx] is not None:
try: setattr(obj, field_name, vals[sql_idx])
except: pass
batch.append(obj)
success += 1
except Exception:
fail += 1
if len(batch) >= BATCH_SIZE:
try: model.objects.bulk_create(batch, ignore_conflicts=True)
except: pass
batch = []
prog.update()
if batch:
try: model.objects.bulk_create(batch, ignore_conflicts=True)
except: pass
prog.finish()
print(f" 成功: {success}, 失败(无user_id): {fail}")
# [6] 迁移其他引用表
print("\n[6/7] 迁移其他引用 yonghuid 的表...")
other_tables = [
'tixianjilu', 'tixian_shenhe_jilu', 'tixian_auto_record',
'xiugaijilu', 'ranking_record',
]
conn = get_raw_conn(db=DATABASE_NAME)
cursor = conn.cursor()
cursor.execute('SET FOREIGN_KEY_CHECKS = 0')
for table_name in other_tables:
inserts = parse_insert_values(table_name, sql_content)
if not inserts:
print(f" {table_name}: 0 条"); continue
success = 0
prog = Progress(len(inserts))
for values_str in inserts:
sql = f"INSERT INTO `{table_name}` VALUES ({values_str})"
try:
cursor.execute(sql)
success += 1
except Exception:
pass
if success % 100 == 0 and success > 0:
try: conn.commit()
except: pass
prog.update()
try: conn.commit()
except MySQLdb.OperationalError:
conn = get_raw_conn(db=DATABASE_NAME)
cursor = conn.cursor()
cursor.execute('SET FOREIGN_KEY_CHECKS = 0')
prog.finish()
print(f" 成功: {success}")
try:
cursor.execute('SET FOREIGN_KEY_CHECKS = 1')
conn.commit(); conn.close()
except: pass
# [7] 验证
print("\n[7/7] 验证迁移结果...")
new_user_count = User.objects.count()
new_ur_count = UserRole.objects.count()
usd_count = UserSensitiveData.objects.count()
print(f" gvsdsdk.User: {new_user_count}")
print(f" UserRole: {new_ur_count}")
print(f" UserSensitiveData: {usd_count}")
print("\n 角色分布:")
for user_type, role in created_roles.items():
ur_count = UserRole.objects.filter(RoleUUID=role.RoleUUID).count()
print(f" {ROLE_MAP[user_type]}: {ur_count} 人")
print("\n 扩展表:")
for name, model in [('user_dashou', UserDashou), ('user_guanshi', UserGuanshi),
('user_kefu', UserKefu), ('user_shangjia', UserShangjia),
('user_boss', UserBoss), ('admin_profile', AdminProfile),
('user_zuzhang', UserZuzhang), ('user_shenheguan', UserShenheguan)]:
print(f" {name}: {model.objects.count()} 条")
print("\n" + "=" * 60)
print("迁移完成!")
print(f"TENANT_UUID = {uuid.UUID(bytes=TENANT_UUID)}")
print(f"COMPANY_UUID = {uuid.UUID(bytes=COMPANY_UUID)}")
print("=" * 60)
if __name__ == '__main__':
run_migration()