108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import sqlite3
|
|
|
|
from database.paths import DATABASE_CODE_ROOT, PROJECT_ROOT
|
|
|
|
|
|
DATABASE_LAYOUT = {
|
|
'L1': ('L1', 'L1.db'),
|
|
'L2': ('L2', 'L2.db'),
|
|
'L3': ('L3', 'L3.db'),
|
|
'Web': ('Web', 'Web_App.sqlite'),
|
|
}
|
|
|
|
|
|
def _sha256(path):
|
|
digest = hashlib.sha256()
|
|
with path.open('rb') as file:
|
|
for chunk in iter(lambda: file.read(1024 * 1024), b''):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _sqlite_copy(source, target):
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
source_db = sqlite3.connect(str(source))
|
|
target_db = sqlite3.connect(str(target))
|
|
try:
|
|
source_db.backup(target_db)
|
|
result = target_db.execute('PRAGMA quick_check').fetchone()[0]
|
|
if result != 'ok':
|
|
raise RuntimeError(f'{target.name} quick_check failed: {result}')
|
|
finally:
|
|
source_db.close()
|
|
target_db.close()
|
|
|
|
|
|
def migrate_data(source_root, target_root, include_imports=False):
|
|
source_root = Path(source_root).expanduser().resolve()
|
|
target_root = Path(target_root).expanduser().resolve()
|
|
if source_root == target_root:
|
|
raise ValueError('Source and target data roots must differ')
|
|
|
|
target_root.mkdir(parents=True, exist_ok=True)
|
|
manifest = {
|
|
'created_at': datetime.now(timezone.utc).isoformat(),
|
|
'source_root': str(source_root),
|
|
'target_root': str(target_root),
|
|
'databases': {},
|
|
}
|
|
for name, (directory, filename) in DATABASE_LAYOUT.items():
|
|
source = source_root / directory / filename
|
|
target = target_root / directory / filename
|
|
if not source.exists():
|
|
raise FileNotFoundError(f'Missing source database: {source}')
|
|
_sqlite_copy(source, target)
|
|
manifest['databases'][name] = {
|
|
'path': str(target),
|
|
'size_bytes': target.stat().st_size,
|
|
'sha256': _sha256(target),
|
|
'quick_check': 'ok',
|
|
}
|
|
|
|
if include_imports:
|
|
legacy_imports = PROJECT_ROOT / 'output_arena'
|
|
if legacy_imports.exists():
|
|
destination = target_root / 'imports'
|
|
shutil.copytree(
|
|
legacy_imports,
|
|
destination,
|
|
dirs_exist_ok=True,
|
|
)
|
|
manifest['imports'] = str(destination)
|
|
|
|
manifest_path = target_root / 'migration-manifest.json'
|
|
manifest_path.write_text(
|
|
json.dumps(manifest, ensure_ascii=True, indent=2),
|
|
encoding='utf-8',
|
|
)
|
|
return manifest_path
|
|
|
|
|
|
def _parse_args():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
'--source',
|
|
default=str(DATABASE_CODE_ROOT),
|
|
help='Existing runtime data root',
|
|
)
|
|
parser.add_argument('--target', required=True)
|
|
parser.add_argument('--include-imports', action='store_true')
|
|
return parser.parse_args()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
args = _parse_args()
|
|
path = migrate_data(
|
|
args.source,
|
|
args.target,
|
|
include_imports=args.include_imports,
|
|
)
|
|
print(f'Migration completed: {path}')
|
|
|