825 lines
33 KiB
Python
825 lines
33 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Portable backup/restore support helpers for UmbraAudio.
|
|
|
|
This module uses only the Python standard library. It snapshots global Umbra
|
|
state with SQLite's backup API, restores it atomically, rewrites machine-local
|
|
paths in a restored copy, and performs basic integrity checks.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def is_database(path: Path) -> bool:
|
|
return path.name.endswith((".db", ".db-wal", ".db-shm"))
|
|
|
|
|
|
def is_sqlite_file(path: Path) -> bool:
|
|
try:
|
|
with path.open("rb") as handle:
|
|
return handle.read(16) == b"SQLite format 3\x00"
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def safe_relative_path(value: str) -> Path:
|
|
path = Path(value)
|
|
if path.is_absolute() or ".." in path.parts:
|
|
raise RuntimeError(f"unsafe relative path in backup metadata: {value!r}")
|
|
return path
|
|
|
|
|
|
def sql_identifier(value: str) -> str:
|
|
return '"' + value.replace('"', '""') + '"'
|
|
|
|
|
|
def sqlite_backup(source: Path, destination: Path) -> str:
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
source_uri = f"file:{source}?mode=ro"
|
|
source_db = sqlite3.connect(source_uri, uri=True)
|
|
destination_db = sqlite3.connect(destination)
|
|
try:
|
|
source_db.backup(destination_db)
|
|
result = destination_db.execute("PRAGMA integrity_check").fetchone()[0]
|
|
if result != "ok":
|
|
raise RuntimeError(f"integrity_check failed for {source}: {result}")
|
|
finally:
|
|
destination_db.close()
|
|
source_db.close()
|
|
return "ok"
|
|
|
|
|
|
def copy_tree_files(
|
|
source: Path,
|
|
destination: Path,
|
|
skip_databases: bool = True,
|
|
reject_symlinks: bool = False,
|
|
) -> list[str]:
|
|
copied: list[str] = []
|
|
for path in sorted(source.rglob("*")):
|
|
relative = path.relative_to(source)
|
|
target = destination / relative
|
|
if path.is_symlink():
|
|
if reject_symlinks:
|
|
raise RuntimeError(f"global state symlink is not allowed: {relative}")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
os.symlink(os.readlink(path), target)
|
|
elif path.is_dir():
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
elif path.is_file():
|
|
if skip_databases and is_database(path):
|
|
continue
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(path, target)
|
|
copied.append(str(relative))
|
|
return copied
|
|
|
|
|
|
def snapshot_global(source: Path, output: Path, original_root: Path) -> None:
|
|
if output.exists():
|
|
raise RuntimeError(f"global-state output already exists: {output}")
|
|
if not source.exists():
|
|
raise RuntimeError(f"global state root does not exist: {source}")
|
|
if source == output or source in output.parents:
|
|
raise RuntimeError("global state output must be outside the source root")
|
|
|
|
temporary = Path(tempfile.mkdtemp(prefix="umbra-global-snapshot-"))
|
|
try:
|
|
files = copy_tree_files(source, temporary, skip_databases=True)
|
|
databases: list[dict[str, Any]] = []
|
|
for path in sorted(source.rglob("*.db")):
|
|
if not path.is_file():
|
|
continue
|
|
relative = path.relative_to(source)
|
|
target = temporary / relative
|
|
status = sqlite_backup(path, target)
|
|
databases.append(
|
|
{"path": str(relative), "integrity_check": status, "sha256": sha256(target)}
|
|
)
|
|
files.append(str(relative))
|
|
|
|
file_hashes = {
|
|
relative: sha256(temporary / relative)
|
|
for relative in sorted(files)
|
|
if (temporary / relative).is_file()
|
|
}
|
|
manifest = {
|
|
"schema": "umbra.global_state/v1",
|
|
"original_root": str(original_root),
|
|
"source_root": str(source),
|
|
"files": sorted(files),
|
|
"file_hashes": file_hashes,
|
|
"databases": databases,
|
|
}
|
|
(temporary / "MANIFEST.json").write_text(
|
|
json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
(temporary / "GLOBAL-STATE-SHA256SUMS").write_text(
|
|
"".join(f"{sha256(temporary / relative)} {relative}\n" for relative in sorted(files)),
|
|
encoding="utf-8",
|
|
)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copytree(temporary, output, symlinks=True)
|
|
finally:
|
|
shutil.rmtree(temporary, ignore_errors=True)
|
|
|
|
|
|
def _replace_prefix(value: Any, old_root: str, new_root: str) -> Any:
|
|
if isinstance(value, str) and old_root and value.startswith(old_root):
|
|
return new_root + value[len(old_root) :]
|
|
return value
|
|
|
|
|
|
def assert_no_old_prefixes(
|
|
connection: sqlite3.Connection,
|
|
old_roots: list[str],
|
|
) -> None:
|
|
for table_row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'"):
|
|
table = table_row[0]
|
|
if table.startswith("sqlite_"):
|
|
continue
|
|
table_identifier = sql_identifier(table)
|
|
for column in connection.execute(f"PRAGMA table_info({table_identifier})").fetchall():
|
|
column_name = column[1]
|
|
column_identifier = sql_identifier(column_name)
|
|
if (column[2] or "").upper() not in {"TEXT", ""}:
|
|
continue
|
|
try:
|
|
values = connection.execute(
|
|
f"SELECT {column_identifier} FROM {table_identifier} "
|
|
f"WHERE {column_identifier} IS NOT NULL"
|
|
).fetchall()
|
|
except sqlite3.Error:
|
|
continue
|
|
for (value,) in values:
|
|
if isinstance(value, str) and any(
|
|
old_root and value.startswith(old_root) for old_root in old_roots
|
|
):
|
|
raise RuntimeError(f"old path prefix remains in {table}.{column_name}: {value}")
|
|
|
|
|
|
def migrate_database_paths(database: Path, replacements: list[tuple[str, str]]) -> None:
|
|
connection = sqlite3.connect(database)
|
|
try:
|
|
tables = {
|
|
row[0]
|
|
for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
|
}
|
|
for table, columns in {
|
|
"voice_registry": ("voice_ref_path", "vmatrix_path", "sample_audio_path"),
|
|
"library_speakers": ("voice_ref_path",),
|
|
"library_utterances": ("voice_ref_path",),
|
|
}.items():
|
|
if table not in tables:
|
|
continue
|
|
for column in columns:
|
|
try:
|
|
rows = connection.execute(
|
|
f'SELECT rowid, "{column}" FROM "{table}" WHERE "{column}" IS NOT NULL'
|
|
).fetchall()
|
|
except sqlite3.Error:
|
|
continue
|
|
for rowid, value in rows:
|
|
replacement = value
|
|
for old_root, new_root in replacements:
|
|
replacement = _replace_prefix(replacement, old_root, new_root)
|
|
if replacement != value:
|
|
connection.execute(
|
|
f'UPDATE "{table}" SET "{column}" = ? WHERE rowid = ?',
|
|
(replacement, rowid),
|
|
)
|
|
connection.commit()
|
|
assert_no_old_prefixes(connection, [old for old, _ in replacements if old])
|
|
result = connection.execute("PRAGMA integrity_check").fetchone()[0]
|
|
if result != "ok":
|
|
raise RuntimeError(f"integrity_check failed after path migration: {database}: {result}")
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def migrate_project_database(database: Path, old_root: str, new_root: str) -> None:
|
|
connection = sqlite3.connect(database)
|
|
try:
|
|
tables = {
|
|
row[0]
|
|
for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
|
if not row[0].startswith("sqlite_")
|
|
}
|
|
for table in tables:
|
|
table_identifier = sql_identifier(table)
|
|
columns = connection.execute(f"PRAGMA table_info({table_identifier})").fetchall()
|
|
for column in columns:
|
|
column_name = column[1]
|
|
column_identifier = sql_identifier(column_name)
|
|
if (column[2] or "").upper() not in {"TEXT", ""}:
|
|
continue
|
|
try:
|
|
query = (
|
|
f"SELECT rowid, {column_identifier} FROM {table_identifier} "
|
|
f"WHERE {column_identifier} IS NOT NULL"
|
|
)
|
|
rows = connection.execute(query).fetchall()
|
|
except sqlite3.Error:
|
|
continue
|
|
for rowid, value in rows:
|
|
replacement = _replace_prefix(value, old_root, new_root)
|
|
if replacement != value:
|
|
connection.execute(
|
|
f"UPDATE {table_identifier} SET {column_identifier} = ? "
|
|
"WHERE rowid = ?",
|
|
(replacement, rowid),
|
|
)
|
|
connection.commit()
|
|
assert_no_old_prefixes(connection, [old_root] if old_root else [])
|
|
result = connection.execute("PRAGMA integrity_check").fetchone()[0]
|
|
if result != "ok":
|
|
raise RuntimeError(
|
|
f"integrity_check failed after project path migration: {database}: {result}"
|
|
)
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def _validate_target(path: Path, force: bool) -> Path:
|
|
if path.is_symlink():
|
|
raise RuntimeError(f"refusing symlink target: {path}")
|
|
if path.exists() and not path.is_dir():
|
|
raise RuntimeError(f"target exists and is not a directory: {path}")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if path.exists() and any(path.iterdir()) and not force:
|
|
raise RuntimeError(f"target is not empty: {path}; use --force")
|
|
return path
|
|
|
|
|
|
def _commit_directory(temporary: Path, target: Path, force: bool) -> Path | None:
|
|
_validate_target(target, force)
|
|
previous: Path | None = None
|
|
if target.exists():
|
|
previous = target.with_name(f"{target.name}.old.{int(time.time())}")
|
|
target.rename(previous)
|
|
try:
|
|
temporary.rename(target)
|
|
except Exception:
|
|
if previous is not None and previous.exists() and not target.exists():
|
|
previous.rename(target)
|
|
raise
|
|
return previous
|
|
|
|
|
|
def restore_global(
|
|
backup_dir: Path,
|
|
target: Path,
|
|
force: bool,
|
|
old_project_root: str = "",
|
|
new_project_root: str = "",
|
|
) -> None:
|
|
source = backup_dir / "global-state"
|
|
manifest_path = source / "MANIFEST.json"
|
|
if not manifest_path.exists():
|
|
raise RuntimeError(f"global-state manifest is missing: {manifest_path}")
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
old_root = manifest.get("original_root", "")
|
|
_validate_target(target, force)
|
|
temporary = Path(tempfile.mkdtemp(prefix=f".{target.name}.global-restore.", dir=target.parent))
|
|
try:
|
|
copy_tree_files(
|
|
source,
|
|
temporary,
|
|
skip_databases=False,
|
|
reject_symlinks=True,
|
|
)
|
|
for metadata_name in ("MANIFEST.json", "GLOBAL-STATE-SHA256SUMS"):
|
|
metadata_path = temporary / metadata_name
|
|
if metadata_path.exists():
|
|
metadata_path.unlink()
|
|
for relative in manifest.get("files", []):
|
|
safe_relative_path(relative)
|
|
file_hashes = manifest.get("file_hashes", {})
|
|
for relative, expected_hash in file_hashes.items():
|
|
safe_relative = safe_relative_path(relative)
|
|
path = temporary / safe_relative
|
|
if not path.is_file() or sha256(path) != expected_hash:
|
|
raise RuntimeError(f"global file checksum mismatch: {relative}")
|
|
replacements = [(old_root, str(target))]
|
|
if old_project_root and new_project_root:
|
|
replacements.append((old_project_root, new_project_root))
|
|
for database in manifest.get("databases", []):
|
|
relative = safe_relative_path(database["path"])
|
|
path = temporary / relative
|
|
if not path.is_file() or sha256(path) != database.get("sha256"):
|
|
raise RuntimeError(f"global database checksum mismatch: {relative}")
|
|
migrate_database_paths(path, replacements)
|
|
previous = _commit_directory(temporary, target, force)
|
|
temporary = Path()
|
|
if previous is not None:
|
|
print(f"Previous global state retained at: {previous}")
|
|
finally:
|
|
if temporary and str(temporary) not in {"", "."}:
|
|
shutil.rmtree(temporary, ignore_errors=True)
|
|
print(f"Restored global state to {target}")
|
|
|
|
|
|
def _replace_text(path: Path, replacements: list[tuple[str, str]]) -> list[str]:
|
|
text = path.read_text(encoding="utf-8")
|
|
changed: list[str] = []
|
|
for old, new in replacements:
|
|
if old in text:
|
|
text = text.replace(old, new)
|
|
changed.append(old)
|
|
path.write_text(text, encoding="utf-8")
|
|
return changed
|
|
|
|
|
|
def patch_restored_paths(
|
|
root: Path,
|
|
old_root: str,
|
|
new_root: Path,
|
|
global_root: Path | None = None,
|
|
) -> None:
|
|
changes: list[str] = []
|
|
env_path = root / ".env"
|
|
if env_path.exists() and old_root:
|
|
original = root / ".env.archive-original"
|
|
if not original.exists():
|
|
shutil.copy2(env_path, original)
|
|
changes.extend(_replace_text(env_path, [(old_root, str(new_root))]))
|
|
if env_path.exists() and global_root is not None:
|
|
lines = env_path.read_text(encoding="utf-8").splitlines()
|
|
keys = {"UMBRA_GLOBAL_STATE_DIR", "UMBRA_VOICE_REGISTRY_PATH"}
|
|
lines = [line for line in lines if line.split("=", 1)[0] not in keys]
|
|
lines.extend(
|
|
[
|
|
f"UMBRA_GLOBAL_STATE_DIR={json.dumps(str(global_root))}",
|
|
f"UMBRA_VOICE_REGISTRY_PATH={json.dumps(str(global_root / 'voice_registry.db'))}",
|
|
]
|
|
)
|
|
env_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
changes.append(".env global-state settings")
|
|
|
|
config_path = root / "umbraaudio" / "config.py"
|
|
if config_path.exists() and old_root:
|
|
text = config_path.read_text(encoding="utf-8")
|
|
if "import os\n" not in text:
|
|
text = text.replace("import re\n", "import os\nimport re\n", 1)
|
|
portable_project_base = (
|
|
"PROJECT_BASE = Path(\n"
|
|
' os.environ.get("UMBRA_PROJECT_BASE", '
|
|
"str(Path(__file__).resolve().parents[1]))\n"
|
|
").expanduser()"
|
|
)
|
|
text = text.replace(
|
|
f'PROJECT_BASE = Path("{old_root}")',
|
|
portable_project_base,
|
|
)
|
|
text = text.replace(
|
|
f'"{old_root}/models/Qwen3-TTS-12Hz-1.7B-VoiceDesign"',
|
|
'str(PROJECT_BASE / "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign")',
|
|
)
|
|
text = text.replace(
|
|
f'"{old_root}/models/Qwen3-TTS-12Hz-1.7B-Base"',
|
|
'str(PROJECT_BASE / "models/Qwen3-TTS-12Hz-1.7B-Base")',
|
|
)
|
|
portable_voice_registry = (
|
|
"voice_registry_path: Path = Field(\n"
|
|
" default_factory=lambda: Path(\n"
|
|
" os.environ.get(\n"
|
|
' "UMBRA_VOICE_REGISTRY_PATH",\n'
|
|
' str(Path.home() / ".umbra" / "voice_registry.db"),\n'
|
|
" )\n"
|
|
" ).expanduser()\n"
|
|
")"
|
|
)
|
|
old_voice_field = (
|
|
'voice_registry_path: Path = Field(default=Path.home() / ".umbra" / '
|
|
'"voice_registry.db")'
|
|
)
|
|
text = text.replace(old_voice_field, portable_voice_registry)
|
|
config_path.write_text(text, encoding="utf-8")
|
|
changes.append("umbraaudio/config.py")
|
|
|
|
library_path = root / "umbraaudio" / "library.py"
|
|
if library_path.exists():
|
|
text = library_path.read_text(encoding="utf-8")
|
|
if "import os\n" not in text:
|
|
text = text.replace("import json\n", "import json\nimport os\n", 1)
|
|
portable_library_root = (
|
|
"def _global_state_root() -> Path:\n"
|
|
" return Path(\n"
|
|
' os.environ.get("UMBRA_GLOBAL_STATE_DIR", str(Path.home() / ".umbra"))\n'
|
|
" ).expanduser()\n"
|
|
)
|
|
text = text.replace(
|
|
'DEFAULT_LIBRARY_DB = Path.home() / ".umbra" / "voice_registry.db"\n'
|
|
'DEFAULT_LIBRARY_VOICES = Path.home() / ".umbra" / "library" / "voices"',
|
|
portable_library_root
|
|
+ 'DEFAULT_LIBRARY_DB = _global_state_root() / "voice_registry.db"\n'
|
|
+ 'DEFAULT_LIBRARY_VOICES = _global_state_root() / "library" / "voices"',
|
|
)
|
|
old_db_getter = "def get_library_db_path() -> Path:\n return DEFAULT_LIBRARY_DB"
|
|
new_db_getter = (
|
|
"def get_library_db_path() -> Path:\n return _global_state_root() / "
|
|
'"voice_registry.db"'
|
|
)
|
|
text = text.replace(old_db_getter, new_db_getter)
|
|
old_voices_getter = (
|
|
"def get_library_voices_dir() -> Path:\n return DEFAULT_LIBRARY_VOICES"
|
|
)
|
|
new_voices_getter = (
|
|
"def get_library_voices_dir() -> Path:\n return _global_state_root() / "
|
|
'"library" / "voices"'
|
|
)
|
|
text = text.replace(old_voices_getter, new_voices_getter)
|
|
library_path.write_text(text, encoding="utf-8")
|
|
changes.append("umbraaudio/library.py")
|
|
|
|
server_path = root / "scripts" / "qwen_tts_server.py"
|
|
if server_path.exists() and old_root:
|
|
portable_prompt_dir = (
|
|
"str(Path(__file__).resolve().parents[1] / "
|
|
'"DealHunterShip_4.0e_whole-to-audio/.umbra/voice_prompts")'
|
|
)
|
|
portable_voicedesign = (
|
|
'str(Path(__file__).resolve().parents[1] / "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign")'
|
|
)
|
|
portable_base = (
|
|
'str(Path(__file__).resolve().parents[1] / "models/Qwen3-TTS-12Hz-1.7B-Base")'
|
|
)
|
|
replacements = [
|
|
(
|
|
f'"{old_root}/DealHunterShip_4.0e_whole-to-audio/.umbra/voice_prompts"',
|
|
portable_prompt_dir,
|
|
),
|
|
(
|
|
f'"{old_root}/models/Qwen3-TTS-12Hz-1.7B-VoiceDesign"',
|
|
portable_voicedesign,
|
|
),
|
|
(
|
|
f'"{old_root}/models/Qwen3-TTS-12Hz-1.7B-Base"',
|
|
portable_base,
|
|
),
|
|
]
|
|
changes.extend(_replace_text(server_path, replacements))
|
|
|
|
storage_path = root / "umbraaudio" / "storage.py"
|
|
if storage_path.exists():
|
|
text = storage_path.read_text(encoding="utf-8")
|
|
if "import os\n" not in text:
|
|
text = text.replace("import logging\n", "import logging\nimport os\n", 1)
|
|
if old_root:
|
|
portable_local_root = (
|
|
"DEFAULT_LOCAL_ROOT = Path(os.environ.get("
|
|
'"UMBRA_PROJECT_ROOT", str(Path(__file__).resolve().parents[1]))'
|
|
").expanduser()"
|
|
)
|
|
text = text.replace(
|
|
f'DEFAULT_LOCAL_ROOT = Path("{old_root}")',
|
|
portable_local_root,
|
|
)
|
|
portable_nas_root = (
|
|
"DEFAULT_NAS_ROOT = Path(os.environ.get("
|
|
'"UMBRA_NAS_ROOT", "/Volumes/Umbra/UmbraAudio"'
|
|
")).expanduser()"
|
|
)
|
|
text = text.replace(
|
|
'DEFAULT_NAS_ROOT = Path("/Volumes/Umbra/UmbraAudio")',
|
|
portable_nas_root,
|
|
)
|
|
storage_path.write_text(text, encoding="utf-8")
|
|
changes.append("umbraaudio/storage.py")
|
|
|
|
if old_root:
|
|
escaped_root = json.dumps(str(new_root))[1:-1]
|
|
operational_files = [
|
|
root / "scripts" / "qwen_tts_server.py",
|
|
root / "tts_server" / "qwen_tts_server.py",
|
|
root / "Qwen3-TTS" / "qwen_tts" / "cli" / "demo.py",
|
|
root / "start_server.sh",
|
|
root / "test_voice_clone.py",
|
|
]
|
|
for operational_path in operational_files:
|
|
if operational_path.exists():
|
|
_replace_text(operational_path, [(old_root, escaped_root)])
|
|
|
|
nas_lifecycle = root / "scripts" / "nas_lifecycle.py"
|
|
if nas_lifecycle.exists():
|
|
text = nas_lifecycle.read_text(encoding="utf-8")
|
|
if "import os\n" not in text:
|
|
text = text.replace("import logging\n", "import logging\nimport os\n", 1)
|
|
text = text.replace(
|
|
f'DEFAULT_LOCAL_ROOT = Path("{old_root}")',
|
|
"DEFAULT_LOCAL_ROOT = Path(__file__).resolve().parents[1]",
|
|
)
|
|
portable_nas = (
|
|
"DEFAULT_NAS_ROOT = Path(os.environ.get("
|
|
'"UMBRA_NAS_ROOT", "/Volumes/Umbra/UmbraAudio"'
|
|
")).expanduser()"
|
|
)
|
|
text = text.replace(
|
|
'DEFAULT_NAS_ROOT = Path("/Volumes/Umbra/UmbraAudio")',
|
|
portable_nas,
|
|
)
|
|
text = text.replace(
|
|
'MOUNT_SCRIPT = Path.home() / ".umbra/mount_nas.sh"',
|
|
'MOUNT_SCRIPT = _global_state_root() / "mount_nas.sh"',
|
|
)
|
|
text = text.replace(
|
|
'CONFIG_PATH = Path.home() / ".umbra/nas_config.json"',
|
|
'CONFIG_PATH = _global_state_root() / "nas_config.json"',
|
|
)
|
|
if "def _global_state_root()" not in text:
|
|
global_root_function = (
|
|
"def _global_state_root() -> Path:\n"
|
|
" return Path(os.environ.get(\n"
|
|
' "UMBRA_GLOBAL_STATE_DIR", str(Path.home() / ".umbra")\n'
|
|
" )).expanduser()\n\n"
|
|
"DEFAULT_LOCAL_ROOT ="
|
|
)
|
|
text = text.replace(
|
|
"DEFAULT_LOCAL_ROOT =",
|
|
global_root_function,
|
|
1,
|
|
)
|
|
nas_lifecycle.write_text(text, encoding="utf-8")
|
|
changes.append("scripts/nas_lifecycle.py")
|
|
|
|
sync_voice = root / "tts_server" / "sync_voice_data.sh"
|
|
if sync_voice.exists():
|
|
text = sync_voice.read_text(encoding="utf-8")
|
|
text = text.replace(
|
|
'UMBRA_DATA="${UMBRA_DATA_DIR:-/Volumes/1TB_APL_Storage/UmbraAudio}"',
|
|
'PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd -P)"\n'
|
|
'UMBRA_DATA="${UMBRA_DATA_DIR:-$PROJECT_ROOT}"',
|
|
)
|
|
sync_voice.write_text(text, encoding="utf-8")
|
|
changes.append("tts_server/sync_voice_data.sh")
|
|
|
|
verify_nas = root / "scripts" / "verify_nas.py"
|
|
if verify_nas.exists():
|
|
text = verify_nas.read_text(encoding="utf-8")
|
|
text = text.replace(
|
|
f'Path("{old_root}")',
|
|
"Path(__file__).resolve().parents[1]",
|
|
)
|
|
verify_nas.write_text(text, encoding="utf-8")
|
|
changes.append("scripts/verify_nas.py")
|
|
|
|
embeddings = root / "scripts" / "extract_embeddings.py"
|
|
if embeddings.exists():
|
|
text = embeddings.read_text(encoding="utf-8")
|
|
text = text.replace(
|
|
f'"{old_root}/models/Qwen3-TTS-12Hz-1.7B-Base"',
|
|
'str(project_root / "models/Qwen3-TTS-12Hz-1.7B-Base")',
|
|
)
|
|
text = text.replace(
|
|
f'"{old_root}/models/Qwen3-TTS-12Hz-1.7B-VoiceDesign"',
|
|
'str(project_root / "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign")',
|
|
)
|
|
embeddings.write_text(text, encoding="utf-8")
|
|
changes.append("scripts/extract_embeddings.py")
|
|
|
|
legacy_start = root / "start_server.sh"
|
|
if legacy_start.exists():
|
|
legacy_start.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
"set -Eeuo pipefail\n"
|
|
"printf '%s\\n' 'Legacy Gradio demo launcher is disabled in this restore.'\n"
|
|
"printf '%s\\n' 'Use scripts/start_web_server.sh or the generated service commands.'\n"
|
|
"exit 1\n",
|
|
encoding="utf-8",
|
|
)
|
|
legacy_start.chmod(0o755)
|
|
changes.append("start_server.sh")
|
|
|
|
start_script = root / "scripts" / "start_web_server.sh"
|
|
if start_script.exists():
|
|
web_command = (
|
|
'exec "$PROJ/.venv/bin/python" -c \''
|
|
"from pathlib import Path; from umbraaudio.web import create_app; "
|
|
"from umbraaudio.config import load_config; import uvicorn; "
|
|
'c = load_config(Path(".")); uvicorn.run(create_app(c), '
|
|
'host="0.0.0.0", port=8420, log_level="info")\'\n'
|
|
)
|
|
global_exports = ""
|
|
if global_root is not None:
|
|
global_exports = (
|
|
f"export UMBRA_GLOBAL_STATE_DIR={shlex.quote(str(global_root))}\n"
|
|
"export UMBRA_VOICE_REGISTRY_PATH="
|
|
f"{shlex.quote(str(global_root / 'voice_registry.db'))}\n"
|
|
)
|
|
start_script.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
"set -Eeuo pipefail\n"
|
|
'PROJ="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd -P)"\n'
|
|
'exec >> "$PROJ/logs/web_server.log" 2>&1\n'
|
|
"echo \"[$(date '+%F %T')] web server starting (wrapper pid $$)\"\n"
|
|
'cd "$PROJ"\n' + global_exports + web_command,
|
|
encoding="utf-8",
|
|
)
|
|
start_script.chmod(0o755)
|
|
changes.append("scripts/start_web_server.sh")
|
|
|
|
if old_root:
|
|
for database in sorted(root.rglob("*.db")):
|
|
if not database.is_file() or any(
|
|
part in {".venv", ".python"} for part in database.parts
|
|
):
|
|
continue
|
|
if not is_sqlite_file(database):
|
|
continue
|
|
migrate_project_database(database, old_root, str(new_root))
|
|
|
|
commands = root / "RESTORED-SERVICE-COMMANDS.md"
|
|
web_command = (
|
|
"python -c 'from pathlib import Path; from umbraaudio.web import create_app; "
|
|
"from umbraaudio.config import load_config; import uvicorn; "
|
|
'uvicorn.run(create_app(load_config(Path("."))), host="0.0.0.0", port=8420)\''
|
|
)
|
|
voicedesign_model = json.dumps(str(new_root / "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign"))
|
|
base_model = json.dumps(str(new_root / "models/Qwen3-TTS-12Hz-1.7B-Base"))
|
|
global_exports = ""
|
|
if global_root is not None:
|
|
global_exports = (
|
|
f"export UMBRA_GLOBAL_STATE_DIR={shlex.quote(str(global_root))}\n"
|
|
"export UMBRA_VOICE_REGISTRY_PATH="
|
|
f"{shlex.quote(str(global_root / 'voice_registry.db'))}\n"
|
|
)
|
|
commands.write_text(
|
|
"# Restored UmbraAudio service commands\n\n"
|
|
"These commands are generated for the restore location and are not installed "
|
|
"automatically.\n\n"
|
|
"```bash\n"
|
|
f"cd {json.dumps(str(new_root))}\n"
|
|
'export PATH="$PWD/.venv/bin:$PATH"\n'
|
|
f"export UMBRA_PROJECT_BASE={json.dumps(str(new_root))}\n"
|
|
+ global_exports
|
|
+ f"{web_command}\n"
|
|
f"QWEN_VOICEDESIGN_MODEL={voicedesign_model}\n"
|
|
f"QWEN_BASE_MODEL={base_model}\n"
|
|
"python scripts/qwen_tts_server.py\n"
|
|
"```\n\n"
|
|
"The external launchd plist is intentionally not installed automatically; review the "
|
|
"context/ directory and adapt paths and permissions first.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(root / "PORTABILITY-PATCH.txt").write_text(
|
|
"The following restored-copy portability adjustments were applied:\n"
|
|
+ "".join(f"- {item}\n" for item in sorted(set(changes)))
|
|
+ "\nThe original .env was preserved as .env.archive-original.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def write_checksum_manifest(root: Path) -> Path:
|
|
output = root / "SHA256SUMS"
|
|
lines: list[str] = []
|
|
for path in sorted(root.rglob("*")):
|
|
if not path.is_file() or path == output:
|
|
continue
|
|
relative = path.relative_to(root).as_posix()
|
|
lines.append(f"{sha256(path)} {relative}\n")
|
|
output.write_text("".join(lines), encoding="utf-8")
|
|
return output
|
|
|
|
|
|
def validate_tree(root: Path, old_root: str | None = None) -> list[str]:
|
|
errors: list[str] = []
|
|
required = (root / "pyproject.toml", root / "umbraaudio", root / "Qwen3-TTS")
|
|
for path in required:
|
|
if not path.exists():
|
|
errors.append(f"missing required path: {path}")
|
|
model = root / "models" / "Qwen3-TTS-12Hz-1.7B-Base" / "model.safetensors"
|
|
if not model.exists():
|
|
errors.append(f"missing model: {model}")
|
|
for database in sorted(root.rglob("*.db")):
|
|
if not database.is_file():
|
|
continue
|
|
if database.stat().st_size == 0:
|
|
continue
|
|
if not is_sqlite_file(database):
|
|
if any(
|
|
part in {".mypy_cache", ".pytest_cache", ".ruff_cache"} for part in database.parts
|
|
):
|
|
continue
|
|
errors.append(f"non-SQLite .db file: {database}")
|
|
continue
|
|
try:
|
|
connection = sqlite3.connect(f"file:{database}?mode=ro", uri=True)
|
|
result = connection.execute("PRAGMA quick_check").fetchone()[0]
|
|
if old_root:
|
|
try:
|
|
assert_no_old_prefixes(connection, [old_root])
|
|
except RuntimeError as exc:
|
|
errors.append(str(exc))
|
|
connection.close()
|
|
if result != "ok":
|
|
errors.append(f"SQLite quick_check failed: {database}: {result}")
|
|
except Exception as exc: # noqa: BLE001
|
|
errors.append(f"SQLite check error: {database}: {exc}")
|
|
if old_root:
|
|
code_files = [
|
|
root / "umbraaudio" / "config.py",
|
|
root / "umbraaudio" / "library.py",
|
|
root / "umbraaudio" / "storage.py",
|
|
root / "scripts" / "nas_lifecycle.py",
|
|
root / "scripts" / "verify_nas.py",
|
|
root / "scripts" / "extract_embeddings.py",
|
|
root / "scripts" / "qwen_tts_server.py",
|
|
root / "tts_server" / "qwen_tts_server.py",
|
|
root / "tts_server" / "sync_voice_data.sh",
|
|
root / "start_server.sh",
|
|
]
|
|
for code_file in code_files:
|
|
if code_file.exists() and old_root in code_file.read_text(encoding="utf-8"):
|
|
errors.append(f"old project root remains in {code_file}")
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
snapshot = sub.add_parser("snapshot-global")
|
|
snapshot.add_argument("--source", type=Path, required=True)
|
|
snapshot.add_argument("--output", type=Path, required=True)
|
|
snapshot.add_argument("--original-root", type=Path, required=True)
|
|
|
|
restore = sub.add_parser("restore-global")
|
|
restore.add_argument("--backup-dir", type=Path, required=True)
|
|
restore.add_argument("--target", type=Path, required=True)
|
|
restore.add_argument("--force", action="store_true")
|
|
restore.add_argument("--old-project-root", default="")
|
|
restore.add_argument("--new-project-root", default="")
|
|
|
|
patch = sub.add_parser("patch-paths")
|
|
patch.add_argument("--root", type=Path, required=True)
|
|
patch.add_argument("--old-root", required=True)
|
|
patch.add_argument("--new-root", type=Path, required=True)
|
|
patch.add_argument("--global-state-dir", type=Path)
|
|
|
|
manifest = sub.add_parser("write-manifest")
|
|
manifest.add_argument("--root", type=Path, required=True)
|
|
|
|
validate = sub.add_parser("validate-tree")
|
|
validate.add_argument("--root", type=Path, required=True)
|
|
validate.add_argument("--old-root", default="")
|
|
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.command == "snapshot-global":
|
|
snapshot_global(args.source, args.output, args.original_root)
|
|
elif args.command == "restore-global":
|
|
restore_global(
|
|
args.backup_dir,
|
|
args.target,
|
|
args.force,
|
|
args.old_project_root,
|
|
args.new_project_root,
|
|
)
|
|
elif args.command == "patch-paths":
|
|
patch_restored_paths(
|
|
args.root,
|
|
args.old_root,
|
|
args.new_root,
|
|
args.global_state_dir,
|
|
)
|
|
elif args.command == "write-manifest":
|
|
write_checksum_manifest(args.root)
|
|
elif args.command == "validate-tree":
|
|
errors = validate_tree(args.root, args.old_root or None)
|
|
if errors:
|
|
for error in errors:
|
|
print(error, file=sys.stderr)
|
|
return 1
|
|
print(f"Tree validation passed: {args.root}")
|
|
return 0
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|