#!/usr/bin/env python3 """Restore a validated migration into a NEW Cloudflare D1/R2 deployment. Install boto3 first. Credentials are entered via getpass, never command arguments. Only the target is written; source data and the running original site are unchanged. """ import argparse, getpass, hashlib, json, os, re, sqlite3, urllib.request, urllib.error from pathlib import Path, PurePosixPath def file_digest(path): with Path(path).open('rb') as inp:return hashlib.file_digest(inp,'sha256').hexdigest() def save_state(path,state): temporary=path.with_suffix('.tmp');temporary.write_text(json.dumps(state,ensure_ascii=False,indent=2),encoding='utf-8');os.chmod(temporary,0o600);os.replace(temporary,path) def migrate(directory,account_id,database_id,bucket_name,query,s3): root=Path(directory);verified=json.loads((root/'verified.json').read_text('utf-8'));manifest=json.loads((root/'manifest.json').read_text('utf-8')) if not verified.get('complete') or verified['databaseSha256']!=file_digest(root/'database.sqlite') or verified['manifestSha256']!=file_digest(root/'manifest.json') or verified['sourceSha256']!=file_digest(root/'source.zip'):raise ValueError('解密后的文件已变化，请从完整迁移包重新解密') target={'accountId':account_id,'databaseId':database_id,'bucket':bucket_name,'jobId':manifest['jobId'],'databaseSha256':verified['databaseSha256']};state_path=root/'migration-upload-state.json' source=sqlite3.connect(root/'database.sqlite');source.row_factory=sqlite3.Row state=json.loads(state_path.read_text('utf-8')) if state_path.exists() else None definitions=manifest['tables']+manifest['emptyTables'];allowed={d['name'] for d in definitions} if state: if state['target']!=target:raise ValueError('继续记录与目标或数据包不一致；不能切换目标继续写入') else: tables=query("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name!='d1_migrations'") if any(t['name'] not in allowed for t in tables):raise ValueError('目标数据库已有其他系统表，请选择新数据库') for table in tables: if query('SELECT count(*) AS n FROM "'+table['name']+'"')[0]['n']:raise ValueError('目标数据库非空；为避免覆盖，请新建数据库') if s3.list_objects_v2(Bucket=bucket_name,MaxKeys=1).get('KeyCount',0):raise ValueError('目标图片桶非空，请新建空桶') state={'target':target,'schema':[],'tables':{},'objects':{},'complete':False};save_state(state_path,state) # Schema statements are replayable. No DROP or DELETE is issued on the target. for definition in definitions: name=definition['name'] if name not in state['schema']: sql=re.sub(r'^CREATE\s+TABLE\s+', 'CREATE TABLE IF NOT EXISTS ',definition['createSql'],count=1,flags=re.I) query(sql) remote=query('PRAGMA table_info("'+name+'")') if [r['name'] for r in remote]!=definition['columns']:raise ValueError('目标表结构不一致：'+name) for index in definition['indexSql']: query(re.sub(r'^(CREATE\s+(?:UNIQUE\s+)?INDEX)\s+',r'\1 IF NOT EXISTS ',index,count=1,flags=re.I)) state['schema'].append(name);save_state(state_path,state) for definition in manifest['tables']: name=definition['name'];columns=definition['columns'];cursor=state['tables'].get(name,0);batch_size=min(5,max(1,100//len(columns))) while True: rows=source.execute('SELECT rowid AS __position,* FROM "'+name+'" WHERE rowid>? ORDER BY rowid LIMIT ?',(cursor,batch_size)).fetchall() if not rows:break # Keep API request bodies bounded for large reason/audit text. selected=[];size=0 for row in rows: current=len(json.dumps(dict(row),ensure_ascii=False).encode('utf-8')) if selected and size+current>1024*1024:break selected.append(row);size+=current quoted=','.join('"'+c+'"' for c in columns);placeholders=','.join('('+','.join('?' for _ in columns)+')' for _ in selected) params=[row[c] for row in selected for c in columns] query('INSERT OR IGNORE INTO "'+name+'"('+quoted+') VALUES '+placeholders,params) ids=[row['id'] for row in selected];stored=query('SELECT * FROM "'+name+'" WHERE id IN ('+','.join('?' for _ in ids)+')',ids);actual={r['id']:r for r in stored} for row in selected: if actual.get(row['id'])!={c:row[c] for c in columns}:raise ValueError('迁入记录冲突或内容不一致：'+name) cursor=selected[-1]['__position'];state['tables'][name]=cursor;save_state(state_path,state) print('数据库：'+name+' 已处理 '+str(cursor)+' 行',flush=True) count=query('SELECT count(*) AS n FROM "'+name+'"')[0]['n'] if count!=manifest['counts'][name]:raise ValueError('目标记录数不一致：'+name) inventory=[json.loads(line) for line in (root/'objects.jsonl').read_text('utf-8').splitlines() if line] def verify_object(item): result=s3.get_object(Bucket=bucket_name,Key=item['key']);digest=hashlib.sha256();total=0 try: while True: part=result['Body'].read(1024*1024) if not part:break total+=len(part);digest.update(part) finally:result['Body'].close() if total!=item['size'] or digest.hexdigest()!=item['sha256']:raise ValueError('目标文件内容不一致：'+item['key']) for i,item in enumerate(inventory): key=item['key'];p=PurePosixPath(key) if p.is_absolute() or '..' in p.parts or '\\' in key:raise ValueError('文件路径无效') local=root/'objects'/Path(*p.parts) if local.stat().st_size!=item['size'] or file_digest(local)!=item['sha256']:raise ValueError('本地迁移文件已变化：'+key) if state['objects'].get(key)!=item['sha256']: s3.upload_file(str(local),bucket_name,key,ExtraArgs={'ContentType':item['type']}) verify_object(item);state['objects'][key]=item['sha256'];save_state(state_path,state) print('图片/备份：'+str(i+1)+' / '+str(len(inventory)),flush=True) # Ensure no transient sessions/OTP or unexpected extra files were introduced. for definition in manifest['emptyTables']: if query('SELECT count(*) AS n FROM "'+definition['name']+'"')[0]['n']:raise ValueError('目标有活动登录或迁移任务，请暂停目标网站后核对') keys=set();token=None while True: args={'Bucket':bucket_name,'MaxKeys':1000} if token:args['ContinuationToken']=token result=s3.list_objects_v2(**args);keys.update(o['Key'] for o in result.get('Contents',[])) if not result.get('IsTruncated'):break token=result['NextContinuationToken'] if keys!={item['key'] for item in inventory}:raise ValueError('目标文件清单不一致') state['complete']=True;save_state(state_path,state);source.close() print('全部数据和文件校验完成。可部署网站并逐个验证原账号、权限和产品。',flush=True) return state if __name__=='__main__': parser=argparse.ArgumentParser(description='将瀚境迁移数据导入你自己的空 Cloudflare D1/R2') parser.add_argument('directory');parser.add_argument('--account-id',required=True);parser.add_argument('--database-id',required=True);parser.add_argument('--bucket',required=True);args=parser.parse_args() if not re.fullmatch('[a-f0-9]{32}',args.account_id) or not re.fullmatch('[a-fA-F0-9-]{36}',args.database_id) or not re.fullmatch('[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]',args.bucket):raise SystemExit('请填写正确的账号 ID、D1 ID 和桶名') token=getpass.getpass('目标 D1 API Token：');access_id=getpass.getpass('目标 R2 S3 Access Key ID：');access_secret=getpass.getpass('目标 R2 S3 Secret Access Key：') endpoint='https://api.cloudflare.com/client/v4/accounts/'+args.account_id+'/d1/database/'+args.database_id+'/query' def query(sql,params=None): body=json.dumps({'sql':sql,'params':params or []},ensure_ascii=False).encode('utf-8');request=urllib.request.Request(endpoint,data=body,headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'}) try: with urllib.request.urlopen(request,timeout=60) as result:data=json.load(result) except urllib.error.HTTPError as error:raise RuntimeError('D1 请求失败，HTTP '+str(error.code)) from None if not data.get('success') or any(not r.get('success') for r in data.get('result',[])):raise RuntimeError('D1 操作未成功，请检查权限或服务限额；可使用同一目录继续') return [row for result in data['result'] for row in result.get('results',[])] try: import boto3 from botocore.config import Config s3=boto3.client('s3',endpoint_url='https://'+args.account_id+'.r2.cloudflarestorage.com',aws_access_key_id=access_id,aws_secret_access_key=access_secret,region_name='auto',config=Config(signature_version='s3v4',retries={'max_attempts':3})) migrate(args.directory,args.account_id,args.database_id,args.bucket,query,s3) except ImportError:raise SystemExit('请先运行：python -m pip install boto3') except Exception as error:raise SystemExit('迁入已停止，继续记录已保留：'+str(error))