#!/usr/bin/env python3 """同盘Blob迁移：默认预演，分阶段执行，不触碰凭据或业务内容归档。""" import argparse import configparser import hashlib import json import os from pathlib import Path import pwd import stat import subprocess import uuid SOURCE = Path('/media/ztl/LinuxDate/files/我的文件') TARGET = Path('/media/ztl/LinuxDate/files/cloudreve') STATE = Path('/root/AI/cloudreve/private/blob-migration/plan.json') EXPECTED_UUID = '8b9cbec5-c4d7-466f-acea-d1c9531ccada' POLICY = 2 USER = 1 COUNT = 101 def require(condition, message): if not condition: raise RuntimeError(message) def sql_hex(value): return 'CONVERT(0x' + value.encode('utf-8').hex() + ' USING utf8mb4)' def mysql(sql): conf = configparser.ConfigParser(interpolation=None) conf.read('/opt/cloudreve/data/conf.ini') db = conf['Database'] # 密码只通过标准输入传递，不进入进程参数或输出。 options = '[client]\nuser=' + db['User'] + '\npassword=' + db['Password'] + '\n' cmd = ['docker', 'exec', '-i', '1Panel-mysql-5qkq', 'mysql', '--defaults-extra-file=/dev/stdin', '--default-character-set=utf8mb4', '--protocol=TCP', '-h', '127.0.0.1', '-D', db['Name'], '--batch', '--skip-column-names', '-e', sql] result = subprocess.run(cmd, input=options, text=True, capture_output=True, timeout=60) require(result.returncode == 0, '数据库操作失败：' + result.stderr[:1000]) return result.stdout.rstrip('\n') def rows(sql): return [line.split('\t') for line in mysql(sql).splitlines()] def check_environment(): mount_uuid = subprocess.check_output( ['findmnt', '-rn', '-M', '/media/ztl/LinuxDate', '-o', 'UUID'], text=True).strip() require(mount_uuid == EXPECTED_UUID, '外盘UUID不符') for path in (SOURCE, TARGET): check_path(path) require(path.is_dir(), '源或目标目录缺失') require(SOURCE.stat().st_dev == TARGET.stat().st_dev, '仅支持同盘硬链接迁移') rule = mysql('SELECT HEX(dir_name_rule),HEX(file_name_rule) FROM storage_policies WHERE id=2;').split('\t') require(len(rule) == 2, '策略不存在') require(bytes.fromhex(rule[0]).decode() == str(TARGET) + '/{uid}/{path}', 'Blob目录模板已改变') require(bytes.fromhex(rule[1]).decode() == '{uuid}_{originname}', 'Blob名称模板已改变') def check_path(path): # 每一层都拒绝符号链接，避免越界落盘或清理。 path = Path(path) require(path.is_absolute(), '需要绝对路径') require('..' not in path.parts, '路径不能含父级跳转') for parent in (path, *path.parents): require(not parent.is_symlink(), '拒绝符号链接路径') def safe_name(name): require(bool(name) and name not in ('.', '..') and '/' not in name and '\x00' not in name, '不安全的逻辑文件名') return name def fingerprint(): # 原实体引用、用户容量和逻辑结构必须在迁移前后完全不变。 query = ('SELECT HEX(CAST(JSON_ARRAY(id,type,HEX(name),size,primary_entity,is_symbolic,props,' 'file_children,storage_policy_files,owner_id) AS CHAR)) FROM files ORDER BY id; ' 'SELECT file_id,entity_id FROM file_entities ORDER BY file_id,entity_id; ' 'SELECT id,storage FROM users ORDER BY id; ' 'SELECT id,type,size,reference_count,upload_session_id,recycle_options,storage_policy_entities,created_by ' 'FROM entities ORDER BY id;') return hashlib.sha256(mysql(query).encode()).hexdigest() def logical_files(): data = rows('SELECT id,HEX(name),type,IFNULL(file_children,0),owner_id,IFNULL(primary_entity,0) FROM files;') return {int(r[0]): {'name': bytes.fromhex(r[1]).decode(), 'type': int(r[2]), 'parent': int(r[3]), 'owner': int(r[4]), 'entity': int(r[5])} for r in data} def logical_parent(file_id, files): cur = files[file_id]['parent'] result = [] seen = set() while cur: require(cur not in seen and cur in files, '逻辑目录循环或缺失') seen.add(cur) node = files[cur] require(node['owner'] == USER and node['type'] == 1, '父目录身份不符') if node['parent']: result.append(safe_name(node['name'])) cur = node['parent'] return list(reversed(result)) def save_plan(plan): STATE.parent.mkdir(mode=0o700, parents=True, exist_ok=True) temp = STATE.with_suffix('.tmp') with temp.open('w', encoding='utf-8') as out: json.dump(plan, out, ensure_ascii=False, indent=2) out.flush() os.fsync(out.fileno()) os.chmod(temp, 0o600) os.replace(temp, STATE) def load_plan(): return json.loads(STATE.read_text(encoding='utf-8')) def stopped(): for unit in ('cloudreve.service', 'smbd.service', 'nmbd.service'): require(subprocess.run(['systemctl', 'is-active', '--quiet', unit]).returncode != 0, '切换前必须停止Cloudreve及SMB') require(mysql('SELECT COUNT(*) FROM tasks WHERE status IN ("queued","processing","suspending");') == '0', '存在未完成任务') def plan_stage(apply): require(not STATE.exists(), '计划已存在，不重新生成UUID') check_environment() files = logical_files() raw = rows('SELECT id,HEX(source),size,reference_count,storage_policy_entities,IFNULL(created_by,0) FROM entities ORDER BY id;') require(len(raw) == COUNT, '实体数量变化') physical = set() for base, dirs, names in os.walk(SOURCE, followlinks=False): for name in dirs + names: p = Path(base) / name check_path(p) require(p.is_dir() or p.is_file(), '存在特殊文件') physical.update(str(Path(base) / name) for name in names) mappings = [] refs = {v['entity']: k for k, v in files.items() if v['type'] == 0} require(len(refs) == COUNT, '文件引用数量变化') for r in raw: eid, size = int(r[0]), int(r[2]) require(int(r[3]) == 1 and int(r[4]) == POLICY and int(r[5]) == USER, '实体引用或所属策略不符') src = Path(bytes.fromhex(r[1]).decode()) require(src.is_relative_to(SOURCE) and src != SOURCE, '实体不在指定源树') check_path(src) st = src.stat() require(stat.S_ISREG(st.st_mode) and st.st_nlink == 1 and st.st_size == size, '源实体类型/链接数/大小不符') fid = refs[eid] require(files[fid]['owner'] == USER, '用户不符') dst = TARGET / str(USER) for folder in logical_parent(fid, files): dst /= folder blob_name = str(uuid.uuid4()) + '_' + safe_name(files[fid]['name']) require(len(blob_name.encode()) <= 255, 'Blob名称超过ext4限制') dst /= blob_name require(not dst.exists(), 'Blob目标冲突') check_path(dst) mappings.append({'entity': eid, 'file': fid, 'src': str(src), 'dst': str(dst), 'size': size, 'dev': st.st_dev, 'ino': st.st_ino, 'mtime_ns': st.st_mtime_ns, 'uid': st.st_uid, 'gid': st.st_gid, 'mode': stat.S_IMODE(st.st_mode)}) require(physical == {m['src'] for m in mappings}, '源树包含未登记文件或遗漏实体') plan = {'version': 1, 'phase': 'planned', 'invariant': fingerprint(), 'mappings': mappings} if apply: save_plan(plan) print(f'预检通过：{len(mappings)}个文件，{sum(m["size"] for m in mappings)}字节，目标按策略模板生成。') def check_stat(m, path): check_path(path) s = Path(path).stat() require((s.st_dev, s.st_ino, s.st_size, s.st_mtime_ns) == (m['dev'], m['ino'], m['size'], m['mtime_ns']), '实体内容身份或修改时间发生变化') return s def db_sources(): return {int(r[0]): bytes.fromhex(r[1]).decode() for r in rows('SELECT id,HEX(source) FROM entities;')} def link_stage(): stopped() check_environment() plan = load_plan() require(fingerprint() == plan['invariant'], '网盘结构或实体引用变化') require(db_sources() == {m['entity']: m['src'] for m in plan['mappings']}, '数据库不处于迁移前状态') for m in plan['mappings']: check_stat(m, m['src']) dst = Path(m['dst']) require(dst.is_relative_to(TARGET / str(USER)), '目标越界') check_path(dst) dst.parent.mkdir(parents=True, exist_ok=True, mode=0o700) if not dst.exists(): os.link(m['src'], dst, follow_symlinks=False) check_stat(m, dst) require(os.path.samefile(m['src'], dst), '新旧实体不是同一个inode') os.sync() plan['phase'] = 'linked' save_plan(plan) print('全部新Blob路径已建立；与源路径为同一设备和inode，内容相同。旧路径尚保留。') def commit_stage(): stopped() plan = load_plan() require(plan['phase'] == 'linked', '先建立并验证链接') require(fingerprint() == plan['invariant'], '迁移前逻辑结构变化') for m in plan['mappings']: check_stat(m, m['src']) check_stat(m, m['dst']) # 一条会话一个事务，任何条件不满足则NOT NULL断言失败并回滚。 sql = ['CREATE TEMPORARY TABLE migration_guard (ok INT NOT NULL);', 'START TRANSACTION;', 'SELECT id FROM entities ORDER BY id FOR UPDATE;', f'INSERT INTO migration_guard SELECT IF((SELECT COUNT(*) FROM entities)={COUNT},1,NULL);'] for m in plan['mappings']: sql.append(f'UPDATE entities SET source={sql_hex(m["dst"])} WHERE id={m["entity"]} ' f'AND BINARY source=BINARY {sql_hex(m["src"])} AND size={m["size"]} ' f'AND reference_count=1 AND storage_policy_entities={POLICY};') sql.append('INSERT INTO migration_guard VALUES(IF(ROW_COUNT()=1,1,NULL));') sql.append('COMMIT;') mysql('\n'.join(sql)) require(db_sources() == {m['entity']: m['dst'] for m in plan['mappings']}, '提交后实体路径不符') require(fingerprint() == plan['invariant'], '提交后逻辑结构或容量变化') plan['phase'] = 'committed' save_plan(plan) print('101个数据库实体路径已原子切换；文件名、目录、ID、引用及容量完全保持。') def cleanup_stage(): stopped() plan = load_plan() require(plan['phase'] in ('committed', 'cleaned'), '数据库未提交，禁止清理') require(fingerprint() == plan['invariant'], '逻辑结构已变化，停止清理') require(db_sources() == {m['entity']: m['dst'] for m in plan['mappings']}, '数据库仍引用旧路径') # 全部逐个检查后才移除明确计划中的旧路径；不删除空目录或未登记文件。 for m in plan['mappings']: require(Path(m['src']).is_relative_to(SOURCE), '源路径越界') check_stat(m, m['dst']) if Path(m['src']).exists(): check_stat(m, m['src']) require(os.path.samefile(m['src'], m['dst']), '不能确认原实体仍存活') for m in plan['mappings']: src = Path(m['src']) if src.exists(): # 明确确认后的绝对路径，只取消旧目录项，新路径仍引用同一实体。 src.unlink() account = pwd.getpwnam('cloudreve') for m in plan['mappings']: dst = Path(m['dst']) require(dst.stat().st_nlink == 1, '仍有其他链接，拒绝调整实体所有权') os.chown(dst, account.pw_uid, account.pw_gid) subprocess.run(['setfacl', '-b', '--', str(dst)], check=True) os.chmod(dst, 0o600) cur = dst.parent while cur != TARGET: check_path(cur) os.chown(cur, account.pw_uid, account.pw_gid) os.chmod(cur, 0o700) cur = cur.parent os.sync() plan['phase'] = 'cleaned' save_plan(plan) print('101个旧文件路径已移除；新Blob实体已交给cloudreve用户管理，空源目录保留。') def verify_stage(): plan = load_plan() require(fingerprint() == plan['invariant'], '逻辑结构或容量变化') require(db_sources() == {m['entity']: m['dst'] for m in plan['mappings']}, '路径不符') for m in plan['mappings']: check_stat(m, m['dst']) require(not Path(m['src']).exists(), '仍有旧文件路径') print('通过：101个Blob实体仍为原inode/大小/mtime，旧文件路径全部消失，逻辑结构和容量不变。') def main(): parser = argparse.ArgumentParser() parser.add_argument('stage', choices=['plan', 'link', 'commit', 'cleanup', 'verify'], default='plan', nargs='?') parser.add_argument('--apply', action='store_true') args = parser.parse_args() require(os.geteuid() == 0, '请在专用虚拟环境中使用root运行') if args.stage == 'plan': plan_stage(args.apply) elif args.stage == 'verify': verify_stage() else: require(args.apply, '修改阶段必须显式--apply') {'link': link_stage, 'commit': commit_stage, 'cleanup': cleanup_stage}[args.stage]() if __name__ == '__main__': main()