60 lines
2.6 KiB
Python
60 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
def _safe_path(root: Path, relative: str) -> Path | None:
|
|
target = (root / relative).resolve()
|
|
if not target.is_relative_to(root.resolve()):
|
|
return None
|
|
return target
|
|
|
|
|
|
def bootstrap_media_store(source_root: Path, target_root: Path) -> dict[str, int]:
|
|
"""Merge the immutable image baseline into the persistent media volume.
|
|
|
|
Existing decisions and files in the volume always win. New baseline assets
|
|
are copied in on release, so a recreated container cannot reset decisions
|
|
while a newer Git baseline can still add assets.
|
|
"""
|
|
source_manifest_path = source_root / "manifest.json"
|
|
target_manifest_path = target_root / "manifest.json"
|
|
source = json.loads(source_manifest_path.read_text(encoding="utf-8"))
|
|
target = json.loads(target_manifest_path.read_text(encoding="utf-8")) if target_manifest_path.exists() else {"version": 1, "assets": []}
|
|
target_by_url = {item.get("asset_url"): item for item in target.get("assets", [])}
|
|
added = 0
|
|
copied = 0
|
|
target_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
for source_item in source.get("assets", []):
|
|
url = source_item.get("asset_url")
|
|
if not url or url in target_by_url:
|
|
continue
|
|
item = dict(source_item)
|
|
target_by_url[url] = item
|
|
target.setdefault("assets", []).append(item)
|
|
added += 1
|
|
|
|
for item in target.get("assets", []):
|
|
for relative in [item.get("local_path"), *[variant.get("local_path") for variant in item.get("derivatives", [])]]:
|
|
if not isinstance(relative, str):
|
|
continue
|
|
source_path = _safe_path(source_root, relative)
|
|
target_path = _safe_path(target_root, relative)
|
|
if not source_path or not target_path or not source_path.is_file() or target_path.exists():
|
|
continue
|
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(source_path, target_path)
|
|
copied += 1
|
|
|
|
target["version"] = max(int(source.get("version", 1)), int(target.get("version", 1)))
|
|
if added or copied or not target_manifest_path.exists():
|
|
target["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
temporary = target_manifest_path.with_suffix(".json.tmp")
|
|
temporary.write_text(json.dumps(target, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
temporary.replace(target_manifest_path)
|
|
return {"added_assets": added, "copied_files": copied}
|