411 lines
15 KiB
Python
411 lines
15 KiB
Python
"""Migrate per-thread media.json from legacy wire shape to canonical v2."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urlparse
|
|
|
|
try:
|
|
from .frozen import frozen_match
|
|
from .media_manifest import (
|
|
SCHEMA_VERSION,
|
|
_from_wire_dict,
|
|
_item_to_wire,
|
|
atomic_write_json,
|
|
find_location,
|
|
hash_file,
|
|
location_of_kind,
|
|
new_derived_item,
|
|
new_fallback_location,
|
|
new_original_item,
|
|
selected_url,
|
|
validate_manifest,
|
|
)
|
|
from .media_refs import apply_rewrite_plan, atomic_write_text, plan_thread_rewrites
|
|
from .models import LegacyMediaJson, LegacyThreadData, MediaItem, MediaManifest
|
|
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 frozen import frozen_match
|
|
from media_manifest import (
|
|
SCHEMA_VERSION,
|
|
_from_wire_dict,
|
|
_item_to_wire,
|
|
atomic_write_json,
|
|
find_location,
|
|
hash_file,
|
|
location_of_kind,
|
|
new_derived_item,
|
|
new_fallback_location,
|
|
new_original_item,
|
|
selected_url,
|
|
validate_manifest,
|
|
)
|
|
from media_refs import apply_rewrite_plan, atomic_write_text, plan_thread_rewrites
|
|
from models import LegacyMediaJson, LegacyThreadData, MediaItem, MediaManifest
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MigrationResult:
|
|
"""Outcome of a per-thread legacy-to-canonical media.json migration.
|
|
|
|
``state`` is one of ``current``, ``migratable``, ``blocked``. ``changed``
|
|
reports whether the apply pass wrote files. ``issues`` collects any
|
|
blocking findings; ``item_count`` echoes the number of legacy items
|
|
inspected.
|
|
"""
|
|
state: str
|
|
changed: bool
|
|
issues: tuple[str, ...]
|
|
item_count: int
|
|
|
|
|
|
def media_id_from_url(url: str, index: int) -> str:
|
|
segment = Path(unquote(urlparse(url).path)).name
|
|
return segment if segment else f"m{index}"
|
|
|
|
|
|
def source_urls(thread_data: LegacyThreadData) -> dict[tuple[str, str], str]:
|
|
"""Map ``(post_id, media_id)`` to the original provider URL for each media reference.
|
|
|
|
Reads the post-level ``media_urls`` list from the typed ``LegacyThreadData``
|
|
and indexes each URL by a media id derived from its filename segment.
|
|
"""
|
|
result: dict[tuple[str, str], str] = {}
|
|
for post in thread_data.posts:
|
|
post_id = post.post_id
|
|
for index, url in enumerate(post.media_urls, start=1):
|
|
key = (post_id, media_id_from_url(url, index))
|
|
if key in result and result[key] != url:
|
|
raise ValueError(f"duplicate source mapping for {key!r}")
|
|
result[key] = url
|
|
return result
|
|
|
|
|
|
def reference_maps(
|
|
legacy: LegacyMediaJson, canonical: MediaManifest
|
|
) -> tuple[dict[str, str], dict[str, list[str]]]:
|
|
"""Build ``(filename → origin URL)`` and ``(legacy URL → origin URLs)`` maps for rewrite planning.
|
|
|
|
The filename map rewrites note references that still point at the local
|
|
file basename. The fallback URL map rewrites notes that still cite the
|
|
pre-migration fallback host.
|
|
"""
|
|
canonical_by_key: dict[tuple[str, str, str], MediaItem] = {}
|
|
for item in canonical.items:
|
|
key = (item.post_id, item.media_id, item.kind or item.role or "")
|
|
canonical_by_key[key] = item
|
|
filename_origins: dict[str, str] = {}
|
|
fallback_origins: dict[str, list[str]] = {}
|
|
for old in legacy.items:
|
|
if (old.role or old.kind or "") != "orig":
|
|
continue
|
|
key = (old.post_id, old.media_id, "orig")
|
|
item = canonical_by_key[key]
|
|
origin = next(location for location in item.locations if location.kind == "origin")
|
|
filename_origins[old.filename or ""] = origin.url or ""
|
|
old_url = old.url or ""
|
|
if old_url.startswith("https://"):
|
|
fallback_origins.setdefault(old_url, []).append(origin.url or "")
|
|
return filename_origins, fallback_origins
|
|
|
|
|
|
def corpus_inventory(
|
|
assets_root: Path,
|
|
note_root: Path,
|
|
frozen_ids: set[str],
|
|
*,
|
|
now: str,
|
|
) -> dict:
|
|
"""Walk the corpus and produce a typed-shape inventory ``dict`` (output of this function).
|
|
|
|
The inventory is a different document than ``media.json``; it is emitted
|
|
to disk as JSON, not consumed as ``LegacyMediaJson``. Internally, every
|
|
media.json read is parsed into ``LegacyMediaJson`` at the boundary.
|
|
"""
|
|
media_paths = sorted(assets_root.rglob("media.json"))
|
|
counts: dict[str, int] = {
|
|
"input_items": 0,
|
|
"migratable": 0,
|
|
"frozen_skipped": 0,
|
|
"missing_origin_mapping": 0,
|
|
"missing_local_file": 0,
|
|
"ambiguous_note_reference": 0,
|
|
"invalid_legacy_row": 0,
|
|
}
|
|
threads: list[dict] = []
|
|
frozen_hashes: dict[str, str] = {}
|
|
for media_path in media_paths:
|
|
asset_dir = media_path.parent
|
|
relative = asset_dir.relative_to(assets_root).as_posix()
|
|
match = frozen_match(asset_dir, frozen_ids)
|
|
legacy = LegacyMediaJson.from_dict(json.loads(media_path.read_text(encoding="utf-8")))
|
|
is_legacy = legacy.schema_version != SCHEMA_VERSION
|
|
legacy_items = list(legacy.items)
|
|
unique_legacy: set[tuple[str, str, str]] = {
|
|
(item.post_id, item.media_id, item.role or item.kind or "")
|
|
for item in legacy_items
|
|
}
|
|
counts["input_items"] += len(unique_legacy)
|
|
for path in sorted(asset_dir.iterdir()):
|
|
if path.is_file() and path.name != "media.json":
|
|
key = f"{relative}/{path.name}"
|
|
frozen_hashes[key] = hash_file(path)
|
|
for archive in sorted((note_root / relative).glob("*.md")) if (note_root / relative).is_dir() else []:
|
|
key = f"{relative}/{archive.name}"
|
|
frozen_hashes[key] = hash_file(archive)
|
|
if match is not None:
|
|
counts["frozen_skipped"] += len(unique_legacy)
|
|
threads.append({
|
|
"relative": relative,
|
|
"state": "frozen_skipped",
|
|
"frozen_id": match,
|
|
"item_count": len(unique_legacy),
|
|
})
|
|
continue
|
|
thread_data_path = asset_dir / "thread_data.json"
|
|
if not thread_data_path.is_file():
|
|
counts["missing_origin_mapping"] += len(unique_legacy)
|
|
threads.append({
|
|
"relative": relative,
|
|
"state": "missing_origin_mapping",
|
|
"item_count": len(unique_legacy),
|
|
"issues": ["missing thread_data.json"],
|
|
})
|
|
continue
|
|
thread_data = LegacyThreadData.from_dict(
|
|
json.loads(thread_data_path.read_text(encoding="utf-8"))
|
|
)
|
|
try:
|
|
canonical_manifest, issues = canonical_from_legacy(legacy, thread_data, asset_dir, now)
|
|
except Exception as exc:
|
|
counts["invalid_legacy_row"] += len(unique_legacy)
|
|
threads.append({
|
|
"relative": relative,
|
|
"state": "invalid_legacy_row",
|
|
"item_count": len(unique_legacy),
|
|
"issues": [str(exc)],
|
|
})
|
|
continue
|
|
if issues:
|
|
counts["invalid_legacy_row"] += len(unique_legacy)
|
|
threads.append({
|
|
"relative": relative,
|
|
"state": "invalid_legacy_row",
|
|
"item_count": len(unique_legacy),
|
|
"issues": list(issues),
|
|
})
|
|
continue
|
|
counts["migratable"] += len(canonical_manifest.items)
|
|
thread_entry: dict = {
|
|
"relative": relative,
|
|
"state": "migratable" if is_legacy else "current",
|
|
"item_count": len(canonical_manifest.items),
|
|
}
|
|
threads.append(thread_entry)
|
|
expected_total = (
|
|
counts["migratable"]
|
|
+ counts["frozen_skipped"]
|
|
+ counts["missing_origin_mapping"]
|
|
+ counts["missing_local_file"]
|
|
+ counts["ambiguous_note_reference"]
|
|
+ counts["invalid_legacy_row"]
|
|
)
|
|
if counts["input_items"] != expected_total:
|
|
raise SystemExit("corpus accounting mismatch")
|
|
return {
|
|
"schema_version": 1,
|
|
"generated_at": now,
|
|
"counts": counts,
|
|
"threads": threads,
|
|
"frozen_hashes": frozen_hashes,
|
|
}
|
|
|
|
|
|
def canonical_from_legacy(
|
|
legacy: LegacyMediaJson,
|
|
thread_data: LegacyThreadData,
|
|
asset_dir: Path,
|
|
now: str,
|
|
) -> tuple[MediaManifest, tuple[str, ...]]:
|
|
"""Build a canonical ``MediaManifest`` from a legacy ``LegacyMediaJson`` + ``LegacyThreadData``.
|
|
|
|
Each legacy item produces a single canonical ``MediaItem``. Legacy
|
|
``orig`` items gain a fallback ``MediaLocation`` if they already cited
|
|
the pre-migration fallback host on the wire.
|
|
"""
|
|
sources = source_urls(thread_data)
|
|
items: list[MediaItem] = []
|
|
issues: list[str] = []
|
|
seen: set[tuple[str, str, str]] = set()
|
|
for old in legacy.items:
|
|
role = old.role or old.kind or ""
|
|
post_id = old.post_id
|
|
media_id = old.media_id
|
|
filename = old.filename or ""
|
|
handle = old.handle or ""
|
|
old_url = old.url or ""
|
|
key = (post_id, media_id, role)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
if role == "orig":
|
|
origin_url = sources.get((post_id, media_id))
|
|
if origin_url is None:
|
|
issues.append(f"missing origin mapping: {post_id}/{media_id}")
|
|
continue
|
|
item = new_original_item(
|
|
post_id=post_id,
|
|
media_id=media_id,
|
|
handle=handle,
|
|
origin_url=origin_url,
|
|
filename=filename,
|
|
local_path=asset_dir / filename,
|
|
now=now,
|
|
)
|
|
if old_url.startswith("https://"):
|
|
local = find_location(item, "local")
|
|
sha = local.sha256 if local is not None else None
|
|
item = MediaItem(
|
|
post_id=item.post_id,
|
|
media_id=item.media_id,
|
|
kind=item.kind,
|
|
role=item.role,
|
|
handle=item.handle,
|
|
embed=item.embed,
|
|
locations=item.locations
|
|
+ (
|
|
new_fallback_location(
|
|
provider="catbox",
|
|
url=old_url,
|
|
content_hash=sha or None,
|
|
recorded_at=now,
|
|
uploaded_at=None,
|
|
),
|
|
),
|
|
publication=item.publication,
|
|
derived_from=item.derived_from,
|
|
caption=item.caption,
|
|
)
|
|
items.append(item)
|
|
elif role in {"crt", "crt_outline", "denoise", "ocr"}:
|
|
item = new_derived_item(
|
|
post_id=post_id,
|
|
media_id=media_id,
|
|
handle=handle,
|
|
role=role,
|
|
filename=filename,
|
|
asset_dir=asset_dir,
|
|
now=now,
|
|
)
|
|
if old_url.startswith("https://"):
|
|
local = find_location(item, "local")
|
|
sha = local.sha256 if local is not None else None
|
|
item = MediaItem(
|
|
post_id=item.post_id,
|
|
media_id=item.media_id,
|
|
kind=item.kind,
|
|
role=item.role,
|
|
handle=item.handle,
|
|
embed=item.embed,
|
|
locations=item.locations
|
|
+ (
|
|
new_fallback_location(
|
|
provider="catbox",
|
|
url=old_url,
|
|
content_hash=sha or None,
|
|
recorded_at=now,
|
|
uploaded_at=None,
|
|
),
|
|
),
|
|
publication=item.publication,
|
|
derived_from=item.derived_from,
|
|
caption=item.caption,
|
|
)
|
|
items.append(item)
|
|
else:
|
|
issues.append(f"invalid legacy role: {post_id}/{media_id}/{role}")
|
|
manifest = MediaManifest(
|
|
schema_version=SCHEMA_VERSION,
|
|
root_post_id=legacy.root_post_id,
|
|
items=tuple(items),
|
|
)
|
|
issues.extend(validate_manifest(manifest))
|
|
return manifest, tuple(issues)
|
|
|
|
|
|
def migrate_legacy_thread(
|
|
asset_dir: Path,
|
|
note_dir: Path,
|
|
*,
|
|
now: str,
|
|
apply: bool,
|
|
) -> MigrationResult:
|
|
"""Migrate a per-thread ``media.json`` to the canonical v2 shape in place.
|
|
|
|
Returns a ``MigrationResult`` describing the outcome (``current``,
|
|
``blocked``, ``migratable``). When ``apply`` is False, no files are
|
|
written; when True, the canonical manifest is atomically written and
|
|
the corresponding note references are rewritten with rollback on
|
|
failure.
|
|
"""
|
|
media_path = asset_dir / "media.json"
|
|
legacy = LegacyMediaJson.from_dict(json.loads(media_path.read_text(encoding="utf-8")))
|
|
if legacy.schema_version == SCHEMA_VERSION:
|
|
return MigrationResult("current", False, (), len(legacy.items))
|
|
thread_data = LegacyThreadData.from_dict(
|
|
json.loads((asset_dir / "thread_data.json").read_text(encoding="utf-8"))
|
|
)
|
|
canonical, issues = canonical_from_legacy(legacy, thread_data, asset_dir, now)
|
|
if issues:
|
|
return MigrationResult("blocked", False, issues, len(legacy.items))
|
|
if apply:
|
|
filename_origins, fallback_origins = reference_maps(legacy, canonical)
|
|
plan = plan_thread_rewrites(
|
|
note_dir,
|
|
filename_origins=filename_origins,
|
|
fallback_origins=fallback_origins,
|
|
)
|
|
if plan.issues:
|
|
return MigrationResult("blocked", False, plan.issues, len(legacy.items))
|
|
note_backup: dict[Path, str] = {}
|
|
try:
|
|
for rewrite in plan.files:
|
|
note_backup[rewrite.path] = rewrite.before
|
|
atomic_write_json(
|
|
media_path,
|
|
_manifest_to_wire_shape(canonical, legacy),
|
|
)
|
|
apply_rewrite_plan(plan)
|
|
except Exception:
|
|
for path, before in note_backup.items():
|
|
atomic_write_text(path, before)
|
|
raise
|
|
return MigrationResult("migratable", apply, (), len(legacy.items))
|
|
|
|
|
|
def _manifest_to_wire_shape(
|
|
manifest: MediaManifest, legacy: LegacyMediaJson
|
|
) -> dict:
|
|
"""Convert a typed manifest back to a wire-shape dict for atomic_write_json.
|
|
|
|
Carries forward legacy mirrors (typed MediaManifest doesn't model
|
|
mirrors). Mirrors are written as a list to preserve wire compatibility
|
|
with downstream producers (the typed view keys them by destination id).
|
|
"""
|
|
mirrors_out: list = []
|
|
if legacy.mirrors is not None:
|
|
mirrors_out = list(copy.deepcopy(legacy.mirrors).values())
|
|
out: dict = {
|
|
"schema_version": manifest.schema_version,
|
|
"root_post_id": manifest.root_post_id,
|
|
"items": [_item_to_wire(item) for item in manifest.items],
|
|
"mirrors": mirrors_out,
|
|
}
|
|
if manifest.captured_at is not None:
|
|
out["captured_at"] = manifest.captured_at
|
|
return out
|