File size: 5,418 Bytes
abd39d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """Copy an old layout without deleting its files. Preview by default; --apply writes.
Use --flat-output only if root-level old output belongs to the active SQLite book.
"""
import argparse
import hashlib
import shutil
import sqlite3
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from storage.store import _safe_filename
def fingerprint(path):
h = hashlib.sha256()
with path.open('rb') as f:
for part in iter(lambda: f.read(1024 * 1024), b''):
h.update(part)
return h.digest()
def same_content(left, right):
if left.suffix == right.suffix == '.sqlite3':
def database_hash(path):
digest = hashlib.sha256()
with sqlite3.connect(f'{path.resolve().as_uri()}?mode=ro', uri=True) as conn:
for statement in conn.iterdump():
digest.update(statement.encode('utf-8'))
return digest.digest()
return database_hash(left) == database_hash(right)
return fingerprint(left) == fingerprint(right)
def build_plan(source, destination, flat_output=False):
source, destination = source.resolve(), destination.resolve()
if source == destination:
raise ValueError('Thư mục nguồn và đích phải khác nhau.')
mappings = [('translator/models', 'models/translation'), ('tts/voice', 'models/voices'),
('translator/data', 'data')]
plan = []
for old, new in mappings:
for f in sorted((source / old).rglob('*')):
if f.is_symlink():
raise ValueError(f'Không tự di chuyển symlink: {f}')
if f.is_file() and f.suffix != '.log' and not f.name.endswith(('-wal', '-shm')):
plan.append((f, destination / new / f.relative_to(source / old)))
db = source / 'translator/data/novel.sqlite3'
title = None
if db.is_file() and db.stat().st_size:
with sqlite3.connect(f'{db.as_uri()}?mode=ro', uri=True) as conn:
row = conn.execute('SELECT title FROM book LIMIT 1').fetchone()
title = _safe_filename(row[0]) if row else None
for kind in ('output_text', 'output_audio'):
base = source / kind
for f in sorted(base.rglob('*')):
if f.is_symlink():
raise ValueError(f'Không tự di chuyển symlink: {f}')
if not f.is_file():
continue
rel = f.relative_to(base)
is_legacy = len(rel.parts) == 1 or rel.parts[0] in ('groups', 'merged')
if is_legacy:
if not flat_output or not title:
continue
name = f.name
if kind == 'output_audio':
# Drop ID aliases and stale fingerprints; chapter-position files are canonical.
if name.startswith('chapter_') or name.endswith('.sources.json'):
continue
is_group = len(rel.parts) > 1 or name.count('_') >= 2
rel = Path(title) / ('audio_gop' if is_group else '') / name
else:
if name.endswith('_full.txt'):
name = 'toan_bo_truyen.txt'
rel = Path(title) / name
plan.append((f, destination / kind / rel))
return plan
def migrate(source, destination, apply=False, flat_output=False):
plan = build_plan(source, destination, flat_output)
targets = {}
# Check ALL conflicts before copying, rather than half-migrating a database.
for old, new in plan:
if not new.resolve().is_relative_to(destination.resolve()):
raise ValueError(f'Đích nằm ngoài thư mục cài đặt: {new}')
if new in targets and not same_content(old, targets[new]):
raise ValueError(f'Hai nguồn khác nhau trùng đích: {new}')
targets[new] = old
if new.exists() and (not new.is_file() or not same_content(old, new)):
raise ValueError(f'Đích đã có dữ liệu khác; không ghi đè: {new}')
for new, old in targets.items():
print(f'{old} -> {new}')
if apply and not new.exists():
new.parent.mkdir(parents=True, exist_ok=True)
if old.suffix == '.sqlite3':
# SQLite backup includes committed WAL pages; the old app must be stopped first.
with sqlite3.connect(f'{old.as_uri()}?mode=ro', uri=True) as src:
with sqlite3.connect(new) as dst:
src.backup(dst)
else:
shutil.copy2(old, new)
print(f'{"Đã sao chép" if apply else "Xem trước"}: {len(targets)} file. Nguồn không bị xóa.')
return len(targets)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--source', required=True, type=Path)
parser.add_argument('--destination', type=Path, default=ROOT)
parser.add_argument('--apply', action='store_true')
parser.add_argument('--flat-output', action='store_true', help='Gán output phẳng cũ cho truyện hiện tại; chỉ bật khi chắc chắn không trộn truyện.')
args = parser.parse_args()
if not args.source.is_dir():
parser.error('Không tìm thấy thư mục nguồn.')
migrate(args.source, args.destination, args.apply, args.flat_output)
if __name__ == '__main__':
main()
|