217 lines
7.2 KiB
Python
217 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
try:
|
|
from .media_manifest import (
|
|
atomic_write_json,
|
|
find_location,
|
|
hash_file,
|
|
new_fallback_location,
|
|
selected_url,
|
|
)
|
|
from .media_refs import atomic_write_text, remote_markup
|
|
except ImportError: # pragma: no cover - script-mode import
|
|
if str(Path(__file__).resolve().parent) not in sys.path:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from media_manifest import (
|
|
atomic_write_json,
|
|
find_location,
|
|
hash_file,
|
|
new_fallback_location,
|
|
selected_url,
|
|
)
|
|
from media_refs import atomic_write_text, remote_markup
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FallbackResult:
|
|
url: str
|
|
reused: bool
|
|
|
|
|
|
def replace_selected_url(note_dir: Path, old_url: str, new_url: str) -> None:
|
|
if old_url == new_url:
|
|
return
|
|
old_markup = remote_markup(old_url)
|
|
new_markup = remote_markup(new_url)
|
|
hits: list[Path] = []
|
|
for path in note_dir.glob("*.md"):
|
|
if old_markup in path.read_text(encoding="utf-8"):
|
|
hits.append(path)
|
|
total = sum(path.read_text(encoding="utf-8").count(old_markup) for path in hits)
|
|
if total != 1:
|
|
if total == 0 and sum(
|
|
path.read_text(encoding="utf-8").count(new_markup)
|
|
for path in note_dir.glob("*.md")
|
|
) == 1:
|
|
return
|
|
raise ValueError("selected origin reference not found exactly once")
|
|
path = hits[0]
|
|
text = path.read_text(encoding="utf-8")
|
|
atomic_write_text(path, text.replace(old_markup, new_markup, 1))
|
|
|
|
|
|
def _group(manifest: dict[str, Any], media_id: str) -> list[dict[str, Any]]:
|
|
return [
|
|
item
|
|
for item in manifest.get("items") or []
|
|
if str(item.get("media_id") or "") == media_id
|
|
]
|
|
|
|
|
|
def _one(items: list[dict[str, Any]], role: str) -> dict[str, Any]:
|
|
matches = [item for item in items if item.get("role") == role]
|
|
if len(matches) != 1:
|
|
raise ValueError(f"expected one role={role}, found {len(matches)}")
|
|
return matches[0]
|
|
|
|
|
|
def _verified_local(item: dict[str, Any], asset_dir: Path) -> tuple[Path, str]:
|
|
local = find_location(item, "local")
|
|
if local is None or local.get("integrity") != "present":
|
|
raise ValueError("selected item has no present local location")
|
|
path = asset_dir / str(local.get("path") or "")
|
|
expected = str(local.get("sha256") or "")
|
|
if not path.is_file() or not expected or hash_file(path) != expected:
|
|
raise ValueError("selected local file failed hash verification")
|
|
return path, expected
|
|
|
|
|
|
def activate_fallback(
|
|
asset_dir: Path,
|
|
note_dir: Path,
|
|
*,
|
|
media_id: str,
|
|
role: str,
|
|
provider: str,
|
|
confirm_origin_unavailable: bool,
|
|
now: str,
|
|
upload: Callable[[Path], str],
|
|
lookup: Callable[[str, str], str | None] | None = None,
|
|
) -> FallbackResult:
|
|
if not confirm_origin_unavailable:
|
|
raise ValueError("origin-unavailable confirmation is required")
|
|
media_path = asset_dir / "media.json"
|
|
manifest = json.loads(media_path.read_text(encoding="utf-8"))
|
|
group = _group(manifest, media_id)
|
|
original = _one(group, "orig")
|
|
target = _one(group, role)
|
|
origin = find_location(original, "origin:x")
|
|
if origin is None:
|
|
raise ValueError("original item has no X origin location")
|
|
old_url = selected_url(next((item for item in group if item.get("embed")), original))
|
|
if old_url is None:
|
|
raise ValueError("media group has no selected HTTPS location")
|
|
local_path, content_hash = _verified_local(target, asset_dir)
|
|
origin["availability"] = "unavailable"
|
|
origin["confirmed_unavailable_at"] = now
|
|
|
|
fallback = next(
|
|
(
|
|
location
|
|
for location in target.get("locations") or []
|
|
if location.get("kind") == "fallback"
|
|
and location.get("provider") == provider
|
|
and location.get("sha256") == content_hash
|
|
),
|
|
None,
|
|
)
|
|
reused = fallback is not None
|
|
if fallback is None and lookup is not None:
|
|
reused_url = lookup(provider, content_hash)
|
|
if reused_url is not None:
|
|
fallback = new_fallback_location(
|
|
provider=provider,
|
|
url=reused_url,
|
|
content_hash=content_hash,
|
|
recorded_at=now,
|
|
uploaded_at=None,
|
|
)
|
|
target.setdefault("locations", []).append(fallback)
|
|
reused = True
|
|
if fallback is None:
|
|
url = upload(local_path)
|
|
fallback = new_fallback_location(
|
|
provider=provider,
|
|
url=url,
|
|
content_hash=content_hash,
|
|
recorded_at=now,
|
|
uploaded_at=now,
|
|
)
|
|
target.setdefault("locations", []).append(fallback)
|
|
|
|
atomic_write_json(media_path, manifest)
|
|
replace_selected_url(note_dir, old_url, str(fallback["url"]))
|
|
for item in group:
|
|
item["embed"] = item is target
|
|
target["publication"] = {
|
|
"selected_location_id": fallback["id"],
|
|
"selected_at": now,
|
|
"reason": "origin-unavailable",
|
|
}
|
|
atomic_write_json(media_path, manifest)
|
|
return FallbackResult(str(fallback["url"]), reused)
|
|
|
|
|
|
def find_existing_fallback(
|
|
assets_root: Path,
|
|
provider: str,
|
|
content_hash: str,
|
|
) -> str | None:
|
|
urls: set[str] = set()
|
|
for media_path in assets_root.rglob("media.json"):
|
|
data = json.loads(media_path.read_text(encoding="utf-8"))
|
|
if data.get("schema_version") != 2:
|
|
continue
|
|
for item in data.get("items") or []:
|
|
for location in item.get("locations") or []:
|
|
if (
|
|
location.get("kind") == "fallback"
|
|
and location.get("provider") == provider
|
|
and location.get("sha256") == content_hash
|
|
and str(location.get("url") or "").startswith("https://")
|
|
):
|
|
urls.add(str(location["url"]))
|
|
return sorted(urls)[0] if urls else None
|
|
|
|
|
|
def restore_origin(
|
|
asset_dir: Path,
|
|
note_dir: Path,
|
|
*,
|
|
media_id: str,
|
|
now: str,
|
|
) -> None:
|
|
media_path = asset_dir / "media.json"
|
|
manifest = json.loads(media_path.read_text(encoding="utf-8"))
|
|
group = _group(manifest, media_id)
|
|
original = _one(group, "orig")
|
|
origin = find_location(original, "origin:x")
|
|
if origin is None:
|
|
raise ValueError("original item has no X origin location")
|
|
visible = next((item for item in group if item.get("embed")), original)
|
|
old_url = selected_url(visible)
|
|
new_url = str(origin.get("url") or "")
|
|
if old_url is None or not new_url.startswith("https://"):
|
|
raise ValueError("media group has no restorable origin")
|
|
replace_selected_url(note_dir, old_url, new_url)
|
|
origin["availability"] = "available"
|
|
origin["checked_at"] = now
|
|
origin["check"] = {
|
|
"status": None,
|
|
"result": "available",
|
|
"detail": "manual restore-origin selection",
|
|
}
|
|
for item in group:
|
|
item["embed"] = item is original
|
|
original["publication"] = {
|
|
"selected_location_id": "origin:x",
|
|
"selected_at": now,
|
|
"reason": "manual",
|
|
}
|
|
atomic_write_json(media_path, manifest) |