#!/usr/bin/env python3 """ 幂等同步：源 SQL dump → 测试库 MySQL，只插入真正缺失的数据。 多次运行安全：所有表按业务唯一键去重。 - user : username 去重（冲突则 old_uid → 现有 new_uid 合并） - assetgroup : remote_group_id 去重 - asset : remote_asset_id 去重（空则按 (group_id, name)） - generationrecord : task_id 去重 - loginrecord : (user_id, created_at, ip_address) 去重 - loginanomaly : (user_id, login_record_id, rule, created_at) 去重 - activesession : session_id 去重 - adminauditlog : (operator_id, action, target_id, created_at) 去重 执行完后，按本次新插入的生成记录增量更新 team 的 total_seconds_used / total_spent / balance。 用法: python3 idempotent_sync.py # dry-run python3 idempotent_sync.py --commit # 写入 """ import re, sys from decimal import Decimal import pymysql import pymysql.cursors SOURCE = '/Users/maidong/Desktop/zyc/研究openclaw/视频生成平台/jimeng-clone/数据库备份/video_auto_原19-55.sql' TARGET_TEAMS = (3, 4, 12) DB_TEST = dict(host='mysql-8351f937d637-public.rds.volces.com', port=3306, user='zyc', password='Zyc188208', database='video_auto', charset='utf8mb4', autocommit=False, cursorclass=pymysql.cursors.DictCursor) DB_PROD = dict(host='mysql-d9bb4e81696d-public.rds.volces.com', port=3306, user='zyc', password='Zyc188208', database='video_auto', charset='utf8mb4', autocommit=False, cursorclass=pymysql.cursors.DictCursor) # ---------- SQL dump 解析 ---------- def split_values(s): vals, cur, in_str, i = [], '', False, 0 while i < len(s): c = s[i] if in_str: if c == '\\' and i+1 < len(s): cur += c + s[i+1]; i += 2; continue if c == "'": in_str = False cur += c else: if c == "'": in_str = True; cur += c elif c == ',': vals.append(cur.strip()); cur = '' else: cur += c i += 1 vals.append(cur.strip()) return vals def parse_table(tbl): rows = [] with open(SOURCE, 'r', encoding='utf-8') as f: for line in f: if not line.startswith(f'INSERT INTO `{tbl}`'): continue m = re.search(r'VALUES \((.*)\);\s*$', line) if not m: continue rows.append(split_values(m.group(1))) return rows def unq(v): if v == 'NULL': return None if v.startswith("'") and v.endswith("'"): return (v[1:-1].replace("\\'", "'").replace('\\"', '"').replace('\\\\', '\\') .replace('\\n', '\n').replace('\\r', '\r').replace('\\t', '\t').replace('\\0', '\0')) return v # ---------- 源数据 → dict 形式 ---------- def row_to_dict(row, cols): return {c: unq(v) for c, v in zip(cols, row)} USER_COLS = ['id','password','last_login','is_superuser','username','first_name','last_name', 'is_staff','is_active','date_joined','email','created_at','updated_at', 'daily_seconds_limit','monthly_seconds_limit','is_team_admin','team_id', 'must_change_password','disabled_by','daily_generation_limit', 'monthly_generation_limit','spending_limit','last_read_announcement','is_team_owner'] AG_COLS = ['id','remote_group_id','name','description','thumbnail_url','created_at','created_by_id','team_id'] ASSET_COLS_SRC = ['id','remote_asset_id','name','url','status','error_message','created_at','group_id'] # 测试库 asset 多 asset_type,duration,thumbnail_url — 插入时补 GEN_COLS = ['id','task_id','prompt','mode','model','aspect_ratio','duration','status','created_at', 'user_id','seconds_consumed','ark_task_id','error_message','reference_urls','result_url', 'base_cost_amount','cost_amount','frozen_amount','resolution','tokens_consumed', 'is_favorited','seed','completed_at','raw_error','updated_at','is_deleted'] LR_COLS = ['id','ip_address','user_agent','created_at','user_id','geo_city','geo_country', 'geo_province','geo_source','team_id'] LA_COLS = ['id','level','rule','detail','alerted','auto_disabled','disabled_target','created_at', 'login_record_id','team_id','user_id'] AS_COLS = ['id','session_id','device_type','user_agent','created_at','user_id'] AL_COLS = ['id','operator_name','action','target_type','target_id','target_name','before','after', 'ip_address','created_at','operator_id'] def main(): commit = '--commit' in sys.argv use_prod = '--prod' in sys.argv DB = DB_PROD if use_prod else DB_TEST target_name = '【正式服】' if use_prod else '【测试服】' print(f'\n🎯 目标: {target_name} {DB["host"]}') if use_prod and commit: print('⚠️ 正在写入正式服！') # ===== 解析源 ===== print('解析源 SQL...') src_users_all = [row_to_dict(r, USER_COLS) for r in parse_table('accounts_user')] src_ags_all = [row_to_dict(r, AG_COLS) for r in parse_table('generation_assetgroup')] src_assets_all = [row_to_dict(r, ASSET_COLS_SRC) for r in parse_table('generation_asset')] src_gens_all = [row_to_dict(r, GEN_COLS) for r in parse_table('generation_generationrecord')] src_lrs_all = [row_to_dict(r, LR_COLS) for r in parse_table('accounts_loginrecord')] src_las_all = [row_to_dict(r, LA_COLS) for r in parse_table('accounts_loginanomaly')] src_ases_all = [row_to_dict(r, AS_COLS) for r in parse_table('accounts_activesession')] src_als_all = [row_to_dict(r, AL_COLS) for r in parse_table('accounts_adminauditlog')] # 只处理目标团队的用户 src_team_users = [u for u in src_users_all if str(u['team_id']) in tuple(str(t) for t in TARGET_TEAMS)] src_uid_set = {int(u['id']) for u in src_team_users} src_uname_set = {u['username'] for u in src_team_users} # 源里目标团队相关的数据 src_ags = [g for g in src_ags_all if str(g['team_id']) in tuple(str(t) for t in TARGET_TEAMS)] src_ag_ids = {int(g['id']) for g in src_ags} src_assets = [a for a in src_assets_all if int(a['group_id']) in src_ag_ids] src_gens = [g for g in src_gens_all if int(g['user_id']) in src_uid_set] src_lrs = [r for r in src_lrs_all if int(r['user_id']) in src_uid_set] src_las = [a for a in src_las_all if int(a['user_id']) in src_uid_set] src_ases = [s for s in src_ases_all if int(s['user_id']) in src_uid_set] src_als = [a for a in src_als_all if a['operator_id'] is not None and int(a['operator_id']) in src_uid_set] # ===== 连测试库 ===== print('连接测试库...') conn = pymysql.connect(**DB) cur = conn.cursor() try: cur.execute('SET FOREIGN_KEY_CHECKS = 0') # ---------- 1. user：按 username 去重 ---------- print('\n[1/8] accounts_user') ph = ','.join(['%s']*len(src_uname_set)) cur.execute(f"SELECT id, username FROM accounts_user WHERE username IN ({ph})", list(src_uname_set)) existing_by_uname = {r['username']: r['id'] for r in cur.fetchall()} uid_map = {} # old_uid → new_uid user_inserts = 0 for u in src_team_users: old_uid = int(u['id']) if u['username'] in existing_by_uname: uid_map[old_uid] = existing_by_uname[u['username']] continue # 插入新用户（AUTO id） insert_cols = [c for c in USER_COLS if c != 'id'] insert_vals = [u[c] for c in insert_cols] cur.execute( f"INSERT INTO `accounts_user` ({','.join('`'+c+'`' for c in insert_cols)}) " f"VALUES ({','.join(['%s']*len(insert_cols))})", insert_vals ) new_uid = cur.lastrowid uid_map[old_uid] = new_uid user_inserts += 1 print(f' ➕ 新用户: {u["username"]} (old {old_uid} → new {new_uid})') print(f' 新增 {user_inserts} 用户，映射 {len(uid_map)}') # ---------- 2. assetgroup：按 remote_group_id 去重 ---------- print('\n[2/8] generation_assetgroup') src_rgids = [g['remote_group_id'] for g in src_ags if g['remote_group_id']] if src_rgids: ph = ','.join(['%s']*len(src_rgids)) cur.execute(f"SELECT id, remote_group_id FROM generation_assetgroup WHERE remote_group_id IN ({ph})", src_rgids) existing_by_rgid = {r['remote_group_id']: r['id'] for r in cur.fetchall()} else: existing_by_rgid = {} ag_map = {} # old_ag_id → new_ag_id ag_inserts = 0 for g in src_ags: old_id = int(g['id']) rgid = g['remote_group_id'] if rgid and rgid in existing_by_rgid: ag_map[old_id] = existing_by_rgid[rgid] continue insert_cols = [c for c in AG_COLS if c != 'id'] vals = [] for c in insert_cols: v = g[c] if c == 'created_by_id' and v is not None: ov = int(v) v = uid_map.get(ov, ov) # 可能 created_by 是两团队之外的用户，直接保留 vals.append(v) cur.execute( f"INSERT INTO `generation_assetgroup` ({','.join('`'+c+'`' for c in insert_cols)}) " f"VALUES ({','.join(['%s']*len(insert_cols))})", vals ) ag_map[old_id] = cur.lastrowid ag_inserts += 1 print(f' 新增 {ag_inserts} assetgroup，映射 {len(ag_map)}') # ---------- 3. asset：按 remote_asset_id 去重，无 remote_asset_id 按 (group_id, name) ---------- print('\n[3/8] generation_asset') # 已有 remote_asset_id 集合 cur.execute("SELECT remote_asset_id FROM generation_asset WHERE remote_asset_id != ''") existing_raids = {r['remote_asset_id'] for r in cur.fetchall()} # 已有 (group_id, name) 组合（当 remote_asset_id 为空） cur.execute("SELECT group_id, name FROM generation_asset WHERE remote_asset_id = ''") existing_namekeys = {(r['group_id'], r['name']) for r in cur.fetchall()} asset_inserts = 0 VIDEO_EXT = ('.mp4', '.mov', '.avi', '.webm', '.mkv', '.m4v') AUDIO_EXT = ('.mp3', '.wav', '.m4a', '.aac', '.flac', '.ogg') for a in src_assets: new_gid = ag_map[int(a['group_id'])] raid = a['remote_asset_id'] key = (new_gid, a['name']) if raid and raid in existing_raids: continue if not raid and key in existing_namekeys: continue # 推断 asset_type（测试库 NOT NULL） url_l = (a['url'] or '').lower() if any(e in url_l for e in VIDEO_EXT): atype = 'Video' elif any(e in url_l for e in AUDIO_EXT): atype = 'Audio' else: atype = 'Image' cur.execute( """INSERT INTO generation_asset (remote_asset_id,name,url,status,error_message,created_at, group_id,asset_type,duration,thumbnail_url) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", (raid or '', a['name'], a['url'], a['status'], a['error_message'], a['created_at'], new_gid, atype, None, '') ) asset_inserts += 1 print(f' 新增 {asset_inserts} asset') # ---------- 4. generationrecord：按 task_id 去重 ---------- print('\n[4/8] generation_generationrecord') src_tids = [g['task_id'] for g in src_gens] if src_tids: ph = ','.join(['%s']*len(src_tids)) cur.execute(f"SELECT task_id FROM generation_generationrecord WHERE task_id IN ({ph})", src_tids) existing_tids = {r['task_id'] for r in cur.fetchall()} else: existing_tids = set() gen_inserts_by_team = {t: [] for t in TARGET_TEAMS} # 用于最后 team 字段重算 gen_insert_cols = [c for c in GEN_COLS if c != 'id'] + ['thumbnail_url'] for g in src_gens: if g['task_id'] in existing_tids: continue new_uid = uid_map[int(g['user_id'])] vals = [g[c] if c != 'user_id' else new_uid for c in GEN_COLS if c != 'id'] + [''] cur.execute( f"INSERT INTO generation_generationrecord " f"({','.join('`'+c+'`' for c in gen_insert_cols)}) " f"VALUES ({','.join(['%s']*len(gen_insert_cols))})", vals ) # 分流到所属 team（根据源用户的 team_id） src_uid = int(g['user_id']) src_user = next(u for u in src_team_users if int(u['id']) == src_uid) tid = int(src_user['team_id']) gen_inserts_by_team[tid].append(g) parts = ', '.join(f'team{t}={len(gen_inserts_by_team[t])}' for t in TARGET_TEAMS) total = sum(len(v) for v in gen_inserts_by_team.values()) print(f' 新增 {total} generationrecord ({parts})') # ---------- 5. loginrecord：按 (user_id, created_at, ip) 去重 ---------- print('\n[5/8] accounts_loginrecord') # 一次查出相关 user_id 的所有 loginrecord mapped_uids = set(uid_map.values()) if mapped_uids: ph = ','.join(['%s']*len(mapped_uids)) cur.execute(f"""SELECT id, user_id, created_at, ip_address FROM accounts_loginrecord WHERE user_id IN ({ph})""", list(mapped_uids)) existing_lr = {(r['user_id'], r['created_at'], r['ip_address']): r['id'] for r in cur.fetchall()} else: existing_lr = {} lr_map = {} # old_lr_id → new_lr_id (本次插入的) lr_inserts = 0 lr_insert_cols = [c for c in LR_COLS if c != 'id'] for r in src_lrs: new_uid = uid_map[int(r['user_id'])] # 解析 created_at → datetime from datetime import datetime ca = r['created_at'] if isinstance(ca, str): try: ca_dt = datetime.strptime(ca, '%Y-%m-%d %H:%M:%S.%f') except ValueError: ca_dt = datetime.strptime(ca, '%Y-%m-%d %H:%M:%S') else: ca_dt = ca key = (new_uid, ca_dt, r['ip_address']) if key in existing_lr: lr_map[int(r['id'])] = existing_lr[key] continue vals = [r[c] if c != 'user_id' else new_uid for c in lr_insert_cols] cur.execute( f"INSERT INTO accounts_loginrecord ({','.join('`'+c+'`' for c in lr_insert_cols)}) " f"VALUES ({','.join(['%s']*len(lr_insert_cols))})", vals ) lr_map[int(r['id'])] = cur.lastrowid lr_inserts += 1 print(f' 新增 {lr_inserts} loginrecord') # ---------- 6. loginanomaly：按 (user_id, login_record_id, rule, created_at) ---------- print('\n[6/8] accounts_loginanomaly') la_inserts = 0 la_insert_cols = [c for c in LA_COLS if c != 'id'] for a in src_las: new_uid = uid_map[int(a['user_id'])] old_lr_id = int(a['login_record_id']) if old_lr_id not in lr_map: # login_record 可能不在源抽出范围（跨团队），跳过 continue new_lr_id = lr_map[old_lr_id] cur.execute("""SELECT 1 FROM accounts_loginanomaly WHERE user_id=%s AND login_record_id=%s AND rule=%s AND created_at=%s""", (new_uid, new_lr_id, a['rule'], a['created_at'])) if cur.fetchone(): continue vals = [] for c in la_insert_cols: if c == 'user_id': vals.append(new_uid) elif c == 'login_record_id': vals.append(new_lr_id) else: vals.append(a[c]) cur.execute( f"INSERT INTO accounts_loginanomaly ({','.join('`'+c+'`' for c in la_insert_cols)}) " f"VALUES ({','.join(['%s']*len(la_insert_cols))})", vals ) la_inserts += 1 print(f' 新增 {la_inserts} loginanomaly') # ---------- 7. activesession：按 session_id 去重 ---------- print('\n[7/8] accounts_activesession') src_sids = [s['session_id'] for s in src_ases] if src_sids: ph = ','.join(['%s']*len(src_sids)) cur.execute(f"SELECT session_id FROM accounts_activesession WHERE session_id IN ({ph})", src_sids) existing_sids = {r['session_id'] for r in cur.fetchall()} else: existing_sids = set() as_inserts = 0 as_insert_cols = [c for c in AS_COLS if c != 'id'] for s in src_ases: if s['session_id'] in existing_sids: continue new_uid = uid_map[int(s['user_id'])] vals = [s[c] if c != 'user_id' else new_uid for c in as_insert_cols] cur.execute( f"INSERT INTO accounts_activesession ({','.join('`'+c+'`' for c in as_insert_cols)}) " f"VALUES ({','.join(['%s']*len(as_insert_cols))})", vals ) as_inserts += 1 print(f' 新增 {as_inserts} activesession') # ---------- 8. adminauditlog：按 (operator_id, action, target_id, created_at) ---------- print('\n[8/8] accounts_adminauditlog') al_inserts = 0 al_insert_cols = [c for c in AL_COLS if c != 'id'] for a in src_als: op_id = int(a['operator_id']) new_op_id = uid_map.get(op_id, op_id) tgt = int(a['target_id']) if a['target_id'] else None new_tgt = uid_map.get(tgt, tgt) if tgt else None cur.execute("""SELECT 1 FROM accounts_adminauditlog WHERE operator_id=%s AND action=%s AND (target_id=%s OR (target_id IS NULL AND %s IS NULL)) AND created_at=%s""", (new_op_id, a['action'], new_tgt, new_tgt, a['created_at'])) if cur.fetchone(): continue vals = [] for c in al_insert_cols: if c == 'operator_id': vals.append(new_op_id) elif c == 'target_id': vals.append(new_tgt) else: vals.append(a[c]) cur.execute( f"INSERT INTO accounts_adminauditlog ({','.join('`'+c+'`' for c in al_insert_cols)}) " f"VALUES ({','.join(['%s']*len(al_insert_cols))})", vals ) al_inserts += 1 print(f' 新增 {al_inserts} adminauditlog') # ---------- 重算 team 统计 ---------- print('\n[重算 team 统计]') for tid in TARGET_TEAMS: gens_added = gen_inserts_by_team[tid] if not gens_added: print(f' Team {tid}: 无新增生成记录，跳过') continue sec_delta = sum(Decimal(g['seconds_consumed']) for g in gens_added) cost_delta = sum(Decimal(g['cost_amount']) for g in gens_added) cur.execute("""UPDATE accounts_team SET total_seconds_used = total_seconds_used + %s, total_spent = total_spent + %s, balance = balance - %s WHERE id=%s""", (sec_delta, cost_delta, cost_delta, tid)) print(f' Team {tid}: +seconds={sec_delta} +spent={cost_delta} -balance={cost_delta}') cur.execute('SET FOREIGN_KEY_CHECKS = 1') if commit: conn.commit() print('\n✅ COMMITTED') else: conn.rollback() print('\n🔎 Rolled back (use --commit to persist)') except Exception as e: conn.rollback() print(f'\n❌ Error: {e}') raise finally: conn.close() if __name__ == '__main__': main()