83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Local-first storage with transparent NAS fallback.
|
|
|
|
Primary store is the local SSD; the NAS mirror is secondary. When a file
|
|
has been pruned locally, resolve_path/fetch restores it from the mirror
|
|
before handing it back to callers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_LOCAL_ROOT = Path("/Volumes/1TB_APL_Storage/UmbraAudio")
|
|
DEFAULT_NAS_ROOT = Path("/Volumes/Umbra/UmbraAudio")
|
|
|
|
|
|
class NasUnavailableError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def nas_mirror_path(path: str | Path, local_root: Path = DEFAULT_LOCAL_ROOT,
|
|
nas_root: Path = DEFAULT_NAS_ROOT) -> Path:
|
|
p = Path(path)
|
|
rel = p.relative_to(local_root) if p.is_absolute() else p
|
|
return nas_root / rel
|
|
|
|
|
|
def _restore_from_nas(mirror: Path, dest: Path) -> None:
|
|
if not mirror.exists():
|
|
raise FileNotFoundError(f"not on local SSD and no NAS copy: {dest}")
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(mirror, dest)
|
|
logger.info("restored %s from NAS mirror", dest)
|
|
|
|
|
|
def resolve_path(path: str | Path, *, local_root: Path = DEFAULT_LOCAL_ROOT,
|
|
nas_root: Path = DEFAULT_NAS_ROOT, restore: bool = True) -> Path:
|
|
"""Return a usable local path for `path`, pulling from NAS when pruned."""
|
|
p = Path(path)
|
|
if not p.is_absolute():
|
|
p = local_root / p
|
|
if p.exists() or not restore:
|
|
return p
|
|
mirror = nas_mirror_path(p, local_root, nas_root)
|
|
if not mirror.exists():
|
|
raise FileNotFoundError(f"not on local SSD and no NAS copy: {p}")
|
|
try:
|
|
_restore_from_nas(mirror, p)
|
|
except OSError as e:
|
|
raise NasUnavailableError(f"NAS not reachable while restoring {p}: {e}") from e
|
|
return p
|
|
|
|
|
|
def fetch(*paths: str | Path, **kw) -> list[Path]:
|
|
return [resolve_path(p, **kw) for p in paths]
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Fetch files from local SSD or NAS fallback")
|
|
parser.add_argument("paths", nargs="+", help="absolute or UmbraAudio-relative paths")
|
|
args = parser.parse_args(argv)
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
|
|
try:
|
|
for resolved in fetch(*args.paths):
|
|
print(resolved)
|
|
except (FileNotFoundError, NasUnavailableError) as e:
|
|
print(f"error: {e}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|