diff --git a/publish/quartz-patches/apply.py b/publish/quartz-patches/apply.py index 89678a5..0390c68 100644 --- a/publish/quartz-patches/apply.py +++ b/publish/quartz-patches/apply.py @@ -22,12 +22,12 @@ import re import sys from pathlib import Path -SCRIPT = Path(__file__).resolve() -VAULT = SCRIPT.parents[2] -QUARTZ_REPO = VAULT / "site" / "quartz" -SPA_FILE = QUARTZ_REPO / "components" / "scripts" / "spa.inline.ts" -PUBLIC = VAULT / "site" / "public" -OVERLAY = VAULT / "publish" / "quartz.config.yaml" +SCRIPT: Path = Path(__file__).resolve() +VAULT: Path = SCRIPT.parents[2] +QUARTZ_REPO: Path = VAULT / "site" / "quartz" +SPA_FILE: Path = QUARTZ_REPO / "components" / "scripts" / "spa.inline.ts" +PUBLIC: Path = VAULT / "site" / "public" +OVERLAY: Path = VAULT / "publish" / "quartz.config.yaml" def read_base_url() -> tuple[str, str]: @@ -63,9 +63,9 @@ def read_base_url() -> tuple[str, str]: ABS_BASE, REL_BASE = read_base_url() -HEAD_RE = re.compile(r"(]*>)", re.IGNORECASE) +HEAD_RE: re.Pattern[str] = re.compile(r"(]*>)", re.IGNORECASE) BASE_TAG = f'\n' -TITLE_LINK = re.compile( +TITLE_LINK: re.Pattern[str] = re.compile( r'(]*class="page-title">)' ) diff --git a/scripts/twitter/__init__.py b/scripts/twitter/__init__.py index cbc350e..2c152db 100644 --- a/scripts/twitter/__init__.py +++ b/scripts/twitter/__init__.py @@ -1 +1,2 @@ -"""Emit Threadwell archive notes from a thread_data.json dump. No fetch, no lift.""" +"""twitter scripts package for Threadwell archive ops.""" +from __future__ import annotations diff --git a/scripts/twitter/audit_handle_index.py b/scripts/twitter/audit_handle_index.py index 84b888b..b6ce3b4 100644 --- a/scripts/twitter/audit_handle_index.py +++ b/scripts/twitter/audit_handle_index.py @@ -28,10 +28,10 @@ import re import sys from pathlib import Path -HERE = Path(__file__).resolve().parent -DEFAULT_VAULT = HERE.parent.parent +HERE: Path = Path(__file__).resolve().parent +DEFAULT_VAULT: Path = HERE.parent.parent -_WIKILINK_ITEM = re.compile(r"^- \[\[([^\]|]+)\]\]\s*$") +_WIKILINK_ITEM: re.Pattern[str] = re.compile(r"^- \[\[([^\]|]+)\]\]\s*$") def _split_frontmatter(text: str) -> tuple[str, str]: @@ -62,7 +62,7 @@ def _index_links(idx_path: Path) -> set[str]: return out -def audit_one(handle_dir: Path) -> dict[str, object]: +def audit_one(handle_dir: Path) -> dict: idx_path = handle_dir / "index.md" real = {d.name for d in handle_dir.iterdir() if d.is_dir()} links = _index_links(idx_path) @@ -78,7 +78,7 @@ def audit_one(handle_dir: Path) -> dict[str, object]: } -def audit_threads_index(threads_root: Path) -> dict[str, object]: +def audit_threads_index(threads_root: Path) -> dict: """Audit the top-level archive/threads/index.md against real handle dirs.""" idx_path = threads_root / "index.md" @@ -115,7 +115,7 @@ def main(argv: list[str] | None = None) -> int: if not threads_root.is_dir(): raise SystemExit(f"missing {threads_root}") - reports: list[dict[str, object]] = [] + reports: list[dict] = [] for entry in sorted(threads_root.iterdir()): if not entry.is_dir() or entry.name.startswith("."): continue diff --git a/scripts/twitter/backup_assets.py b/scripts/twitter/backup_assets.py index 2582bc1..2ee7ee4 100644 --- a/scripts/twitter/backup_assets.py +++ b/scripts/twitter/backup_assets.py @@ -1,3 +1,5 @@ +"""Back up thread assets to a configured destination. Verifies every file.""" + from __future__ import annotations import json @@ -9,30 +11,38 @@ from pathlib import Path try: from .media_manifest import ( + _from_wire_dict, + _manifest_to_wire, atomic_write_json, hash_file, inventory_digest, payload_inventory, ) + from .models import 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 media_manifest import ( + _from_wire_dict, + _manifest_to_wire, atomic_write_json, hash_file, inventory_digest, payload_inventory, ) + from models import MediaManifest @dataclass(frozen=True) class BackupResult: + """Outcome of a single-thread backup attempt to one destination.""" state: str destination: Path error: str | None def load_destination_root(vault: Path, destination_id: str) -> Path: + """Read the configured backup destination root from ``vault/secrets/credentials.toml``.""" path = vault / "secrets" / "credentials.toml" if not path.is_file(): raise RuntimeError(f"missing {path}") @@ -71,13 +81,16 @@ def backup_thread( now: str, require_destination_root: bool = False, ) -> BackupResult: + """Mirror ``asset_dir`` into ``destination_root/`` and verify every payload by sha256.""" media_path = asset_dir / "media.json" - manifest = json.loads(media_path.read_text(encoding="utf-8")) + media_path = asset_dir / "media.json" + raw = json.loads(media_path.read_text(encoding="utf-8")) + manifest = _from_wire_dict(raw) relative = asset_dir.relative_to(assets_root) destination = destination_root / relative inventory = payload_inventory(asset_dir, manifest) digest = inventory_digest(inventory) - mirror = { + mirror: dict = { "destination_id": destination_id, "relative_path": relative.as_posix(), "state": "error", @@ -89,7 +102,7 @@ def backup_thread( } others = [ entry - for entry in manifest.get("mirrors") or [] + for entry in raw.get("mirrors") or [] if entry.get("destination_id") != destination_id ] try: @@ -110,13 +123,28 @@ def backup_thread( mirror["verified_at"] = now except OSError as exc: mirror["error"] = str(exc)[:300] - manifest["mirrors"] = others + [mirror] - atomic_write_json(media_path, manifest) + raw["mirrors"] = others + [mirror] + media_path.write_text( + json.dumps(_manifest_to_wire_for_disk(manifest, raw), indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + newline="\n", + ) if mirror["state"] == "synced": shutil.copy2(media_path, destination / "media.json") if hash_file(media_path) != hash_file(destination / "media.json"): mirror["state"] = "error" mirror["error"] = "final manifest verification failed" - atomic_write_json(media_path, manifest) + media_path.write_text( + json.dumps(_manifest_to_wire_for_disk(manifest, raw), indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + newline="\n", + ) return BackupResult("error", destination, mirror["error"]) - return BackupResult(mirror["state"], destination, mirror["error"]) \ No newline at end of file + return BackupResult(mirror["state"], destination, mirror["error"]) + + +def _manifest_to_wire_for_disk(manifest: MediaManifest, raw: dict) -> dict: + """Merge typed manifest with the on-disk wire shape (incl. mirrors).""" + typed_wire = _manifest_to_wire(manifest) + typed_wire["mirrors"] = list(raw.get("mirrors") or []) + return typed_wire diff --git a/scripts/twitter/catbox_client.py b/scripts/twitter/catbox_client.py index d0f2e9f..18844c3 100644 --- a/scripts/twitter/catbox_client.py +++ b/scripts/twitter/catbox_client.py @@ -1,3 +1,4 @@ +"""Catbox userhash loader and uploader (curl.exe wrapper).""" from __future__ import annotations import subprocess @@ -6,6 +7,7 @@ from pathlib import Path def load_userhash(vault: Path) -> str: + """Load the catbox ``userhash`` from ``vault/secrets/credentials.toml`` (TOML or lenient line scan).""" path = vault / "secrets" / "credentials.toml" if not path.is_file(): raise RuntimeError(f"missing {path}") @@ -31,6 +33,7 @@ def load_userhash(vault: Path) -> str: def upload_file(path: Path, userhash: str) -> str: + """Upload ``path`` to catbox.moe via curl.exe and return the public HTTPS URL.""" cmd = [ "curl.exe", "-sS", diff --git a/scripts/twitter/crt_pass.py b/scripts/twitter/crt_pass.py index 0582650..2ca9f31 100644 --- a/scripts/twitter/crt_pass.py +++ b/scripts/twitter/crt_pass.py @@ -29,30 +29,30 @@ import subprocess import sys from pathlib import Path -LOTTES_SGP = Path( +LOTTES_SGP: Path = Path( r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64" r"\LOTTES_CODE_(LOTTES_MULTIPASS 2X) V2.sgp" ) -CYCLON_SGP = Path( +CYCLON_SGP: Path = Path( r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64" r"\LOTTES_CRT_CODE_(CYCLON 6X).sgp" ) -SLANG_LOTTES = Path( +SLANG_LOTTES: Path = Path( r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders" r"\shaders_slang\crt\crt-lottes-multipass.slangp" ) -SLANG_CYCLON = Path( +SLANG_CYCLON: Path = Path( r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders" r"\shaders_slang\crt\crt-Cyclon.slangp" ) -CLI_DEFAULT = Path(r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe") +CLI_DEFAULT: Path = Path(r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe") # Knobs copied from the .sgp files. Not a substitute for PixelSize. -LOTTES_PARAMS = ( +LOTTES_PARAMS: str = ( "hardScan=-19,hardPix=-7,warpX=0,warpY=0,maskDark=1.3,maskLight=1.3," "shadowMask=4,brightBoost=1.15,bloomAmount=0,shape=1.9" ) -CYCLON_PARAMS = ( +CYCLON_PARAMS: str = ( "SCANLINE=0.2,INTERLACE=0,M_TYPE=0,SLOT=0,SLOTW=2,Maskl=0.55,Maskh=1," "bzl=0,ambient=0,REFLECT=0,WARPX=0,WARPY=0,BR_DEP=0,BRIGHTNESS_=1.05," "BLACK=0.01,vig=0" @@ -60,6 +60,7 @@ CYCLON_PARAMS = ( def probe_size(path: Path) -> tuple[int, int]: + """Return ``(width, height)`` of an image or video via ``ffprobe``.""" raw = subprocess.check_output( [ "ffprobe", @@ -80,6 +81,7 @@ def probe_size(path: Path) -> tuple[int, int]: def pick_preset(width: int, height: int, name: str, auto_max: int) -> tuple[str, Path, Path, str]: + """Resolve the preset name to ``(name, .sgp_path, .slangp_path, params)``; ``auto`` picks cyclon under ``auto_max``.""" if name == "auto": name = "cyclon" if max(width, height) <= auto_max else "lottes" if name == "lottes": @@ -90,6 +92,7 @@ def pick_preset(width: int, height: int, name: str, auto_max: int) -> tuple[str, def find_cli(explicit: Path | None) -> Path: + """Locate ``librashader-cli``: explicit path, then PATH, then the scoop-installed default.""" if explicit is not None: if not explicit.is_file(): raise SystemExit(f"missing --cli {explicit}") @@ -107,6 +110,7 @@ def find_cli(explicit: Path | None) -> Path: def parse_crop(spec: str) -> str | None: + """Parse a ``--crop`` value into an ffmpeg ``crop=...`` filter, or ``None`` to skip cropping.""" if spec in ("none", "", "off"): return None if spec == "auto": @@ -119,6 +123,7 @@ def parse_crop(spec: str) -> str | None: def detect_crop(path: Path) -> str | None: + """Run ``ffmpeg cropdetect`` on ``path`` and return the auto-detected crop filter (or ``None``).""" proc = subprocess.run( [ "ffmpeg", @@ -144,6 +149,7 @@ def detect_crop(path: Path) -> str | None: def run_ffmpeg(src: Path, dest: Path, vf: str) -> None: + """Run ``ffmpeg -y`` with the given ``-vf`` filter to render a single still ``src`` -> ``dest``.""" dest.parent.mkdir(parents=True, exist_ok=True) subprocess.check_call( [ @@ -175,6 +181,8 @@ def run_librashader( dimensions: str, params: str, ) -> None: + """Render ``src`` through ``slang`` via the ``librashader-cli`` binary into ``dest``.""" + dest.parent.mkdir(parents=True, exist_ok=True) dest.parent.mkdir(parents=True, exist_ok=True) cmd = [ str(cli), diff --git a/scripts/twitter/emit_archive.py b/scripts/twitter/emit_archive.py index 6bc2e63..278cea5 100644 --- a/scripts/twitter/emit_archive.py +++ b/scripts/twitter/emit_archive.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse import json +import logging import re import shutil import sys @@ -12,10 +13,14 @@ from datetime import date from pathlib import Path from urllib.parse import parse_qs, unquote, urlparse +logger = logging.getLogger(__name__) + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) try: from twitter.media_manifest import ( + _from_wire_dict, + _item_to_wire, atomic_write_json, find_location, item_key, @@ -24,7 +29,7 @@ try: selected_url, ) from twitter.media_refs import remote_markup - from twitter.models import ThreadData, load_thread + from twitter.models import MediaItem, ThreadData, load_thread from twitter.render import render_branch, render_spine from twitter.slug import branch_file_name, date_prefix, thread_dir_name from twitter.tree import ( @@ -37,6 +42,8 @@ try: ) except ImportError: # pragma: no cover - script-mode import from media_manifest import ( + _from_wire_dict, + _item_to_wire, atomic_write_json, find_location, item_key, @@ -45,7 +52,7 @@ except ImportError: # pragma: no cover - script-mode import selected_url, ) from media_refs import remote_markup - from models import ThreadData, load_thread + from models import MediaItem, ThreadData, load_thread from render import render_branch, render_spine from slug import branch_file_name, date_prefix, thread_dir_name from tree import ( @@ -59,6 +66,7 @@ except ImportError: # pragma: no cover - script-mode import def _frontmatter_lines(text: str) -> tuple[list[str], str]: + """Return (frontmatter_lines, body_after_closer) for an archive note.""" if not text.startswith("---\n"): return [], text end = text.find("\n---", 4) @@ -68,6 +76,7 @@ def _frontmatter_lines(text: str) -> tuple[list[str], str]: def _tag_values(lines: list[str]) -> list[str]: + """Extract tag strings from the ``tags:`` block of a YAML frontmatter lines list.""" tags: list[str] = [] in_tags = False for line in lines: @@ -121,21 +130,22 @@ def preserve_review_state( return "---\n" + "\n".join(output) + "\n---" + fresh_body -def missing_local_media(items: list[dict[str, object]]) -> list[str]: +def missing_local_media(items: list[MediaItem]) -> list[str]: missing: list[str] = [] for item in items: local = find_location(item, "local") - if local is not None and local.get("integrity") == "missing": + if local is not None and local.integrity == "missing": post_id, media_id, role = item_key(item) missing.append(f"{post_id}/{media_id}/{role}") return missing -_POST_ID_LINE = re.compile(r'^post_id:\s*"?([^"\s]+)"?\s*$') -_WIKILINK_ITEM = re.compile(r"^- \[\[([^\]|]+)\]\]\s*$") +_POST_ID_LINE: re.Pattern[str] = re.compile(r'^post_id:\s*"?([^"\s]+)"?\s*$') +_WIKILINK_ITEM: re.Pattern[str] = re.compile(r"^- \[\[([^\]|]+)\]\]\s*$") @dataclass(frozen=True) class EmitResult: + """Outcome of one emit pass: which post+handle were written and how many branches.""" post_id: str handle: str dir_name: str @@ -144,11 +154,13 @@ class EmitResult: def _url_media_id(url: str, index: int) -> str: + """Derive a stable media id from the URL's last path segment, falling back to ``m``.""" segment = Path(unquote(urlparse(url).path)).name return segment if segment else f"m{index}" def _url_ext(url: str) -> str | None: + """Extract a lowercase file extension from a URL's ``?format=`` query or path suffix.""" parsed = urlparse(url) fmt = parse_qs(parsed.query).get("format", [None])[0] if fmt: @@ -166,6 +178,7 @@ def _media_files(media_dir: Path) -> list[Path]: def _take_preferred(unused: list[Path], post_id: str, index: int) -> Path | None: + """Pop the unused file whose stem equals ``_`` if any.""" want = f"{post_id}_{index}" for path in unused: if path.stem == want: @@ -174,6 +187,7 @@ def _take_preferred(unused: list[Path], post_id: str, index: int) -> Path | None def _take_fallback(unused: list[Path], post_id: str) -> Path | None: + """Pop the first unused file whose name starts with ``_`` if any.""" prefix = f"{post_id}_" for path in unused: if path.name.startswith(prefix): @@ -187,9 +201,10 @@ def collect_media( dest_dir: Path, *, now: str, -) -> list[dict[str, object]]: +) -> list[MediaItem]: + """Copy each post's media files from ``input_dir/media`` into ``dest_dir`` and build canonical MediaItems.""" unused = _media_files(input_dir / "media") - items: list[dict[str, object]] = [] + items: list[MediaItem] = [] for post in thread.posts: for index, url in enumerate(post.media_urls, start=1): media_id = _url_media_id(url, index) @@ -221,22 +236,23 @@ def collect_media( def selected_media_by_post( - items: list[dict[str, object]], + items: list[MediaItem], ) -> dict[str, tuple[str, ...]]: + """Group embedded items by post_id, returning the HTTPS URL of each visible location.""" grouped: dict[str, list[str]] = {} for item in items: - if not item.get("embed"): + if not item.embed: continue url = selected_url(item) if url is not None: - grouped.setdefault(str(item.get("post_id") or ""), []).append(url) + grouped.setdefault(item.post_id, []).append(url) return {post_id: tuple(urls) for post_id, urls in grouped.items()} def merge_existing_media( dest_dir: Path, - items: list[dict[str, object]], -) -> list[dict[str, object]]: + items: list[MediaItem], +) -> list[MediaItem]: """Merge existing derived/fallback rows with fresh canonical items.""" old_path = dest_dir / "media.json" if not old_path.is_file(): @@ -249,7 +265,8 @@ def merge_existing_media( raise SystemExit( "legacy media.json requires: tw.py migrate-media --id --apply" ) - return merge_manifest_items(list(old.get("items") or []), items) + existing = _from_wire_dict(old).items + return list(merge_manifest_items(existing, tuple(items))) def _legacy_emit_message(*args: object, **kwargs: object) -> None: @@ -276,6 +293,7 @@ def empty_text_ids(thread: ThreadData) -> list[str]: def render_gaps(thread: ThreadData, spine: list[str], *, input_kind: str = "root") -> str: + """Render a human-readable gaps report (quote_of, missing_reply_to, empty_text) for the thread.""" ids = by_id(thread) suggested = spine[-1] if spine else "" quote_ids = [p.post_id for p in thread.posts if p.quote_of_id] @@ -304,6 +322,7 @@ def render_gaps(thread: ThreadData, spine: list[str], *, input_kind: str = "root def _frontmatter_block(text: str) -> str: + """Return the raw YAML block between the ``---`` fences, or ``""`` if not parseable.""" if not text.startswith("---"): return "" rest = text[3:] @@ -316,6 +335,7 @@ def _frontmatter_block(text: str) -> str: def _frontmatter_post_id(text: str) -> str | None: + """Pull ``post_id`` out of an archive note's YAML frontmatter (returns ``None`` if absent).""" for line in _frontmatter_block(text).splitlines(): match = _POST_ID_LINE.match(line.strip()) if match: @@ -341,6 +361,7 @@ def collect_existing_ids(vault: Path) -> tuple[set[str], dict[str, Path]]: def unique_dir_name(parent: Path, base: str) -> str: + """Return ``base`` if unused, else ``base-2``, ``base-3``, ... until a free name is found.""" if not (parent / base).exists(): return base n = 2 @@ -350,6 +371,7 @@ def unique_dir_name(parent: Path, base: str) -> str: def _split_frontmatter(text: str) -> tuple[str, str]: + """Split an archive note into (frontmatter_with_closing_fence, body). Returns ``("", text)`` if no fence.""" if not text.startswith("---"): return "", text end = text.find("\n---", 3) @@ -425,14 +447,17 @@ def _upsert_wikilink(path: Path, target: str) -> None: def wiki_thread(handle: str, dir_name: str) -> str: + """Build the vault-root wikilink target for a thread spine folder.""" return f"archive/threads/{handle}/{dir_name}" def wiki_branch(handle: str, dir_name: str, name: str) -> str: + """Build the vault-root wikilink target for one branch file.""" return f"archive/threads/{handle}/{dir_name}/{name}" def ensure_handle_index(vault: Path, handle: str, dir_name: str) -> None: + """Create the per-handle index.md if missing, otherwise upsert the new thread wikilink.""" path = vault / "archive" / "threads" / handle / "index.md" target = wiki_thread(handle, dir_name) if not path.is_file(): @@ -459,6 +484,7 @@ def ensure_handle_index(vault: Path, handle: str, dir_name: str) -> None: def ensure_threads_index(vault: Path, handle: str) -> None: + """Create the top-level archive/threads/index.md if missing, otherwise upsert the handle wikilink.""" path = vault / "archive" / "threads" / "index.md" if not path.is_file(): path.parent.mkdir(parents=True, exist_ok=True) @@ -492,6 +518,7 @@ def ensure_threads_index(vault: Path, handle: str) -> None: def discover_dumps(root: Path) -> list[Path]: + """Return the sorted list of dump subdirectories that contain ``thread_data.json``.""" dumps = [ path for path in root.iterdir() if path.is_dir() and (path / "thread_data.json").is_file() @@ -642,7 +669,7 @@ def emit( media_doc = { "schema_version": 2, "root_post_id": thread.root_post_id, - "items": items, + "items": [_item_to_wire(item) for item in items], "mirrors": [], } atomic_write_json(asset_dir / "media.json", media_doc) @@ -691,7 +718,8 @@ def emit_all( missing = missing_parent_ids(thread) empty = empty_text_ids(thread) post_id = thread.root_post_id - except Exception: + except Exception as exc: + logger.warning("emit_all: skipping %s: %s", dump, exc) lines.append(f"{dump.name} ? error - -") errors += 1 continue @@ -712,7 +740,8 @@ def emit_all( force=force, reuse_dir=reuse, ) - except Exception: + except Exception as exc: + logger.warning("emit_all: skipping %s: %s", dump, exc) errors += 1 continue diff --git a/scripts/twitter/fallback_media.py b/scripts/twitter/fallback_media.py index 257a801..ef83d90 100644 --- a/scripts/twitter/fallback_media.py +++ b/scripts/twitter/fallback_media.py @@ -1,3 +1,4 @@ +"""Activate, restore, and dedupe fallback media locations for a thread.""" from __future__ import annotations import json @@ -8,6 +9,8 @@ from typing import Any, Callable try: from .media_manifest import ( + _from_wire_dict, + _item_to_wire, atomic_write_json, find_location, hash_file, @@ -15,10 +18,19 @@ try: selected_url, ) from .media_refs import atomic_write_text, remote_markup + from .models import ( + MediaItem, + MediaLocation, + MediaLocationCheck, + MediaManifest, + Publication, + ) 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 ( + _from_wire_dict, + _item_to_wire, atomic_write_json, find_location, hash_file, @@ -26,15 +38,24 @@ except ImportError: # pragma: no cover - script-mode import selected_url, ) from media_refs import atomic_write_text, remote_markup + from models import ( + MediaItem, + MediaLocation, + MediaLocationCheck, + MediaManifest, + Publication, + ) @dataclass(frozen=True) class FallbackResult: + """Outcome of activating one fallback location for a media item.""" url: str reused: bool def replace_selected_url(note_dir: Path, old_url: str, new_url: str) -> None: + """Swap ``old_url`` for ``new_url`` in exactly one note under ``note_dir`` (no-op if already swapped).""" if old_url == new_url: return old_markup = remote_markup(old_url) @@ -56,27 +77,27 @@ def replace_selected_url(note_dir: Path, old_url: str, new_url: str) -> None: 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 _group(manifest: MediaManifest, media_id: str) -> list[MediaItem]: + """Return the slice of ``manifest.items`` sharing a media_id (orig + derived variants).""" + return [item for item in manifest.items if item.media_id == 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] +def _one(items: list[MediaItem], role: str) -> MediaItem: + """Return the single item in ``items`` whose ``kind`` matches ``role`` (raises otherwise).""" + matches = [item for item in items if item.kind == 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]: +def _verified_local(item: MediaItem, asset_dir: Path) -> tuple[Path, str]: + """Return ``(local_path, sha256)`` of ``item``'s local location, verified against its declared hash.""" local = find_location(item, "local") - if local is None or local.get("integrity") != "present": + if local is None or local.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 "") + filename = str(local.local_path) if local.local_path is not None else "" + path = asset_dir / filename + expected = local.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 @@ -94,30 +115,32 @@ def activate_fallback( upload: Callable[[Path], str], lookup: Callable[[str, str], str | None] | None = None, ) -> FallbackResult: + """Activate a fallback URL for ``media_id``: upload (or reuse) and update manifest + note references.""" 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")) + raw = json.loads(media_path.read_text(encoding="utf-8")) + manifest: MediaManifest = _from_wire_dict(raw) 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)) + visible = next((item for item in group if item.embed), original) + old_url = selected_url(visible) 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 + target = _with_origin_unavailable(target, origin, 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 + for location in target.locations + if location.kind == "fallback" + and location.provider == provider + and location.sha256 == content_hash ), None, ) @@ -132,7 +155,7 @@ def activate_fallback( recorded_at=now, uploaded_at=None, ) - target.setdefault("locations", []).append(fallback) + target = _append_location(target, fallback) reused = True if fallback is None: url = upload(local_path) @@ -143,19 +166,141 @@ def activate_fallback( recorded_at=now, uploaded_at=now, ) - target.setdefault("locations", []).append(fallback) + target = _append_location(target, 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", + manifest = _with_replaced_item(manifest, target) + atomic_write_json(media_path, _manifest_to_wire_shape(manifest, raw)) + assert fallback.url is not None + replace_selected_url(note_dir, old_url, fallback.url) + manifest = _with_item_attr(manifest, target.post_id, target.media_id, "embed", True) + manifest = _with_publication( + manifest, + target, + Publication( + selected_location_id=fallback.id or "", + selected_at=now, + reason="origin-unavailable", + ), + ) + atomic_write_json(media_path, _manifest_to_wire_shape(manifest, raw)) + return FallbackResult(str(fallback.url or ""), reused) + + +def _with_origin_unavailable(item: MediaItem, origin: MediaLocation, now: str) -> MediaItem: + """Return a copy of `item` with the matching origin location marked unavailable.""" + new_locations: list[MediaLocation] = [] + for location in item.locations: + if location.id == origin.id: + new_locations.append( + MediaLocation( + kind=location.kind, + id=location.id, + local_path=location.local_path, + url=location.url, + bytes=location.bytes, + sha256=location.sha256, + media_type=location.media_type, + integrity=location.integrity, + verified_at=location.verified_at, + provider=location.provider, + availability="unavailable", + checked_at=location.checked_at, + checked_status=location.checked_status, + recorded_at=location.recorded_at, + uploaded_at=location.uploaded_at, + confirmed_unavailable_at=now, + check=location.check, + ) + ) + else: + new_locations.append(location) + return _copy_item_with(item, locations=tuple(new_locations)) + + +def _append_location(item: MediaItem, location: MediaLocation) -> MediaItem: + return _copy_item_with(item, locations=item.locations + (location,)) + + +def _copy_item_with(item: MediaItem, **changes: Any) -> MediaItem: + return MediaItem( + post_id=changes.get("post_id", item.post_id), + media_id=changes.get("media_id", item.media_id), + kind=changes.get("kind", item.kind), + role=changes.get("role", item.role), + handle=changes.get("handle", item.handle), + embed=changes.get("embed", item.embed), + caption=changes.get("caption", item.caption), + derived_from=changes.get("derived_from", item.derived_from), + publication=changes.get("publication", item.publication), + locations=changes.get("locations", item.locations), + ) + + +def _with_replaced_item(manifest: MediaManifest, replacement: MediaItem) -> MediaManifest: + new_items: list[MediaItem] = [] + for existing in manifest.items: + if existing.post_id == replacement.post_id and existing.media_id == replacement.media_id: + new_items.append(replacement) + else: + new_items.append(existing) + return MediaManifest( + root_post_id=manifest.root_post_id, + items=tuple(new_items), + schema_version=manifest.schema_version, + captured_at=manifest.captured_at, + ) + + +def _with_item_attr( + manifest: MediaManifest, + post_id: str, + media_id: str, + attr: str, + value: Any, +) -> MediaManifest: + new_items: list[MediaItem] = [] + for existing in manifest.items: + if existing.post_id == post_id and existing.media_id == media_id: + new_items.append(_copy_item_with(existing, **{attr: value})) + else: + new_items.append(existing) + return MediaManifest( + root_post_id=manifest.root_post_id, + items=tuple(new_items), + schema_version=manifest.schema_version, + captured_at=manifest.captured_at, + ) + + +def _with_publication( + manifest: MediaManifest, + item: MediaItem, + publication: Publication, +) -> MediaManifest: + new_items: list[MediaItem] = [] + for existing in manifest.items: + if existing.post_id == item.post_id and existing.media_id == item.media_id: + new_items.append(_copy_item_with(existing, publication=publication)) + else: + new_items.append(existing) + return MediaManifest( + root_post_id=manifest.root_post_id, + items=tuple(new_items), + schema_version=manifest.schema_version, + captured_at=manifest.captured_at, + ) + + +def _manifest_to_wire_shape(manifest: MediaManifest, raw: dict) -> dict: + out: dict = { + "schema_version": manifest.schema_version or 2, + "root_post_id": manifest.root_post_id, + "items": [_item_to_wire(item) for item in manifest.items], + "mirrors": list(raw.get("mirrors") or []), } - atomic_write_json(media_path, manifest) - return FallbackResult(str(fallback["url"]), reused) + if manifest.captured_at is not None: + out["captured_at"] = manifest.captured_at + return out def find_existing_fallback( @@ -163,20 +308,23 @@ def find_existing_fallback( provider: str, content_hash: str, ) -> str | None: + """Scan every thread's media.json for a fallback URL with matching sha256; returns the first or ``None``.""" + urls: set[str] = set() 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: + raw = json.loads(media_path.read_text(encoding="utf-8")) + if raw.get("schema_version") != 2: continue - for item in data.get("items") or []: - for location in item.get("locations") or []: + manifest = _from_wire_dict(raw) + for item in manifest.items: + for location in item.locations: if ( - location.get("kind") == "fallback" - and location.get("provider") == provider - and location.get("sha256") == content_hash - and str(location.get("url") or "").startswith("https://") + location.kind == "fallback" + and location.provider == provider + and location.sha256 == content_hash + and (location.url or "").startswith("https://") ): - urls.add(str(location["url"])) + urls.add(location.url or "") return sorted(urls)[0] if urls else None @@ -187,31 +335,74 @@ def restore_origin( media_id: str, now: str, ) -> None: + """Switch the visible location for ``media_id`` back to its X origin and rewrite the note references.""" media_path = asset_dir / "media.json" - manifest = json.loads(media_path.read_text(encoding="utf-8")) + media_path = asset_dir / "media.json" + raw = json.loads(media_path.read_text(encoding="utf-8")) + manifest = _from_wire_dict(raw) 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) + visible = next((item for item in group if item.embed), original) old_url = selected_url(visible) - new_url = str(origin.get("url") or "") + new_url = origin.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) \ No newline at end of file + new_locations: list[MediaLocation] = [] + for location in original.locations: + if location.id == origin.id: + new_locations.append( + MediaLocation( + kind=location.kind, + id=location.id, + local_path=location.local_path, + url=location.url, + bytes=location.bytes, + sha256=location.sha256, + media_type=location.media_type, + integrity=location.integrity, + verified_at=location.verified_at, + provider=location.provider, + availability="available", + checked_at=now, + checked_status=location.checked_status, + recorded_at=location.recorded_at, + uploaded_at=location.uploaded_at, + confirmed_unavailable_at=location.confirmed_unavailable_at, + check=MediaLocationCheck( + status=None, + result="available", + detail="manual restore-origin selection", + ), + ) + ) + else: + new_locations.append(location) + manifest = _with_replaced_item( + manifest, _copy_item_with(original, locations=tuple(new_locations)) + ) + new_items: list[MediaItem] = [] + for existing in manifest.items: + if existing.post_id == original.post_id and existing.media_id == original.media_id: + new_items.append(_copy_item_with(existing, embed=True)) + else: + new_items.append(_copy_item_with(existing, embed=False)) + manifest = MediaManifest( + root_post_id=manifest.root_post_id, + items=tuple(new_items), + schema_version=manifest.schema_version, + captured_at=manifest.captured_at, + ) + manifest = _with_publication( + manifest, + original, + Publication( + selected_location_id="origin:x", + selected_at=now, + reason="manual", + ), + ) + atomic_write_json(media_path, _manifest_to_wire_shape(manifest, raw)) diff --git a/scripts/twitter/frozen.py b/scripts/twitter/frozen.py index 77593d5..7f67fa5 100644 --- a/scripts/twitter/frozen.py +++ b/scripts/twitter/frozen.py @@ -1,3 +1,4 @@ +"""Frozen-id helpers: refuse writes to threads matching the do-not-refetch list.""" from __future__ import annotations import json diff --git a/scripts/twitter/graph_dry_run.py b/scripts/twitter/graph_dry_run.py index 09beac4..06c72e9 100644 --- a/scripts/twitter/graph_dry_run.py +++ b/scripts/twitter/graph_dry_run.py @@ -7,7 +7,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from twitter.models import load_thread +from twitter.models import PostData, ThreadData, load_thread from twitter.paths import SCRATCH from twitter.slug import slugify from twitter.tree import branch_roots, by_id, children_map, descendants, spine_ids @@ -18,11 +18,11 @@ def _one_line(text: str, n: int = 56) -> str: return s if len(s) <= n else s[: n - 1] + "…" -def _node(post) -> str: +def _node(post: PostData) -> str: return f"[I:{post.handle}:{post.post_id}]" -def render_ssdl(thread) -> str: +def render_ssdl(thread: ThreadData) -> str: ids = by_id(thread) kids = children_map(thread) spine = spine_ids(thread) @@ -45,7 +45,7 @@ def render_ssdl(thread) -> str: return "\n".join(lines) -def render_ascii(thread) -> str: +def render_ascii(thread: ThreadData) -> str: ids = by_id(thread) kids = children_map(thread) spine = spine_ids(thread) diff --git a/scripts/twitter/ingest_gallery.py b/scripts/twitter/ingest_gallery.py index 9a93c29..59f594d 100644 --- a/scripts/twitter/ingest_gallery.py +++ b/scripts/twitter/ingest_gallery.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse import json import sys +from collections.abc import Mapping from dataclasses import asdict from pathlib import Path from typing import Any @@ -13,77 +14,106 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from twitter.models import PostData, PostMetrics, ThreadData -def _find_tweet_meta(entry: Any) -> dict[str, Any] | None: - if isinstance(entry, dict): - if "tweet_id" in entry or ("content" in entry and "author" in entry): - return entry - return None - if isinstance(entry, (list, tuple)): - for item in entry: - meta = _find_tweet_meta(item) - if meta is not None: - return meta - return None +def _parse_wire_post(meta: Any) -> PostData | None: + """Boundary: parse an MCP wire entry into a typed PostData. + + A wire entry may be a dict (tweet meta) or a nested list/tuple + containing the meta dict. Recursively walks to find the first + tweet-meta dict (one with ``tweet_id`` or both ``content`` and + ``author``) and converts it to PostData. Returns ``None`` if no + tweet-meta dict is found in the entry — that entry is then + dropped by the caller, matching the prior ``_find_tweet_meta`` + skip-on-None behavior. + """ + result: PostData | None = None + + def build_post(meta_dict: Mapping[str, object]) -> PostData: + author = meta_dict.get("author") or {} + if isinstance(author, dict): + author_name = str(author.get("nick") or author.get("name") or "") + handle = str(author.get("name") or author.get("nick") or "") + else: + author_name = str(author) + handle = str(author) + view = meta_dict.get("view_count") + reply = meta_dict.get("reply_id") + quote = meta_dict.get("quote_id") + return PostData( + post_id=str(meta_dict.get("tweet_id") or meta_dict.get("id") or ""), + author=author_name, + handle=handle, + text=str(meta_dict.get("content") or ""), + timestamp=str(meta_dict.get("date") or ""), + media_urls=(), + reply_to_id=str(reply) if reply else None, + quote_of_id=str(quote) if quote else None, + metrics=PostMetrics( + reply_count=int(meta_dict.get("reply_count") or 0), + repost_count=int(meta_dict.get("retweet_count") or 0), + like_count=int(meta_dict.get("favorite_count") or 0), + view_count=int(view) if view is not None else None, + ), + ) + + def walk(node: object) -> None: + nonlocal result + if result is not None: + return + if isinstance(node, dict): + if "tweet_id" in node or ("content" in node and "author" in node): + result = build_post(node) + return + if isinstance(node, (list, tuple)): + for item in node: + if result is not None: + return + walk(item) + + walk(meta) + return result -def _media_url_from_entry(entry: Any) -> str: - if isinstance(entry, (list, tuple)): - for item in entry: - if isinstance(item, str) and ( - "pbs.twimg.com" in item or "video.twimg.com" in item - ): - return item - return "" +def _media_urls_from_entry(entry: object) -> tuple[str, ...]: + """Walk a wire entry to collect ``pbs.twimg.com`` / ``video.twimg.com`` URL strings. + + Mirrors the recursive descent of the boundary parser (without the + dict-detection short-circuit) so a single walk collects both the + tweet meta (in the boundary) and any media URLs in the same shape. + """ + urls: list[str] = [] + + def walk(node: object) -> None: + if isinstance(node, (list, tuple)): + for item in node: + if isinstance(item, str) and ( + "pbs.twimg.com" in item or "video.twimg.com" in item + ): + urls.append(item) + else: + walk(item) + + walk(entry) + return tuple(urls) -def _post_from_meta(meta: dict[str, Any]) -> PostData: - author = meta.get("author") or {} - if isinstance(author, dict): - author_name = str(author.get("nick") or author.get("name") or "") - handle = str(author.get("name") or author.get("nick") or "") - else: - author_name = str(author) - handle = str(author) - view = meta.get("view_count") - reply = meta.get("reply_id") - quote = meta.get("quote_id") - return PostData( - post_id=str(meta.get("tweet_id") or meta.get("id") or ""), - author=author_name, - handle=handle, - text=str(meta.get("content") or ""), - timestamp=str(meta.get("date") or ""), - media_urls=(), - reply_to_id=str(reply) if reply else None, - quote_of_id=str(quote) if quote else None, - metrics=PostMetrics( - reply_count=int(meta.get("reply_count") or 0), - repost_count=int(meta.get("retweet_count") or 0), - like_count=int(meta.get("favorite_count") or 0), - view_count=int(view) if view is not None else None, - ), - ) - - -def posts_from_gallery(data: Any) -> list[PostData]: +def posts_from_gallery(data: object) -> list[PostData]: by: dict[str, PostData] = {} media: dict[str, list[str]] = {} order: list[str] = [] entries = data if isinstance(data, list) else [] for entry in entries: - meta = _find_tweet_meta(entry) - if meta is None: - continue - tid = str(meta.get("tweet_id") or meta.get("id") or "") - if not tid: + post = _parse_wire_post(entry) + if post is None or not post.post_id: continue + tid = post.post_id if tid not in by: - by[tid] = _post_from_meta(meta) - media[tid] = [] + by[tid] = post + media[tid] = list(_media_urls_from_entry(entry)) order.append(tid) - url = _media_url_from_entry(entry) - if url: - media[tid].append(url) + else: + for url in _media_urls_from_entry(entry): + if url not in media[tid]: + media[tid].append(url) posts: list[PostData] = [] for tid in order: base = by[tid] diff --git a/scripts/twitter/media_audit.py b/scripts/twitter/media_audit.py index 66fb4f9..af44032 100644 --- a/scripts/twitter/media_audit.py +++ b/scripts/twitter/media_audit.py @@ -1,17 +1,20 @@ +"""Audit per-thread media manifest: local integrity, note refs, mirror freshness, origin probes.""" from __future__ import annotations import json import sys from collections import Counter -from dataclasses import dataclass +from dataclasses import dataclass, replace +from http.client import HTTPResponse from pathlib import Path -from typing import Any +from typing import Callable from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen try: from .frozen import frozen_match from .media_manifest import ( + _from_wire_dict, find_location, hash_file, inventory_digest, @@ -21,11 +24,13 @@ try: validate_manifest, ) from .media_refs import remote_markup + from .models import LegacyMediaJson, MediaItem, MediaLocation, MediaLocationCheck, MediaManifest, OriginCheck 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 ( + _from_wire_dict, find_location, hash_file, inventory_digest, @@ -35,42 +40,68 @@ except ImportError: # pragma: no cover - script-mode import validate_manifest, ) from media_refs import remote_markup + from models import ( + LegacyMediaJson, + MediaItem, + MediaLocation, + MediaLocationCheck, + MediaManifest, + OriginCheck, + ) @dataclass(frozen=True) class AuditReport: + """Outcome of a per-thread media audit: frozen flag plus blocking issues.""" frozen: bool issues: tuple[str, ...] -def audit_local_item(item: dict[str, Any], asset_dir: Path) -> str | None: +def audit_local_item(item: MediaItem, asset_dir: Path) -> str | None: local = find_location(item, "local") if local is None: return f"{'/'.join(item_key(item))} missing local location" - path = asset_dir / str(local.get("path") or "") + filename = str(local.local_path) if local.local_path is not None else "" + path = asset_dir / filename if not path.is_file(): return f"{'/'.join(item_key(item))} local missing" - if path.stat().st_size != local.get("bytes") or hash_file(path) != local.get("sha256"): + sha_expected = local.sha256 or "" + bytes_expected = local.bytes + if bytes_expected is not None and path.stat().st_size != bytes_expected: + return f"{'/'.join(item_key(item))} local mismatch" + if hash_file(path) != sha_expected: return f"{'/'.join(item_key(item))} local mismatch" return None -def classify_origin_response(status: int | None, detail: str, checked_at: str) -> dict[str, Any]: +def classify_origin_response(status: int | None, detail: str, checked_at: str) -> OriginCheck: + """Build an ``OriginCheck`` from an HTTP probe status + detail string. + + The detail string is truncated to 200 chars so the persisted check + does not bloat the manifest. ``confirms_unavailable`` is always False + here; callers that detect a definitive unavailability set it True. + """ available = status is not None and 200 <= status < 400 - return { - "checked_at": checked_at, - "status": status, - "result": "available" if available else "error", - "detail": detail[:200], - "confirms_unavailable": False, - } + return OriginCheck( + checked_at=checked_at, + status=status, + result="available" if available else "error", + detail=detail[:200], + confirms_unavailable=False, + ) def check_origin_url( url: str, checked_at: str, - opener: Any = urlopen, -) -> dict[str, Any]: + opener: Callable[..., HTTPResponse] = urlopen, +) -> OriginCheck: + """Probe ``url`` with HEAD and return a typed ``OriginCheck``. + + ``opener`` is variadic because ``urlopen`` accepts both ``url`` and + ``Request`` shapes with different keyword argument sets; callers + passing a custom opener can match either. + """ request = Request(url, method="HEAD", headers={"User-Agent": "Threadwell-media-audit/1"}) try: with opener(request, timeout=30) as response: @@ -85,18 +116,50 @@ def check_origin_url( return classify_origin_response(None, type(exc).__name__, checked_at) -def record_origin_check(item: dict[str, Any], outcome: dict[str, Any]) -> None: - origin = find_location(item, "origin:x") - if origin is None: +def record_origin_check( + manifest: LegacyMediaJson, item_index: int, outcome: OriginCheck +) -> LegacyMediaJson: + """Return a new ``LegacyMediaJson`` with the origin check at ``item_index`` persisted. + + The input ``LegacyMediaJson`` is frozen, so the update is a + round-trip: each affected ``MediaItem`` is reconstructed with the + updated ``MediaLocation`` for ``origin:x`` carrying the check + metadata. Returns the new manifest; the caller is responsible for + serializing it back to disk. + """ + items = list(manifest.items) + if not (0 <= item_index < len(items)): + raise ValueError("item_index out of range") + item = items[item_index] + new_locations: list[MediaLocation] = [] + found_origin = False + for location in item.locations: + if location.id == "origin:x": + found_origin = True + new_availability = ( + "available" + if outcome.result == "available" + else location.availability + ) + new_locations.append( + replace( + location, + availability=new_availability, + checked_at=outcome.checked_at, + checked_status=outcome.status, + check=MediaLocationCheck( + status=outcome.status, + result=outcome.result, + detail=outcome.detail, + ), + ) + ) + else: + new_locations.append(location) + if not found_origin: raise ValueError("item has no origin:x location") - origin["checked_at"] = outcome["checked_at"] - origin["check"] = { - "status": outcome["status"], - "result": outcome["result"], - "detail": outcome["detail"], - } - if outcome["result"] == "available": - origin["availability"] = "available" + items[item_index] = replace(item, locations=tuple(new_locations)) + return replace(manifest, items=tuple(items)) def audit_thread( @@ -104,21 +167,30 @@ def audit_thread( note_dir: Path, frozen_ids: set[str], ) -> AuditReport: + """Audit a single thread's media manifest + note references. + + Reads ``media.json`` at the boundary into ``LegacyMediaJson`` (for the + opaque mirrors blob) and a typed ``MediaManifest`` (for item / + location validation). Reports issues with the local file presence, + note reference counts, and mirror staleness. + """ match = frozen_match(asset_dir, frozen_ids) if match is not None: return AuditReport(True, (f"frozen: skipped ({match})",)) - manifest = json.loads((asset_dir / "media.json").read_text(encoding="utf-8")) - issues = list(validate_manifest(manifest)) - for item in manifest.get("items") or []: + manifest_dict = json.loads((asset_dir / "media.json").read_text(encoding="utf-8")) + legacy = LegacyMediaJson.from_dict(manifest_dict) + manifest = _from_wire_dict(manifest_dict) + issues: list[str] = list(validate_manifest(manifest)) + for item in manifest.items: if find_location(item, "local") is not None: issue = audit_local_item(item, asset_dir) if issue is not None: issues.append(issue) expected = Counter( - remote_markup(url) - for item in manifest.get("items") or [] - if item.get("embed") and (url := selected_url(item)) is not None + remote_markup(url or "") + for item in manifest.items + if item.embed and (url := selected_url(item)) is not None ) note_text = "\n".join( path.read_text(encoding="utf-8") @@ -132,7 +204,7 @@ def audit_thread( ) current_digest = inventory_digest(payload_inventory(asset_dir, manifest)) - for mirror in manifest.get("mirrors") or []: + for mirror in (legacy.mirrors or {}).values(): if mirror.get("state") == "synced" and mirror.get("inventory_digest") != current_digest: issues.append(f"mirror stale: {mirror.get('destination_id')}") - return AuditReport(False, tuple(sorted(set(issues)))) \ No newline at end of file + return AuditReport(False, tuple(sorted(set(issues)))) diff --git a/scripts/twitter/media_embed.py b/scripts/twitter/media_embed.py index 374c8ab..cd71648 100644 --- a/scripts/twitter/media_embed.py +++ b/scripts/twitter/media_embed.py @@ -12,14 +12,28 @@ import sys from pathlib import Path try: - from .media_manifest import atomic_write_json, find_location, selected_url, validate_manifest + from .media_manifest import ( + _from_wire_dict, + _item_to_wire, + atomic_write_json, + find_location, + selected_url, + validate_manifest, + ) 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, selected_url, validate_manifest + from media_manifest import ( + _from_wire_dict, + _item_to_wire, + atomic_write_json, + find_location, + selected_url, + validate_manifest, + ) -OCR_DETAILS = "Text from still" +OCR_DETAILS: str = "Text from still" def ocr_block(text: str) -> str: @@ -83,34 +97,36 @@ def main(argv: list[str] | None = None) -> int: media_path = args.thread / "media.json" data = json.loads(media_path.read_text(encoding="utf-8")) - issues = validate_manifest(data) + manifest = _from_wire_dict(data) + issues = validate_manifest(manifest) if issues: raise SystemExit(f"manifest invalid: {issues[0]}") - target = None - for item in data.get("items") or []: + raw_items: list[dict] = list(data.get("items") or []) + target_dict: dict | None = None + for item in raw_items: if ( str(item.get("media_id") or "") == args.media_id and str(item.get("role") or "") == "ocr" ): - target = item + target_dict = item break - if target is None: + if target_dict is None: raise SystemExit(f"no role=ocr for {args.media_id}; run ocr_pass.py first") - ocr_path = args.thread / str(target.get("filename") or "") + ocr_path = args.thread / str(target_dict.get("filename") or "") if not ocr_path.is_file(): raise SystemExit(f"missing {ocr_path}") visible = None - for item in data.get("items") or []: - if str(item.get("media_id") or "") != args.media_id: + for item in manifest.items: + if item.media_id != args.media_id: continue - if item.get("embed") and selected_url(item): + if item.embed and selected_url(item): visible = item break if visible is None: raise SystemExit("media group has no embedded origin") - image_token = str(selected_url(visible) or "") + image_token = selected_url(visible) or "" if not image_token: raise SystemExit("media group has no selected HTTPS location") attach_ocr(args.notes, image_token, ocr_path.read_text(encoding="utf-8")) diff --git a/scripts/twitter/media_manifest.py b/scripts/twitter/media_manifest.py index 926ad14..e95f99c 100644 --- a/scripts/twitter/media_manifest.py +++ b/scripts/twitter/media_manifest.py @@ -1,3 +1,4 @@ +"""Canonical media manifest: boundary parsing, validation, and wire round-trip.""" from __future__ import annotations import copy @@ -9,9 +10,28 @@ from hashlib import sha256 from pathlib import Path from typing import Any +try: + from .models import ( + DerivedFromRef, + MediaItem, + MediaLocation, + MediaLocationCheck, + MediaManifest, + Publication, + ) +except ImportError: # pragma: no cover - script-mode import + from models import ( + DerivedFromRef, + MediaItem, + MediaLocation, + MediaLocationCheck, + MediaManifest, + Publication, + ) -SCHEMA_VERSION = 2 -VISUAL_ROLES = {"orig", "crt", "crt_outline", "denoise"} + +SCHEMA_VERSION: int = 2 +VISUAL_ROLES: set[str] = {"orig", "crt", "crt_outline", "denoise"} def utc_now() -> str: @@ -41,40 +61,51 @@ def atomic_write_json(path: Path, data: dict) -> None: temp.unlink() -def item_key(item: dict[str, Any]) -> tuple[str, str, str]: - return ( - str(item.get("post_id") or ""), - str(item.get("media_id") or ""), - str(item.get("role") or ""), - ) +def _from_wire_dict(raw: dict[str, Any]) -> MediaManifest: + """Boundary parser: turn a wire-shape dict into a typed MediaManifest.""" + return MediaManifest.from_dict(raw) -def find_location(item: dict[str, Any], location_id: str) -> dict[str, Any] | None: - for location in item.get("locations") or []: - if str(location.get("id") or "") == location_id: +def _from_wire_item(raw: dict[str, Any]) -> MediaItem: + """Boundary parser: turn a wire-shape dict into a typed MediaItem.""" + return MediaItem.from_dict(raw) + + +def _from_wire_items(raw: list[dict[str, Any]]) -> tuple[MediaItem, ...]: + """Boundary parser: turn a wire-shape items list into typed items.""" + return tuple(MediaItem.from_dict(item) for item in raw) + + +def item_key(item: MediaItem) -> tuple[str, str, str]: + return (item.post_id, item.media_id, item.kind or item.role or "") + + +def find_location(item: MediaItem, location_id: str) -> MediaLocation | None: + for location in item.locations: + if location.id == location_id: return location return None -def location_of_kind(item: dict[str, Any], kind: str) -> dict[str, Any] | None: - for location in item.get("locations") or []: - if location.get("kind") == kind: +def location_of_kind(item: MediaItem, kind: str) -> MediaLocation | None: + for location in item.locations: + if location.kind == kind: return location return None -def selected_location(item: dict[str, Any]) -> dict[str, Any] | None: - publication = item.get("publication") - if not isinstance(publication, dict): +def selected_location(item: MediaItem) -> MediaLocation | None: + publication = item.publication + if publication is None: return None - return find_location(item, str(publication.get("selected_location_id") or "")) + return find_location(item, publication.selected_location_id) -def selected_url(item: dict[str, Any]) -> str | None: +def selected_url(item: MediaItem) -> str | None: location = selected_location(item) - if not location or location.get("kind") not in {"origin", "fallback"}: + if location is None or location.kind not in {"origin", "fallback"}: return None - value = str(location.get("url") or "") + value = location.url or "" return value if value.startswith("https://") else None @@ -83,18 +114,18 @@ def _media_type(filename: str) -> str: return guessed or "application/octet-stream" -def _local_location(filename: str, local_path: Path, now: str) -> dict[str, Any]: +def _local_location(filename: str, local_path: Path, now: str) -> MediaLocation: present = local_path.is_file() - return { - "id": "local", - "kind": "local", - "path": filename, - "sha256": hash_file(local_path) if present else None, - "bytes": local_path.stat().st_size if present else None, - "media_type": _media_type(filename), - "integrity": "present" if present else "missing", - "verified_at": now, - } + return MediaLocation( + id="local", + kind="local", + local_path=Path(filename), + sha256=hash_file(local_path) if present else None, + bytes=local_path.stat().st_size if present else None, + media_type=_media_type(filename), + integrity="present" if present else "missing", + verified_at=now, + ) def new_original_item( @@ -106,73 +137,78 @@ def new_original_item( filename: str, local_path: Path, now: str, -) -> dict[str, Any]: - return { - "post_id": post_id, - "media_id": media_id, - "handle": handle, - "role": "orig", - "derived_from": None, - "embed": True, - "locations": [ - { - "id": "origin:x", - "kind": "origin", - "provider": "x", - "url": origin_url, - "availability": "unknown", - "checked_at": None, - "check": None, - "confirmed_unavailable_at": None, - }, +) -> MediaItem: + return MediaItem( + post_id=post_id, + media_id=media_id, + handle=handle, + kind="orig", + role="orig", + embed=True, + locations=( + MediaLocation( + id="origin:x", + kind="origin", + provider="x", + url=origin_url, + availability="unknown", + ), _local_location(filename, local_path, now), - ], - "publication": { - "selected_location_id": "origin:x", - "selected_at": now, - "reason": "default", - }, - } + ), + publication=Publication( + selected_location_id="origin:x", + selected_at=now, + reason="default", + ), + ) -def merge_item(existing: dict[str, Any], fresh: dict[str, Any]) -> dict[str, Any]: +def merge_item(existing: MediaItem, fresh: MediaItem) -> MediaItem: if item_key(existing) != item_key(fresh): raise ValueError("cannot merge different media items") - merged = copy.deepcopy(existing) - merged["handle"] = fresh["handle"] + fresh_handle = fresh.handle or existing.handle existing_origin = location_of_kind(existing, "origin") fresh_origin = location_of_kind(fresh, "origin") if ( existing_origin is not None and fresh_origin is not None - and existing_origin.get("url") != fresh_origin.get("url") + and existing_origin.url != fresh_origin.url ): raise ValueError(f"origin URL changed for {item_key(existing)!r}") fresh_local = location_of_kind(fresh, "local") - locations = [ + locations: list[MediaLocation] = [ copy.deepcopy(location) - for location in merged.get("locations") or [] - if location.get("kind") != "local" + for location in existing.locations + if location.kind != "local" ] if fresh_local is not None: locations.append(copy.deepcopy(fresh_local)) - if location_of_kind(merged, "origin") is None: - fresh_origin = location_of_kind(fresh, "origin") - if fresh_origin is not None: - locations.insert(0, copy.deepcopy(fresh_origin)) - merged["locations"] = locations - if merged.get("publication") is None: - merged["publication"] = copy.deepcopy(fresh.get("publication")) + if location_of_kind(existing, "origin") is None and fresh_origin is not None: + locations.insert(0, copy.deepcopy(fresh_origin)) + merged = MediaItem( + post_id=existing.post_id, + media_id=existing.media_id, + kind=existing.kind, + role=existing.role, + handle=fresh_handle, + embed=existing.embed, + locations=tuple(locations), + publication=existing.publication + if existing.publication is not None + else copy.deepcopy(fresh.publication), + ) return merged def merge_manifest_items( - existing_items: list[dict[str, Any]], - fresh_items: list[dict[str, Any]], -) -> list[dict[str, Any]]: - existing_by_key = {item_key(item): item for item in existing_items} + existing_items: tuple[MediaItem, ...], + fresh_items: tuple[MediaItem, ...], +) -> tuple[MediaItem, ...]: + existing_by_key: dict[tuple[str, str, str], MediaItem] = { + item_key(item): item for item in existing_items + } fresh_keys: set[tuple[str, str, str]] = set() - merged: list[dict[str, Any]] = [] + merged: list[MediaItem] = [] for fresh in fresh_items: key = item_key(fresh) fresh_keys.add(key) @@ -181,7 +217,7 @@ def merge_manifest_items( for previous in existing_items: if item_key(previous) not in fresh_keys: merged.append(copy.deepcopy(previous)) - return merged + return tuple(merged) def fallback_location_id(provider: str, url: str) -> str: @@ -196,49 +232,20 @@ def new_fallback_location( content_hash: str | None, recorded_at: str, uploaded_at: str | None, -) -> dict[str, Any]: - return { - "id": fallback_location_id(provider, url), - "kind": "fallback", - "provider": provider, - "url": url, - "sha256": content_hash, - "recorded_at": recorded_at, - "uploaded_at": uploaded_at, - "availability": "unknown", - "checked_at": None, - "check": None, - } +) -> MediaLocation: + return MediaLocation( + id=fallback_location_id(provider, url), + kind="fallback", + provider=provider, + url=url, + sha256=content_hash, + recorded_at=recorded_at, + uploaded_at=uploaded_at, + availability="unknown", + ) def new_derived_item( - *, - legacy: dict[str, Any], - asset_dir: Path, - now: str, -) -> dict[str, Any]: - filename = str(legacy.get("filename") or "") - post_id = str(legacy.get("post_id") or "") - media_id = str(legacy.get("media_id") or "") - role = str(legacy.get("role") or "") - return { - "post_id": post_id, - "media_id": media_id, - "handle": str(legacy.get("handle") or ""), - "role": role, - "derived_from": { - "post_id": post_id, - "media_id": media_id, - "role": "orig", - }, - "embed": False, - "locations": [_local_location(filename, asset_dir / filename, now)], - "publication": None, - } - - -def upsert_derived_item( - manifest: dict[str, Any], *, post_id: str, media_id: str, @@ -247,44 +254,178 @@ def upsert_derived_item( filename: str, asset_dir: Path, now: str, -) -> dict[str, Any]: - legacy_shape = { - "post_id": post_id, - "media_id": media_id, - "handle": handle, - "role": role, - "filename": filename, - } - fresh = new_derived_item(legacy=legacy_shape, asset_dir=asset_dir, now=now) +) -> MediaItem: + """Construct a MediaItem for a derived file (crt / crt_outline / denoise / ocr).""" + return MediaItem( + post_id=post_id, + media_id=media_id, + handle=handle, + kind=role, + role=role, + derived_from=DerivedFromRef( + post_id=post_id, + media_id=media_id, + role="orig", + ), + embed=False, + locations=(_local_location(filename, asset_dir / filename, now),), + publication=None, + ) + + +def upsert_derived_item( + manifest: MediaManifest, + *, + post_id: str, + media_id: str, + handle: str, + role: str, + filename: str, + asset_dir: Path, + now: str, +) -> tuple[MediaManifest, MediaItem]: + """Insert or update a derived item. Returns (manifest, item).""" + fresh = new_derived_item( + post_id=post_id, + media_id=media_id, + handle=handle, + role=role, + filename=filename, + asset_dir=asset_dir, + now=now, + ) key = item_key(fresh) - items = list(manifest.get("items") or []) + items: list[MediaItem] = list(manifest.items) for index, existing in enumerate(items): if item_key(existing) == key: - fresh["locations"] = [ - location - for location in existing.get("locations") or [] - if location.get("kind") != "local" - ] + fresh["locations"] - fresh["embed"] = bool(existing.get("embed")) - fresh["publication"] = copy.deepcopy(existing.get("publication")) - items[index] = fresh - manifest["items"] = items - return fresh + kept = tuple(loc for loc in existing.locations if loc.kind != "local") + updated = MediaItem( + post_id=fresh.post_id, + media_id=fresh.media_id, + kind=fresh.kind, + role=fresh.role, + handle=fresh.handle, + embed=existing.embed, + derived_from=fresh.derived_from, + locations=kept + fresh.locations, + publication=copy.deepcopy(existing.publication), + ) + items[index] = updated + new_manifest = MediaManifest( + root_post_id=manifest.root_post_id, + items=tuple(items), + schema_version=manifest.schema_version, + captured_at=manifest.captured_at, + ) + return new_manifest, updated items.append(fresh) - manifest["items"] = items - return fresh + new_manifest = MediaManifest( + root_post_id=manifest.root_post_id, + items=tuple(items), + schema_version=manifest.schema_version, + captured_at=manifest.captured_at, + ) + return new_manifest, fresh -def canonical_manifest_bytes(data: dict[str, Any]) -> bytes: - stable = copy.deepcopy(data) - stable["mirrors"] = [] +def _item_to_wire(item: MediaItem) -> dict[str, Any]: + """Convert a typed MediaItem back into a wire-shape dict for round-tripping to disk.""" + locations: list[dict[str, Any]] = [] + for loc in item.locations: + location_dict: dict[str, Any] = {"kind": loc.kind} + if loc.id is not None: + location_dict["id"] = loc.id + if loc.local_path is not None: + location_dict["path"] = str(loc.local_path) + if loc.url is not None: + location_dict["url"] = loc.url + if loc.bytes is not None: + location_dict["bytes"] = loc.bytes + if loc.sha256 is not None: + location_dict["sha256"] = loc.sha256 + if loc.media_type is not None: + location_dict["media_type"] = loc.media_type + if loc.integrity is not None: + location_dict["integrity"] = loc.integrity + if loc.verified_at is not None: + location_dict["verified_at"] = loc.verified_at + if loc.provider is not None: + location_dict["provider"] = loc.provider + if loc.availability is not None: + location_dict["availability"] = loc.availability + if loc.checked_at is not None: + location_dict["checked_at"] = loc.checked_at + if loc.checked_status is not None: + location_dict["checked_status"] = loc.checked_status + if loc.recorded_at is not None: + location_dict["recorded_at"] = loc.recorded_at + if loc.uploaded_at is not None: + location_dict["uploaded_at"] = loc.uploaded_at + if loc.confirmed_unavailable_at is not None: + location_dict["confirmed_unavailable_at"] = loc.confirmed_unavailable_at + if loc.check is not None: + location_dict["check"] = { + "status": loc.check.status, + "result": loc.check.result, + "detail": loc.check.detail, + } + locations.append(location_dict) + out: dict[str, Any] = { + "post_id": item.post_id, + "media_id": item.media_id, + "locations": locations, + } + if item.kind is not None: + out["kind"] = item.kind + if item.handle is not None: + out["handle"] = item.handle + if item.role is not None: + out["role"] = item.role + if item.embed is not None: + out["embed"] = item.embed + if item.caption is not None: + out["caption"] = item.caption + if item.derived_from is not None: + out["derived_from"] = { + "post_id": item.derived_from.post_id, + "media_id": item.derived_from.media_id, + "role": item.derived_from.role, + } + if item.publication is not None: + out["publication"] = { + "selected_location_id": item.publication.selected_location_id, + "selected_at": item.publication.selected_at, + "reason": item.publication.reason, + } + return out + + +def _manifest_to_wire(manifest: MediaManifest) -> dict[str, Any]: + """Convert a typed MediaManifest back into a wire-shape dict for JSON serialization. + + Schema version, root_post_id, items, captured_at are copied through; + mirrors are reset to [] (per the original canonical_manifest_bytes contract). + """ + out: dict[str, Any] = { + "schema_version": SCHEMA_VERSION, + "root_post_id": manifest.root_post_id, + "items": [_item_to_wire(item) for item in manifest.items], + "mirrors": [], + } + if manifest.captured_at is not None: + out["captured_at"] = manifest.captured_at + return out + + +def canonical_manifest_bytes(manifest: MediaManifest) -> bytes: + stable = _manifest_to_wire(manifest) return ( json.dumps(stable, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + "\n" ).encode("utf-8") -def payload_inventory(asset_dir: Path, manifest: dict[str, Any]) -> dict[str, str]: +def payload_inventory(asset_dir: Path, manifest: MediaManifest) -> dict[str, str]: inventory: dict[str, str] = { "media.json#canonical": sha256(canonical_manifest_bytes(manifest)).hexdigest() } @@ -299,14 +440,14 @@ def inventory_digest(inventory: dict[str, str]) -> str: return sha256(encoded).hexdigest() -def validate_manifest(data: dict[str, Any]) -> list[str]: +def validate_manifest(manifest: MediaManifest) -> list[str]: issues: list[str] = [] - if data.get("schema_version") != SCHEMA_VERSION: + if manifest.schema_version != SCHEMA_VERSION: issues.append(f"schema_version must be {SCHEMA_VERSION}") - if not str(data.get("root_post_id") or ""): + if not manifest.root_post_id: issues.append("root_post_id is empty") seen: set[tuple[str, str, str]] = set() - for index, item in enumerate(data.get("items") or []): + for index, item in enumerate(manifest.items): key = item_key(item) if not all(key): issues.append(f"items[{index}] has incomplete key {key!r}") @@ -314,24 +455,30 @@ def validate_manifest(data: dict[str, Any]) -> list[str]: issues.append(f"items[{index}] duplicates key {key!r}") seen.add(key) location_ids: set[str] = set() - for location in item.get("locations") or []: - location_id = str(location.get("id") or "") + for location in item.locations: + location_id = location.id or "" if not location_id: issues.append(f"items[{index}] has empty location id") elif location_id in location_ids: issues.append(f"items[{index}] duplicates location {location_id}") location_ids.add(location_id) - publication = item.get("publication") + publication = item.publication if publication is None: - if item.get("role") in VISUAL_ROLES and item.get("embed"): + role = item.kind or item.role or "" + if role in VISUAL_ROLES and item.embed: issues.append(f"items[{index}] embeds without publication") continue - selected_id = str(publication.get("selected_location_id") or "") - location = find_location(item, selected_id) + location = find_location(item, publication.selected_location_id) if location is None: - issues.append(f"items[{index}] selects missing location {selected_id}") - elif location.get("kind") not in {"origin", "fallback"}: - issues.append(f"items[{index}] selects non-HTTPS location {selected_id}") - elif not str(location.get("url") or "").startswith("https://"): - issues.append(f"items[{index}] selects non-HTTPS location {selected_id}") - return issues \ No newline at end of file + issues.append( + f"items[{index}] selects missing location {publication.selected_location_id}" + ) + elif location.kind not in {"origin", "fallback"}: + issues.append( + f"items[{index}] selects non-HTTPS location {publication.selected_location_id}" + ) + elif location.url is None or not location.url.startswith("https://"): + issues.append( + f"items[{index}] selects non-HTTPS location {publication.selected_location_id}" + ) + return issues diff --git a/scripts/twitter/media_merge.py b/scripts/twitter/media_merge.py index 7766b55..97d936b 100644 --- a/scripts/twitter/media_merge.py +++ b/scripts/twitter/media_merge.py @@ -13,7 +13,22 @@ import sys from datetime import datetime, timezone from pathlib import Path -ROLE_SUFFIX = { +try: + from .media_manifest import ( + _from_wire_dict, + _manifest_to_wire, + atomic_write_json, + upsert_derived_item, + ) +except ImportError: # pragma: no cover - script-mode import + from media_manifest import ( + _from_wire_dict, + _manifest_to_wire, + atomic_write_json, + upsert_derived_item, + ) + +ROLE_SUFFIX: dict[str, str] = { "_crt.png": "crt", "_crt_outline.png": "crt_outline", "_denoise.png": "denoise", @@ -22,6 +37,7 @@ ROLE_SUFFIX = { def role_of(name: str) -> str | None: + """Map a filename suffix (``_crt.png``, ``_ocr.txt``, ...) to a media role label.""" for suffix, role in ROLE_SUFFIX.items(): if name.endswith(suffix): return role @@ -29,6 +45,8 @@ def role_of(name: str) -> str | None: def parse_ids(name: str) -> tuple[str, str]: + """Extract ``(post_id, media_id)`` from a derived filename; returns ``("", stem)`` on mismatch.""" + stem = name stem = name for suffix in ROLE_SUFFIX: if stem.endswith(suffix): @@ -52,15 +70,12 @@ def main(argv: list[str] | None = None) -> int: raise SystemExit( "legacy media.json requires: tw.py migrate-media --id --apply" ) - try: - from media_manifest import atomic_write_json, upsert_derived_item - except ImportError: # pragma: no cover - script-mode import - from .media_manifest import atomic_write_json, upsert_derived_item now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") - items = list(data.get("items") or []) - have = {(str(i.get("filename") or ""), str(i.get("role") or "")) for i in items} - handles = {str(i.get("media_id") or ""): str(i.get("handle") or "") for i in items} + items_raw = list(data.get("items") or []) + have = {(str(i.get("filename") or ""), str(i.get("role") or "")) for i in items_raw} + handles = {str(i.get("media_id") or ""): str(i.get("handle") or "") for i in items_raw} + manifest = _from_wire_dict(data) added = 0 for path in sorted(args.thread.iterdir()): role = role_of(path.name) @@ -70,8 +85,8 @@ def main(argv: list[str] | None = None) -> int: continue post_id, media_id = parse_ids(path.name) handle = handles.get(media_id, "") - upsert_derived_item( - data, + manifest, _ = upsert_derived_item( + manifest, post_id=post_id, media_id=media_id, handle=handle, @@ -82,7 +97,9 @@ def main(argv: list[str] | None = None) -> int: ) added += 1 print(f"restore {path.name} role={role}") - atomic_write_json(media_path, data) + wire = _manifest_to_wire(manifest) + wire["mirrors"] = list(data.get("mirrors") or []) + atomic_write_json(media_path, wire) print(f"added {added} -> {media_path}") return 0 diff --git a/scripts/twitter/media_migrate.py b/scripts/twitter/media_migrate.py index b01319d..57e0659 100644 --- a/scripts/twitter/media_migrate.py +++ b/scripts/twitter/media_migrate.py @@ -1,3 +1,4 @@ +"""Migrate per-thread media.json from legacy wire shape to canonical v2.""" from __future__ import annotations import copy @@ -5,41 +6,57 @@ import json import sys from dataclasses import dataclass from pathlib import Path -from typing import Any 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, ...] @@ -51,49 +68,48 @@ def media_id_from_url(url: str, index: int) -> str: return segment if segment else f"m{index}" -def source_urls(thread_data: dict[str, Any]) -> dict[tuple[str, str], str]: +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.get("posts") or []: - post_id = str(post.get("post_id") or "") - for index, url in enumerate(post.get("media_urls") or [], start=1): - key = (post_id, media_id_from_url(str(url), index)) + 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] = str(url) + result[key] = url return result def reference_maps( - legacy: dict[str, Any], canonical: dict[str, Any] + legacy: LegacyMediaJson, canonical: MediaManifest ) -> tuple[dict[str, str], dict[str, list[str]]]: - canonical_by_key = { - ( - str(item.get("post_id") or ""), - str(item.get("media_id") or ""), - str(item.get("role") or ""), - ): item - for item in canonical.get("items") or [] - } + """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.get("items") or []: - if old.get("role") != "orig": + for old in legacy.items: + if (old.role or old.kind or "") != "orig": continue - key = ( - str(old.get("post_id") or ""), - str(old.get("media_id") or ""), - "orig", - ) + key = (old.post_id, old.media_id, "orig") item = canonical_by_key[key] - origin = next( - location - for location in item["locations"] - if location.get("kind") == "origin" - )["url"] - filename_origins[str(old.get("filename") or "")] = origin - old_url = str(old.get("url") or "") + 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) + fallback_origins.setdefault(old_url, []).append(origin.url or "") return filename_origins, fallback_origins @@ -103,7 +119,13 @@ def corpus_inventory( frozen_ids: set[str], *, now: str, -) -> dict[str, object]: +) -> 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, @@ -114,21 +136,17 @@ def corpus_inventory( "ambiguous_note_reference": 0, "invalid_legacy_row": 0, } - threads: list[dict[str, object]] = [] + 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 = json.loads(media_path.read_text(encoding="utf-8")) - is_legacy = legacy.get("schema_version") != SCHEMA_VERSION - legacy_items = list(legacy.get("items") or []) - unique_legacy = { - ( - str(item.get("post_id") or ""), - str(item.get("media_id") or ""), - str(item.get("role") or ""), - ) + 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) @@ -158,9 +176,11 @@ def corpus_inventory( "issues": ["missing thread_data.json"], }) continue - thread_data = json.loads(thread_data_path.read_text(encoding="utf-8")) + thread_data = LegacyThreadData.from_dict( + json.loads(thread_data_path.read_text(encoding="utf-8")) + ) try: - canonical, issues = canonical_from_legacy(legacy, thread_data, asset_dir, now) + 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({ @@ -179,11 +199,11 @@ def corpus_inventory( "issues": list(issues), }) continue - counts["migratable"] += len(canonical["items"]) - thread_entry: dict[str, object] = { + counts["migratable"] += len(canonical_manifest.items) + thread_entry: dict = { "relative": relative, "state": "migratable" if is_legacy else "current", - "item_count": len(canonical["items"]), + "item_count": len(canonical_manifest.items), } threads.append(thread_entry) expected_total = ( @@ -206,20 +226,28 @@ def corpus_inventory( def canonical_from_legacy( - legacy: dict[str, Any], - thread_data: dict[str, Any], + legacy: LegacyMediaJson, + thread_data: LegacyThreadData, asset_dir: Path, now: str, -) -> tuple[dict[str, Any], tuple[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[dict[str, Any]] = [] + items: list[MediaItem] = [] issues: list[str] = [] seen: set[tuple[str, str, str]] = set() - for old in legacy.get("items") or []: - role = str(old.get("role") or "") - post_id = str(old.get("post_id") or "") - media_id = str(old.get("media_id") or "") - filename = str(old.get("filename") or "") + 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 @@ -232,48 +260,79 @@ def canonical_from_legacy( item = new_original_item( post_id=post_id, media_id=media_id, - handle=str(old.get("handle") or ""), + handle=handle, origin_url=origin_url, filename=filename, local_path=asset_dir / filename, now=now, ) - old_url = str(old.get("url") or "") if old_url.startswith("https://"): local = find_location(item, "local") - item["locations"].append( - new_fallback_location( - provider="catbox", - url=old_url, - content_hash=str(local.get("sha256") or "") or None, - recorded_at=now, - uploaded_at=None, - ) + 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(legacy=old, asset_dir=asset_dir, now=now) - old_url = str(old.get("url") or "") + 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") - item["locations"].append( - new_fallback_location( - provider="catbox", - url=old_url, - content_hash=str(local.get("sha256") or "") or None, - recorded_at=now, - uploaded_at=None, - ) + 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 = { - "schema_version": SCHEMA_VERSION, - "root_post_id": str(legacy.get("root_post_id") or ""), - "items": items, - "mirrors": copy.deepcopy(legacy.get("mirrors") or []), - } + 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) @@ -285,14 +344,24 @@ def migrate_legacy_thread( 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 = json.loads(media_path.read_text(encoding="utf-8")) - if legacy.get("schema_version") == SCHEMA_VERSION: - return MigrationResult("current", False, (), len(legacy.get("items") or [])) - thread_data = json.loads((asset_dir / "thread_data.json").read_text(encoding="utf-8")) + 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.get("items") or [])) + return MigrationResult("blocked", False, issues, len(legacy.items)) if apply: filename_origins, fallback_origins = reference_maps(legacy, canonical) plan = plan_thread_rewrites( @@ -301,15 +370,41 @@ def migrate_legacy_thread( fallback_origins=fallback_origins, ) if plan.issues: - return MigrationResult("blocked", False, plan.issues, len(legacy.get("items") or [])) + 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, canonical) + 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.get("items") or [])) \ No newline at end of file + 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 diff --git a/scripts/twitter/media_refs.py b/scripts/twitter/media_refs.py index 4f10936..8a95875 100644 --- a/scripts/twitter/media_refs.py +++ b/scripts/twitter/media_refs.py @@ -1,3 +1,4 @@ +"""Helpers for rewriting archive note media references to origin URLs.""" from __future__ import annotations import os @@ -7,13 +8,14 @@ from dataclasses import dataclass from pathlib import Path -MEDIA_LINE = re.compile(r"(?m)^Media \(not lifted\):\s*(?P.+?)\s*$") -BACKTICK = re.compile(r"`([^`]+)`") -IMAGE_LINE = re.compile(r"(?m)^!\[[^\]]*\]\((?Phttps://[^)]+)\)\s*$") +MEDIA_LINE: re.Pattern[str] = re.compile(r"(?m)^Media \(not lifted\):\s*(?P.+?)\s*$") +BACKTICK: re.Pattern[str] = re.compile(r"`([^`]+)`") +IMAGE_LINE: re.Pattern[str] = re.compile(r"(?m)^!\[[^\]]*\]\((?Phttps://[^)]+)\)\s*$") @dataclass(frozen=True) class FileRewrite: + """A single note-file before/after text pair queued for rewrite.""" path: Path before: str after: str @@ -21,6 +23,7 @@ class FileRewrite: @dataclass(frozen=True) class RewritePlan: + """A batch of FileRewrite entries plus any blocking issues found during planning.""" files: tuple[FileRewrite, ...] issues: tuple[str, ...] diff --git a/scripts/twitter/models.py b/scripts/twitter/models.py index 16bec5c..9da2622 100644 --- a/scripts/twitter/models.py +++ b/scripts/twitter/models.py @@ -1,3 +1,4 @@ +"""Typed data model for the Twitter archive pipeline.""" from __future__ import annotations from dataclasses import dataclass @@ -37,67 +38,192 @@ class ThreadData: source_url: str +@dataclass(frozen=True) +class MediaLocationCheck: + """Origin/fallback URL HEAD probe outcome stored on a location.""" + status: int | None + result: str + detail: str + + @dataclass(frozen=True) class MediaLocation: - """A single media location record (local / remote / fallback / mirror / derived).""" + """A single media location record (local / origin / fallback / mirror / derived).""" kind: str + id: str | None = None local_path: Path | None = None url: str | None = None bytes: int | None = None sha256: str | None = None + media_type: str | None = None + integrity: str | None = None + verified_at: str | None = None + provider: str | None = None + availability: str | None = None checked_at: str | None = None checked_status: int | None = None + recorded_at: str | None = None + uploaded_at: str | None = None + confirmed_unavailable_at: str | None = None + check: MediaLocationCheck | None = None @classmethod def from_dict(cls, raw: dict[str, Any]) -> Self: if "kind" not in raw: raise KeyError("MediaLocation.from_dict: 'kind' is required") + # Wire uses 'path' (basename) or 'local_path' (Path string) interchangeably + path_raw = raw.get("local_path") + if path_raw is None: + path_raw = raw.get("path") + check_raw = raw.get("check") + check: MediaLocationCheck | None = None + if check_raw is not None: + if not isinstance(check_raw, dict): + raise TypeError( + f"check: expected dict, got {type(check_raw).__name__}" + ) + if "result" not in check_raw or "detail" not in check_raw: + raise KeyError("MediaLocation.from_dict: 'check' requires 'result' and 'detail'") + check = MediaLocationCheck( + status=_optional_int(check_raw.get("status"), "check.status"), + result=_required_str(check_raw.get("result"), "check.result"), + detail=_required_str(check_raw.get("detail"), "check.detail"), + ) return cls( kind=_required_str(raw["kind"], "kind"), - local_path=_optional_path(raw.get("local_path"), "local_path"), + id=_optional_str(raw.get("id"), "id"), + local_path=_optional_path(path_raw, "local_path"), url=_optional_str(raw.get("url"), "url"), bytes=_optional_int(raw.get("bytes"), "bytes"), sha256=_optional_str(raw.get("sha256"), "sha256"), + media_type=_optional_str(raw.get("media_type"), "media_type"), + integrity=_optional_str(raw.get("integrity"), "integrity"), + verified_at=_optional_str(raw.get("verified_at"), "verified_at"), + provider=_optional_str(raw.get("provider"), "provider"), + availability=_optional_str(raw.get("availability"), "availability"), checked_at=_optional_str(raw.get("checked_at"), "checked_at"), checked_status=_optional_int(raw.get("checked_status"), "checked_status"), + recorded_at=_optional_str(raw.get("recorded_at"), "recorded_at"), + uploaded_at=_optional_str(raw.get("uploaded_at"), "uploaded_at"), + confirmed_unavailable_at=_optional_str( + raw.get("confirmed_unavailable_at"), "confirmed_unavailable_at" + ), + check=check, ) @dataclass(frozen=True) -class MediaItem: - """A single media entry attached to a post (image / crt / crt_outline / denoise / ocr).""" +class DerivedFromRef: + """Pointer to the parent (orig) media item for derived items (crt, ocr, etc.).""" post_id: str media_id: str - kind: str - locations: tuple[MediaLocation, ...] + role: str + + +@dataclass(frozen=True) +class Publication: + """Selected location for a media item; embedded as 'publication' on the wire.""" + selected_location_id: str + selected_at: str + reason: str + + +@dataclass(frozen=True) +class MediaItem: + """A single media entry attached to a post (orig / crt / crt_outline / denoise / ocr). + + For legacy (pre-v2) wire shapes, ``locations`` may be empty and ``filename`` + plus ``url`` carry the original on-disk filename and pre-migration fallback + URL. v2 items leave both ``None`` and put the URL inside a ``MediaLocation``. + """ + post_id: str + media_id: str + locations: tuple[MediaLocation, ...] = () + kind: str | None = None + handle: str | None = None + role: str | None = None + filename: str | None = None + url: str | None = None embed: bool | None = None caption: str | None = None + derived_from: DerivedFromRef | None = None + publication: Publication | None = None @classmethod def from_dict(cls, raw: dict[str, Any]) -> Self: - for required in ("post_id", "media_id", "kind", "locations"): + for required in ("post_id", "media_id"): if required not in raw: raise KeyError(f"MediaItem.from_dict: '{required}' is required") - locations_raw = raw["locations"] - if not isinstance(locations_raw, list): - raise TypeError( - f"MediaItem.from_dict: 'locations' must be list, got {type(locations_raw).__name__}" + locations_raw = raw.get("locations") + locations: tuple[MediaLocation, ...] = () + if locations_raw is not None: + if not isinstance(locations_raw, list): + raise TypeError( + f"MediaItem.from_dict: 'locations' must be list, got {type(locations_raw).__name__}" + ) + locations = tuple(MediaLocation.from_dict(loc) for loc in locations_raw) + # Wire uses 'role' as the kind identifier; accept 'kind' for direct callers + kind_raw = raw.get("role") + if kind_raw is None: + kind_raw = raw.get("kind") + if kind_raw is None: + raise KeyError("MediaItem.from_dict: 'role' or 'kind' is required") + kind = _required_str(kind_raw, "role") + derived_from_raw = raw.get("derived_from") + derived_from: DerivedFromRef | None = None + if derived_from_raw is not None: + if not isinstance(derived_from_raw, dict): + raise TypeError( + f"derived_from: expected dict, got {type(derived_from_raw).__name__}" + ) + derived_from = DerivedFromRef( + post_id=_required_str( + derived_from_raw.get("post_id"), "derived_from.post_id" + ), + media_id=_required_str( + derived_from_raw.get("media_id"), "derived_from.media_id" + ), + role=_required_str(derived_from_raw.get("role"), "derived_from.role"), + ) + publication_raw = raw.get("publication") + publication: Publication | None = None + if publication_raw is not None: + if not isinstance(publication_raw, dict): + raise TypeError( + f"publication: expected dict, got {type(publication_raw).__name__}" + ) + publication = Publication( + selected_location_id=_required_str( + publication_raw.get("selected_location_id"), + "publication.selected_location_id", + ), + selected_at=_required_str( + publication_raw.get("selected_at"), "publication.selected_at" + ), + reason=_required_str(publication_raw.get("reason"), "publication.reason"), ) return cls( post_id=_required_str(raw["post_id"], "post_id"), media_id=_required_str(raw["media_id"], "media_id"), - kind=_required_str(raw["kind"], "kind"), - locations=tuple(MediaLocation.from_dict(loc) for loc in locations_raw), + kind=kind, + handle=_optional_str(raw.get("handle"), "handle"), + role=_optional_str(raw.get("role"), "role"), + filename=_optional_str(raw.get("filename"), "filename"), + url=_optional_str(raw.get("url"), "url"), + locations=locations, embed=_optional_bool(raw.get("embed"), "embed"), caption=_optional_str(raw.get("caption"), "caption"), + derived_from=derived_from, + publication=publication, ) @dataclass(frozen=True) class MediaManifest: - """A whole media.json document (one thread).""" + """A whole media.json document (one thread). schema_version is the wire key.""" root_post_id: str items: tuple[MediaItem, ...] + schema_version: int | None = None captured_at: str | None = None @classmethod @@ -113,6 +239,7 @@ class MediaManifest: return cls( root_post_id=_required_str(raw["root_post_id"], "root_post_id"), items=tuple(MediaItem.from_dict(item) for item in items_raw), + schema_version=_optional_int(raw.get("schema_version"), "schema_version"), captured_at=_optional_str(raw.get("captured_at"), "captured_at"), ) @@ -157,6 +284,8 @@ class LegacyManifest: threads: tuple[LegacyThreadData, ...] schema_version: int | None = None generated_at: str | None = None + counts: dict[str, int] | None = None + frozen_hashes: dict[str, str] | None = None @classmethod def from_dict(cls, raw: dict[str, Any]) -> Self: @@ -167,13 +296,132 @@ class LegacyManifest: raise TypeError( f"LegacyManifest.from_dict: 'threads' must be list, got {type(threads_raw).__name__}" ) + counts: dict[str, int] | None = _coerce_str_int_map(raw.get("counts"), "counts") + frozen_hashes: dict[str, str] | None = _coerce_str_str_map( + raw.get("frozen_hashes"), "frozen_hashes" + ) return cls( threads=tuple(LegacyThreadData.from_dict(t) for t in threads_raw), schema_version=_optional_int(raw.get("schema_version"), "schema_version"), generated_at=_optional_str(raw.get("generated_at"), "generated_at"), + counts=counts, + frozen_hashes=frozen_hashes, ) +@dataclass(frozen=True) +class OriginCheck: + """Outcome of a HEAD probe against a media item's origin URL. + + ``status`` is the HTTP status code returned by the probe (or ``None`` if + the request errored before a response arrived). ``result`` is the + classifier label (``"available"`` for 2xx/3xx, ``"error"`` otherwise). + ``confirms_unavailable`` distinguishes transient failures like 429 from + definitive unavailability so the caller does not flip the location's + availability state on a rate-limit blip. + """ + checked_at: str + status: int | None + result: str + detail: str + confirms_unavailable: bool + + +@dataclass(frozen=True) +class LegacyMediaJson: + """Per-thread ``media.json`` shape consumed by ``media_migrate.py`` and ``media_audit.py``. + + Holds both the pre-v2 (legacy) and v2 (canonical) wire shapes so the + migration and audit entry points can share one boundary type. The + ``items`` field is parsed as ``MediaItem`` instances; for legacy items + ``locations`` is empty and ``filename`` plus ``url`` carry the + on-disk filename and pre-migration fallback URL. + + ``mirrors`` hold opaque destination/auth fields that are intentionally + not yet typed. This is the only non-typed dict in the data model — it + is consumed only by iteration, never by attribute access, and we + deliberately do not commit to a structure until the backup subsystem + settles on a wire shape. + """ + schema_version: int | None + root_post_id: str + items: tuple[MediaItem, ...] + captured_at: str | None + mirrors: dict[str, Any] | None = None + frozen: bool | None = None + + @classmethod + def from_dict(cls, raw: dict[str, Any]) -> Self: + if "root_post_id" not in raw: + raise KeyError("LegacyMediaJson.from_dict: 'root_post_id' is required") + items_raw = raw.get("items") + if items_raw is None: + items_raw = [] + if not isinstance(items_raw, list): + raise TypeError( + f"LegacyMediaJson.from_dict: 'items' must be list, got {type(items_raw).__name__}" + ) + mirrors = _coerce_mirrors(raw.get("mirrors"), "mirrors") + return cls( + schema_version=_optional_int(raw.get("schema_version"), "schema_version"), + root_post_id=_required_str(raw["root_post_id"], "root_post_id"), + items=tuple(MediaItem.from_dict(item) for item in items_raw), + captured_at=_optional_str(raw.get("captured_at"), "captured_at"), + mirrors=mirrors, + frozen=_optional_bool(raw.get("frozen"), "frozen"), + ) + + +def _coerce_str_int_map(value: object, field: str) -> dict[str, int] | None: + if value is None: + return None + if not isinstance(value, dict): + raise TypeError(f"{field}: expected dict, got {type(value).__name__}") + out: dict[str, int] = {} + for key, item in value.items(): + if isinstance(item, bool) or not isinstance(item, int): + raise TypeError(f"{field}[{key}]: expected int") + out[str(key)] = item + return out + + +def _coerce_str_str_map(value: object, field: str) -> dict[str, str] | None: + if value is None: + return None + if not isinstance(value, dict): + raise TypeError(f"{field}: expected dict, got {type(value).__name__}") + out: dict[str, str] = {} + for key, item in value.items(): + if not isinstance(item, str): + raise TypeError(f"{field}[{key}]: expected str") + out[str(key)] = item + return out + + +def _coerce_mirrors(value: object, field: str) -> dict[str, Any] | None: + """Accept the legacy ``mirrors: list[dict]`` wire and coerce to the typed mirrors shape. + + Legacy producers (pre-LegacyMediaJson) wrote ``mirrors`` as a list of + destination entries; current producers may write it as a dict keyed by + destination id. Both shapes are accepted; the returned dict is keyed + by ``destination_id`` when present on each entry, falling back to the + string index for entries without one. + """ + if value is None: + return None + if isinstance(value, dict): + return value + if isinstance(value, list): + coerced: dict = {} + for index, entry in enumerate(value): + if isinstance(entry, dict) and "destination_id" in entry: + coerced[str(entry["destination_id"])] = entry + else: + coerced[str(index)] = entry + return coerced + raise TypeError(f"{field}: expected dict or list, got {type(value).__name__}") + + def _norm_handle(raw: str) -> str: h = (raw or "").strip() return h[1:] if h.startswith("@") else h diff --git a/scripts/twitter/ocr_pass.py b/scripts/twitter/ocr_pass.py index 5b31e41..d0a3f44 100644 --- a/scripts/twitter/ocr_pass.py +++ b/scripts/twitter/ocr_pass.py @@ -21,18 +21,36 @@ import urllib.request from datetime import datetime, timezone from pathlib import Path -UMI_URL = "http://127.0.0.1:1224/api/ocr" +try: + from .media_manifest import ( + _from_wire_dict, + _manifest_to_wire, + atomic_write_json, + upsert_derived_item, + ) +except ImportError: # pragma: no cover - script-mode import + from media_manifest import ( + _from_wire_dict, + _manifest_to_wire, + atomic_write_json, + upsert_derived_item, + ) + +UMI_URL: str = "http://127.0.0.1:1224/api/ocr" def find_umi() -> str | None: + """Locate the ``umi-ocr`` / ``Umi-OCR`` binary on PATH (or ``None``).""" return shutil.which("umi-ocr") or shutil.which("Umi-OCR") def find_tesseract() -> str | None: + """Locate the ``tesseract`` binary on PATH (or ``None``).""" return shutil.which("tesseract") def umi_http_up() -> bool: + """Return True if Umi-OCR's HTTP server is reachable on 127.0.0.1:1224.""" try: urllib.request.urlopen("http://127.0.0.1:1224/api/ocr/get_options", timeout=2) return True @@ -41,6 +59,7 @@ def umi_http_up() -> bool: def run_umi(src: Path, dest: Path) -> None: + """OCR ``src`` via Umi-OCR's HTTP API and write UTF-8 text to ``dest``.""" dest.parent.mkdir(parents=True, exist_ok=True) if not umi_http_up(): exe = find_umi() @@ -82,6 +101,8 @@ def run_umi(src: Path, dest: Path) -> None: def run_tesseract(src: Path, dest: Path) -> None: + """OCR ``src`` via the local ``tesseract`` binary and write UTF-8 text to ``dest``.""" + exe = find_tesseract() exe = find_tesseract() if not exe: raise SystemExit("tesseract not on PATH. scoop install tesseract") @@ -96,6 +117,7 @@ def run_tesseract(src: Path, dest: Path) -> None: def run_windows_ocr(src: Path, dest: Path) -> None: + """OCR ``src`` via the Windows.Media.Ocr PowerShell shim and write UTF-8 text to ``dest``.""" ps = r""" $ErrorActionPreference = 'Stop' Add-Type -AssemblyName System.Runtime.WindowsRuntime @@ -135,10 +157,7 @@ $result = Await ($engine.RecognizeAsync($bitmap)) def append_media_row(media_json: Path, src: Path, dest: Path) -> None: - try: - from media_manifest import atomic_write_json, upsert_derived_item - except ImportError: # pragma: no cover - script-mode import - from .media_manifest import atomic_write_json, upsert_derived_item + """Append or update the ``role=ocr`` row in ``media_json`` for the (src -> dest) file pair.""" data = json.loads(media_json.read_text(encoding="utf-8")) if data.get("schema_version") != 2: raise SystemExit( @@ -158,8 +177,9 @@ def append_media_row(media_json: Path, src: Path, dest: Path) -> None: post_id = parts[0] media_id = parts[1] if len(parts) > 1 else "" asset_dir = media_json.parent - upsert_derived_item( - data, + manifest = _from_wire_dict(data) + manifest, _ = upsert_derived_item( + manifest, post_id=post_id, media_id=media_id, handle=handle, @@ -168,7 +188,9 @@ def append_media_row(media_json: Path, src: Path, dest: Path) -> None: asset_dir=asset_dir, now=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), ) - atomic_write_json(media_json, data) + wire = _manifest_to_wire(manifest) + wire["mirrors"] = list(data.get("mirrors") or []) + atomic_write_json(media_json, wire) print(f"updated {media_json}") diff --git a/scripts/twitter/paths.py b/scripts/twitter/paths.py index 8841e40..fd6fcb3 100644 --- a/scripts/twitter/paths.py +++ b/scripts/twitter/paths.py @@ -3,11 +3,11 @@ from __future__ import annotations from pathlib import Path -HERE = Path(__file__).resolve().parent -VAULT = HERE.parent.parent -PROJECTS = VAULT.parent -DUMPS = PROJECTS / "manual_slop" / "docs" / "twitter" -WORKSPACE = PROJECTS / "Threadwell-ai" -SCRATCH = WORKSPACE / "scratch" -COOKIES = VAULT / "secrets" / "twitter_cookies.txt" -FROZEN = HERE / "do_not_refetch.txt" +HERE: Path = Path(__file__).resolve().parent +VAULT: Path = HERE.parent.parent +PROJECTS: Path = VAULT.parent +DUMPS: Path = PROJECTS / "manual_slop" / "docs" / "twitter" +WORKSPACE: Path = PROJECTS / "Threadwell-ai" +SCRATCH: Path = WORKSPACE / "scratch" +COOKIES: Path = VAULT / "secrets" / "twitter_cookies.txt" +FROZEN: Path = HERE / "do_not_refetch.txt" diff --git a/scripts/twitter/render.py b/scripts/twitter/render.py index 95daf99..ad03770 100644 --- a/scripts/twitter/render.py +++ b/scripts/twitter/render.py @@ -30,6 +30,7 @@ def _first_line(text: str) -> str: def _title_text(text: str) -> str: + """Derive a title from the first line, truncating at the first sentence end (``. ``).""" line = _first_line(text) idx = line.find(". ") if idx >= 0: @@ -59,6 +60,7 @@ def _frontmatter( in_reply_to: str, parent_post_id: str | None = None, ) -> str: + """Build the YAML frontmatter block for an archive note (always ``draft: true``).""" lines = [ "---", f"title: {_yaml_quote(title)}", @@ -85,6 +87,8 @@ def _frontmatter( def _source_block(post: PostData) -> str: + """Render the ``## Source`` block listing URL, author handle, and post timestamp.""" + url = _status_url(post.handle, post.post_id) url = _status_url(post.handle, post.post_id) return ( "## Source\n" @@ -101,6 +105,8 @@ def _post_block( media_by_post: dict[str, tuple[str, ...]] | None, branch_links: list[str] | None = None, ) -> str: + """Render one post body (``**N/**`` + text + media embeds + branch links).""" + parts = [f"**{n}/**", "", post.text] parts = [f"**{n}/**", "", post.text] media = (media_by_post or {}).get(post.post_id) or () if media: @@ -117,6 +123,8 @@ def _foreign_parent( first: PostData, ids: dict[str, PostData], ) -> tuple[str, PostData | None]: + """Return ``(in_reply_to_url, parent_post)`` for the first spine post, when the parent belongs to another author.""" + parent_id = first.reply_to_id or "" parent_id = first.reply_to_id or "" if not parent_id: return "", None diff --git a/scripts/twitter/slug.py b/scripts/twitter/slug.py index 1e080b2..25e17ce 100644 --- a/scripts/twitter/slug.py +++ b/scripts/twitter/slug.py @@ -1,9 +1,10 @@ +"""Slug helpers for archive dir and branch filenames.""" from __future__ import annotations import re -_MENTION = re.compile(r"^(@[A-Za-z0-9_]+(\s+|$))+") -_APOS = re.compile(r"['\u2019\u2018]") -_NON_SLUG = re.compile(r"[^a-z0-9]+") +_MENTION: re.Pattern[str] = re.compile(r"^(@[A-Za-z0-9_]+(\s+|$))+") +_APOS: re.Pattern[str] = re.compile(r"['\u2019\u2018]") +_NON_SLUG: re.Pattern[str] = re.compile(r"[^a-z0-9]+") def date_prefix(timestamp: str) -> str: if len(timestamp) >= 10 and timestamp[4] == "-" and timestamp[7] == "-": diff --git a/scripts/twitter/tests/__init__.py b/scripts/twitter/tests/__init__.py index ff2b71a..72996b7 100644 --- a/scripts/twitter/tests/__init__.py +++ b/scripts/twitter/tests/__init__.py @@ -1 +1,2 @@ -"""Isolated tests for Threadwell Twitter tooling.""" \ No newline at end of file +"""Isolated tests for Threadwell Twitter tooling.""" +from __future__ import annotations \ No newline at end of file diff --git a/scripts/twitter/tests/test_backup_assets.py b/scripts/twitter/tests/test_backup_assets.py index 1107b22..306b708 100644 --- a/scripts/twitter/tests/test_backup_assets.py +++ b/scripts/twitter/tests/test_backup_assets.py @@ -7,6 +7,7 @@ from pathlib import Path from twitter.backup_assets import backup_thread from twitter.media_audit import audit_thread from twitter.media_manifest import ( + _from_wire_dict, canonical_manifest_bytes, hash_file, inventory_digest, @@ -74,14 +75,15 @@ class BackupTests(unittest.TestCase): now=NOW, ) manifest = read_json(asset_dir / "media.json") - inventory = payload_inventory(asset_dir, manifest) + typed_manifest = _from_wire_dict(manifest) + inventory = payload_inventory(asset_dir, typed_manifest) self.assertEqual( manifest["mirrors"][0]["inventory_digest"], inventory_digest(inventory), ) - backup_inventory_before = inventory_digest(payload_inventory(asset_dir, manifest)) + backup_inventory_before = inventory_digest(payload_inventory(asset_dir, typed_manifest)) (asset_dir / "100_AAA_orig.png").write_bytes(b"modified") - new_inventory = payload_inventory(asset_dir, manifest) + new_inventory = payload_inventory(asset_dir, typed_manifest) self.assertNotEqual( inventory_digest(new_inventory), backup_inventory_before, diff --git a/scripts/twitter/tests/test_fallback_media.py b/scripts/twitter/tests/test_fallback_media.py index 32fc280..8847d78 100644 --- a/scripts/twitter/tests/test_fallback_media.py +++ b/scripts/twitter/tests/test_fallback_media.py @@ -5,7 +5,7 @@ import unittest from pathlib import Path from twitter.fallback_media import activate_fallback, restore_origin -from twitter.media_manifest import selected_url +from twitter.media_manifest import _from_wire_dict, selected_url from twitter.media_migrate import migrate_legacy_thread from twitter.tests.helpers import NOW, legacy_thread, read_json @@ -34,8 +34,8 @@ class FallbackTests(unittest.TestCase): ) self.assertEqual(calls, []) self.assertEqual(result.url, "https://files.catbox.moe/fallback.png") - manifest = read_json(asset_dir / "media.json") - self.assertEqual(selected_url(manifest["items"][0]), result.url) + manifest = _from_wire_dict(read_json(asset_dir / "media.json")) + self.assertEqual(selected_url(manifest.items[0]), result.url) def test_note_rewrite_failure_keeps_origin_selected(self) -> None: with tempfile.TemporaryDirectory() as raw: @@ -54,8 +54,10 @@ class FallbackTests(unittest.TestCase): now=NOW, upload=lambda _path: "https://files.catbox.moe/fallback.png", ) - manifest = read_json(asset_dir / "media.json") - self.assertTrue(selected_url(manifest["items"][0]).startswith("https://pbs.twimg.com/")) + manifest = _from_wire_dict(read_json(asset_dir / "media.json")) + self.assertTrue( + (selected_url(manifest.items[0]) or "").startswith("https://pbs.twimg.com/") + ) def test_restore_origin_retains_fallback(self) -> None: with tempfile.TemporaryDirectory() as raw: @@ -73,10 +75,13 @@ class FallbackTests(unittest.TestCase): upload=lambda _path: "https://files.catbox.moe/fallback.png", ) restore_origin(asset_dir, note_dir, media_id="AAA", now=NOW) - item = read_json(asset_dir / "media.json")["items"][0] - self.assertTrue(selected_url(item).startswith("https://pbs.twimg.com/")) - self.assertTrue(any(loc.get("kind") == "fallback" for loc in item["locations"])) + manifest = _from_wire_dict(read_json(asset_dir / "media.json")) + item = manifest.items[0] + self.assertTrue( + (selected_url(item) or "").startswith("https://pbs.twimg.com/") + ) + self.assertTrue(any(loc.kind == "fallback" for loc in item.locations)) if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/scripts/twitter/tests/test_media_audit.py b/scripts/twitter/tests/test_media_audit.py index 2724e89..66ad235 100644 --- a/scripts/twitter/tests/test_media_audit.py +++ b/scripts/twitter/tests/test_media_audit.py @@ -8,7 +8,9 @@ from twitter.media_audit import ( AuditReport, audit_local_item, audit_thread, + check_origin_url, classify_origin_response, + record_origin_check, ) from twitter.media_manifest import ( atomic_write_json, @@ -17,6 +19,12 @@ from twitter.media_manifest import ( new_original_item, ) from twitter.media_migrate import migrate_legacy_thread +from twitter.models import ( + LegacyMediaJson, + MediaItem, + MediaManifest, + OriginCheck, +) from twitter.tests.helpers import NOW, legacy_thread, read_json @@ -40,8 +48,8 @@ class MediaAuditTests(unittest.TestCase): def test_rate_limit_does_not_mean_unavailable(self) -> None: outcome = classify_origin_response(429, "rate limited", NOW) - self.assertEqual(outcome["result"], "error") - self.assertFalse(outcome["confirms_unavailable"]) + self.assertEqual(outcome.result, "error") + self.assertFalse(outcome.confirms_unavailable) def test_frozen_thread_is_reported(self) -> None: with tempfile.TemporaryDirectory() as raw: @@ -53,5 +61,179 @@ class MediaAuditTests(unittest.TestCase): self.assertTrue(any("frozen: skipped" in issue for issue in report.issues)) +class OriginCheckTests(unittest.TestCase): + """Tests for the OriginCheck boundary dataclass and its producers.""" + + def test_classify_available_2xx_marks_available(self) -> None: + outcome = classify_origin_response(200, "HEAD completed", NOW) + self.assertEqual(outcome.status, 200) + self.assertEqual(outcome.result, "available") + self.assertFalse(outcome.confirms_unavailable) + + def test_classify_truncates_detail_to_200_chars(self) -> None: + long_detail = "x" * 500 + outcome = classify_origin_response(500, long_detail, NOW) + self.assertEqual(len(outcome.detail), 200) + + def test_check_origin_url_uses_opener(self) -> None: + class _FakeResponse: + status = 200 + + def __enter__(self) -> "_FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + def fake_opener(*args: object, **kwargs: object) -> _FakeResponse: + return _FakeResponse() + + outcome = check_origin_url( + "https://example.com/x.png", + NOW, + opener=fake_opener, + ) + self.assertEqual(outcome.result, "available") + self.assertEqual(outcome.status, 200) + + +class RecordOriginCheckTests(unittest.TestCase): + """Tests for the record_origin_check round-trip helper.""" + + def test_record_check_updates_origin_x_location(self) -> None: + item = new_original_item( + post_id="100", + media_id="AAA", + handle="example", + origin_url="https://pbs.twimg.com/media/AAA", + filename="asset.png", + local_path=Path("asset.png"), + now=NOW, + ) + manifest = LegacyMediaJson.from_dict( + { + "schema_version": 2, + "root_post_id": "100", + "items": [ + { + "post_id": item.post_id, + "media_id": item.media_id, + "kind": item.kind, + "role": item.role, + "handle": item.handle, + "locations": [ + { + "kind": loc.kind, + "id": loc.id, + "url": loc.url, + "provider": loc.provider, + "availability": loc.availability, + } + for loc in item.locations + ], + "publication": { + "selected_location_id": "origin:x", + "selected_at": NOW, + "reason": "default", + }, + } + ], + } + ) + outcome = classify_origin_response(200, "HEAD completed", NOW) + updated = record_origin_check(manifest, 0, outcome) + updated_item: MediaItem = updated.items[0] + origin = find_location(updated_item, "origin:x") + self.assertIsNotNone(origin) + self.assertEqual(origin.checked_at, NOW) + self.assertEqual(origin.checked_status, 200) + self.assertEqual(origin.availability, "available") + self.assertIsNotNone(origin.check) + self.assertEqual(origin.check.result, "available") + + def test_record_check_out_of_range_raises(self) -> None: + manifest = LegacyMediaJson.from_dict( + {"root_post_id": "100", "items": []} + ) + outcome = classify_origin_response(200, "HEAD", NOW) + with self.assertRaises(ValueError): + record_origin_check(manifest, 0, outcome) + + +class LegacyMediaJsonTests(unittest.TestCase): + """Tests for the LegacyMediaJson boundary dataclass.""" + + def test_minimal_legacy_round_trip(self) -> None: + legacy = LegacyMediaJson.from_dict( + { + "root_post_id": "100", + "items": [ + { + "post_id": "100", + "media_id": "AAA", + "role": "orig", + "filename": "x.png", + "url": "https://files.catbox.moe/x.png", + } + ], + } + ) + self.assertEqual(legacy.root_post_id, "100") + self.assertIsNone(legacy.schema_version) + self.assertEqual(len(legacy.items), 1) + self.assertEqual(legacy.items[0].filename, "x.png") + self.assertEqual(legacy.items[0].url, "https://files.catbox.moe/x.png") + self.assertEqual(legacy.items[0].locations, ()) + self.assertIsNone(legacy.mirrors) + self.assertIsNone(legacy.frozen) + + def test_v2_with_mirrors_dict(self) -> None: + legacy = LegacyMediaJson.from_dict( + { + "schema_version": 2, + "root_post_id": "100", + "captured_at": NOW, + "frozen": True, + "mirrors": { + "dest1": {"destination_id": "dest1", "state": "synced"}, + }, + "items": [], + } + ) + self.assertEqual(legacy.schema_version, 2) + self.assertEqual(legacy.captured_at, NOW) + self.assertEqual(legacy.frozen, True) + self.assertIsNotNone(legacy.mirrors) + self.assertIn("dest1", legacy.mirrors) + + def test_mirrors_list_wire_is_coerced_to_dict(self) -> None: + legacy = LegacyMediaJson.from_dict( + { + "root_post_id": "100", + "items": [], + "mirrors": [ + {"destination_id": "alpha", "state": "synced"}, + {"destination_id": "beta", "state": "synced"}, + ], + } + ) + self.assertIsNotNone(legacy.mirrors) + self.assertEqual(set(legacy.mirrors), {"alpha", "beta"}) + + def test_missing_root_post_id_raises_keyerror(self) -> None: + with self.assertRaises(KeyError): + LegacyMediaJson.from_dict({"items": []}) + + def test_items_not_list_raises_typeerror(self) -> None: + with self.assertRaisesRegex(TypeError, "items"): + LegacyMediaJson.from_dict({"root_post_id": "100", "items": {}}) + + def test_mirrors_invalid_type_raises_typeerror(self) -> None: + with self.assertRaisesRegex(TypeError, "mirrors"): + LegacyMediaJson.from_dict( + {"root_post_id": "100", "items": [], "mirrors": "nope"} + ) + + if __name__ == "__main__": unittest.main() \ No newline at end of file diff --git a/scripts/twitter/tests/test_media_manifest.py b/scripts/twitter/tests/test_media_manifest.py index 63145f4..73c736e 100644 --- a/scripts/twitter/tests/test_media_manifest.py +++ b/scripts/twitter/tests/test_media_manifest.py @@ -8,14 +8,18 @@ from pathlib import Path from twitter.media_manifest import ( atomic_write_json, + canonical_manifest_bytes, find_location, hash_file, merge_item, merge_manifest_items, new_original_item, + payload_inventory, selected_url, validate_manifest, + _from_wire_dict, ) +from twitter.models import MediaItem, MediaLocation, MediaManifest, Publication from twitter.tests.helpers import NOW @@ -55,9 +59,9 @@ class CanonicalManifestTests(unittest.TestCase): local_path=local, now=NOW, ) - self.assertEqual(selected_url(item), item["locations"][0]["url"]) - self.assertEqual(find_location(item, "local")["integrity"], "present") - self.assertEqual(find_location(item, "local")["sha256"], hash_file(local)) + self.assertEqual(selected_url(item), item.locations[0].url) + self.assertEqual(find_location(item, "local").integrity, "present") + self.assertEqual(find_location(item, "local").sha256, hash_file(local)) def test_missing_local_is_explicit(self) -> None: with tempfile.TemporaryDirectory() as raw: @@ -71,8 +75,8 @@ class CanonicalManifestTests(unittest.TestCase): local_path=local, now=NOW, ) - self.assertEqual(find_location(item, "local")["integrity"], "missing") - self.assertIsNone(find_location(item, "local")["sha256"]) + self.assertEqual(find_location(item, "local").integrity, "missing") + self.assertIsNone(find_location(item, "local").sha256) def test_merge_preserves_fallback_and_publication(self) -> None: with tempfile.TemporaryDirectory() as raw: @@ -87,24 +91,30 @@ class CanonicalManifestTests(unittest.TestCase): local_path=local, now=NOW, ) - old["locations"].append( - { - "id": "fallback:catbox:fixture", - "kind": "fallback", - "provider": "catbox", - "url": "https://files.catbox.moe/fallback.png", - "sha256": find_location(old, "local")["sha256"], - "uploaded_at": NOW, - "availability": "unknown", - "checked_at": None, - "check": None, - } + local_sha = find_location(old, "local").sha256 + fallback = MediaLocation( + id="fallback:catbox:fixture", + kind="fallback", + provider="catbox", + url="https://files.catbox.moe/fallback.png", + sha256=local_sha, + uploaded_at=NOW, + availability="unknown", + ) + old_with_fallback = MediaItem( + post_id=old.post_id, + media_id=old.media_id, + kind=old.kind, + role=old.role, + handle=old.handle, + embed=old.embed, + locations=old.locations + (fallback,), + publication=Publication( + selected_location_id="fallback:catbox:fixture", + selected_at=NOW, + reason="origin-unavailable", + ), ) - old["publication"] = { - "selected_location_id": "fallback:catbox:fixture", - "selected_at": NOW, - "reason": "origin-unavailable", - } fresh = new_original_item( post_id="100", media_id="AAA", @@ -114,8 +124,10 @@ class CanonicalManifestTests(unittest.TestCase): local_path=local, now="2026-08-25T12:00:00Z", ) - merged = merge_item(old, fresh) - self.assertEqual(selected_url(merged), "https://files.catbox.moe/fallback.png") + merged = merge_item(old_with_fallback, fresh) + self.assertEqual( + selected_url(merged), "https://files.catbox.moe/fallback.png" + ) self.assertIsNotNone(find_location(merged, "fallback:catbox:fixture")) self.assertEqual(merge_item(merged, fresh), merged) @@ -145,53 +157,239 @@ class CanonicalManifestTests(unittest.TestCase): merge_item(old, fresh) def test_unmatched_derived_item_survives_manifest_merge(self) -> None: - derived = { - "post_id": "100", - "media_id": "AAA", - "handle": "example", - "role": "ocr", - "derived_from": {"post_id": "100", "media_id": "AAA", "role": "orig"}, - "embed": False, - "locations": [], - "publication": None, - } - self.assertEqual(merge_manifest_items([derived], []), [derived]) + derived = MediaItem( + post_id="100", + media_id="AAA", + kind="ocr", + role="ocr", + handle="example", + locations=(), + publication=None, + ) + result = merge_manifest_items((), (derived,)) + self.assertEqual(result, (derived,)) def test_validator_rejects_local_publication(self) -> None: - manifest = { + item = MediaItem( + post_id="100", + media_id="AAA", + kind="orig", + role="orig", + handle="example", + embed=True, + locations=( + MediaLocation( + id="local", + kind="local", + local_path=Path("100_AAA_orig.png"), + sha256=None, + bytes=None, + media_type="image/png", + integrity="missing", + verified_at=NOW, + ), + ), + publication=Publication( + selected_location_id="local", + selected_at=NOW, + reason="manual", + ), + ) + manifest = MediaManifest( + schema_version=2, + root_post_id="100", + items=(item,), + ) + self.assertIn( + "items[0] selects non-HTTPS location local", validate_manifest(manifest) + ) + + +class FromWireDictTests(unittest.TestCase): + """Tests for the _from_wire_dict boundary parser.""" + + def test_full_wire_payload_round_trips(self) -> None: + raw: dict = { + "schema_version": 2, + "root_post_id": "100", + "captured_at": NOW, + "items": [ + { + "post_id": "100", + "media_id": "AAA", + "role": "orig", + "embed": True, + "locations": [ + { + "id": "origin:x", + "kind": "origin", + "provider": "x", + "url": "https://pbs.twimg.com/media/AAA", + "availability": "unknown", + "checked_at": None, + }, + { + "id": "local", + "kind": "local", + "path": "100_AAA_orig.png", + "sha256": "deadbeef", + "bytes": 12345, + "media_type": "image/png", + "integrity": "present", + "verified_at": NOW, + }, + ], + "publication": { + "selected_location_id": "origin:x", + "selected_at": NOW, + "reason": "default", + }, + } + ], + "mirrors": [], + } + manifest = _from_wire_dict(raw) + self.assertEqual(manifest.schema_version, 2) + self.assertEqual(manifest.root_post_id, "100") + self.assertEqual(manifest.captured_at, NOW) + self.assertEqual(len(manifest.items), 1) + item = manifest.items[0] + self.assertEqual(item.post_id, "100") + self.assertEqual(item.media_id, "AAA") + self.assertEqual(item.kind, "orig") + self.assertEqual(item.embed, True) + self.assertEqual(len(item.locations), 2) + loc0, loc1 = item.locations + self.assertEqual(loc0.id, "origin:x") + self.assertEqual(loc0.kind, "origin") + self.assertEqual(loc0.url, "https://pbs.twimg.com/media/AAA") + self.assertEqual(loc0.provider, "x") + self.assertEqual(loc0.availability, "unknown") + self.assertEqual(loc1.id, "local") + self.assertEqual(loc1.kind, "local") + self.assertEqual(loc1.integrity, "present") + self.assertEqual(loc1.sha256, "deadbeef") + self.assertEqual(loc1.bytes, 12345) + self.assertIsNotNone(item.publication) + assert item.publication is not None + self.assertEqual(item.publication.selected_location_id, "origin:x") + self.assertEqual(item.publication.reason, "default") + + def test_minimal_wire_payload(self) -> None: + raw: dict = { "schema_version": 2, "root_post_id": "100", "items": [ { "post_id": "100", "media_id": "AAA", - "handle": "example", "role": "orig", - "derived_from": None, - "embed": True, - "locations": [ - { - "id": "local", - "kind": "local", - "path": "100_AAA_orig.png", - "sha256": None, - "bytes": None, - "media_type": "image/png", - "integrity": "missing", - "verified_at": NOW, - } - ], - "publication": { - "selected_location_id": "local", - "selected_at": NOW, - "reason": "manual", - }, + "locations": [], } ], - "mirrors": [], } - self.assertIn("items[0] selects non-HTTPS location local", validate_manifest(manifest)) + manifest = _from_wire_dict(raw) + self.assertEqual(manifest.root_post_id, "100") + self.assertEqual(len(manifest.items), 1) + self.assertEqual(manifest.items[0].kind, "orig") + + def test_wire_payload_with_derived_from(self) -> None: + raw: dict = { + "schema_version": 2, + "root_post_id": "100", + "items": [ + { + "post_id": "100", + "media_id": "AAA", + "role": "ocr", + "derived_from": { + "post_id": "100", + "media_id": "AAA", + "role": "orig", + }, + "embed": False, + "locations": [], + "publication": None, + } + ], + } + manifest = _from_wire_dict(raw) + item = manifest.items[0] + self.assertEqual(item.kind, "ocr") + self.assertIsNotNone(item.derived_from) + assert item.derived_from is not None + self.assertEqual(item.derived_from.role, "orig") + self.assertIsNone(item.publication) + + +class CanonicalManifestBytesTests(unittest.TestCase): + """Tests for canonical_manifest_bytes with typed MediaManifest.""" + + def test_canonical_manifest_bytes_is_stable(self) -> None: + manifest = MediaManifest( + schema_version=2, + root_post_id="100", + items=( + MediaItem( + post_id="100", + media_id="AAA", + kind="orig", + role="orig", + locations=( + MediaLocation( + id="origin:x", + kind="origin", + url="https://pbs.twimg.com/media/AAA", + availability="unknown", + ), + ), + publication=Publication( + selected_location_id="origin:x", + selected_at=NOW, + reason="default", + ), + ), + ), + ) + first = canonical_manifest_bytes(manifest) + second = canonical_manifest_bytes(manifest) + self.assertEqual(first, second) + decoded = json.loads(first.decode("utf-8")) + self.assertEqual(decoded["schema_version"], 2) + self.assertEqual(decoded["root_post_id"], "100") + self.assertEqual(decoded["mirrors"], []) + self.assertEqual(len(decoded["items"]), 1) + self.assertEqual(decoded["items"][0]["media_id"], "AAA") + + def test_canonical_manifest_bytes_omits_mirrors_and_canonicalizes(self) -> None: + manifest = MediaManifest( + schema_version=2, + root_post_id="100", + items=( + MediaItem( + post_id="100", + media_id="AAA", + kind="orig", + role="orig", + locations=( + MediaLocation( + id="origin:x", + kind="origin", + url="https://x", + availability="unknown", + ), + ), + publication=Publication( + selected_location_id="origin:x", + selected_at=NOW, + reason="default", + ), + ), + ), + ) + encoded = canonical_manifest_bytes(manifest) + self.assertIn(b'"mirrors":[]', encoded) if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/scripts/twitter/tests/test_media_migrate.py b/scripts/twitter/tests/test_media_migrate.py index 37bfc00..0c8ea17 100644 --- a/scripts/twitter/tests/test_media_migrate.py +++ b/scripts/twitter/tests/test_media_migrate.py @@ -4,8 +4,9 @@ import tempfile import unittest from pathlib import Path -from twitter.media_manifest import find_location, selected_url, validate_manifest +from twitter.media_manifest import _from_wire_dict, find_location, selected_url, validate_manifest from twitter.media_migrate import migrate_legacy_thread +from twitter.models import MediaItem from twitter.tests.helpers import NOW, legacy_thread, read_json @@ -22,20 +23,19 @@ class LegacyMigrationTests(unittest.TestCase): applied = migrate_legacy_thread(asset_dir, note_dir, now=NOW, apply=True) self.assertTrue(applied.changed) - manifest = read_json(asset_dir / "media.json") + manifest_dict = read_json(asset_dir / "media.json") + manifest = _from_wire_dict(manifest_dict) self.assertEqual(validate_manifest(manifest), []) - item = manifest["items"][0] + item: MediaItem = manifest.items[0] self.assertEqual( - find_location(item, "origin:x")["url"], + find_location(item, "origin:x").url, "https://pbs.twimg.com/media/AAA?format=png&name=orig", ) fallback = next( - location - for location in item["locations"] - if location.get("kind") == "fallback" + location for location in item.locations if location.kind == "fallback" ) - self.assertEqual(fallback["url"], "https://files.catbox.moe/fallback.png") - self.assertEqual(selected_url(item), find_location(item, "origin:x")["url"]) + self.assertEqual(fallback.url, "https://files.catbox.moe/fallback.png") + self.assertEqual(selected_url(item), find_location(item, "origin:x").url) def test_second_migration_is_idempotent(self) -> None: with tempfile.TemporaryDirectory() as raw: @@ -49,4 +49,4 @@ class LegacyMigrationTests(unittest.TestCase): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/scripts/twitter/tree.py b/scripts/twitter/tree.py index a76ef28..d04f157 100644 --- a/scripts/twitter/tree.py +++ b/scripts/twitter/tree.py @@ -1,3 +1,4 @@ +"""Tree-walking helpers for thread spine and branch construction.""" from __future__ import annotations from .models import ThreadData, PostData diff --git a/scripts/twitter/tw.py b/scripts/twitter/tw.py index cb1671f..1b208b8 100644 --- a/scripts/twitter/tw.py +++ b/scripts/twitter/tw.py @@ -18,12 +18,11 @@ import argparse import json import subprocess import sys +from collections.abc import Callable +from datetime import datetime, timezone from pathlib import Path -import sys -from pathlib import Path - -_HERE = Path(__file__).resolve().parent +_HERE: Path = Path(__file__).resolve().parent if str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE)) @@ -46,10 +45,12 @@ except ImportError: # pragma: no cover - script-mode import def frozen_ids() -> set[str]: + """Return the loaded set of frozen post_ids from the do-not-refetch list.""" return load_frozen_ids(FROZEN) def check_frozen(post_id: str) -> None: + """Raise SystemExit if ``post_id`` or any of its captured descendants matches a frozen id.""" direct = dump_dir(post_id) assets, _notes = locate(post_id) candidate = assets if assets is not None else direct @@ -70,6 +71,8 @@ def scratch_dir(post_id: str) -> Path: def locate(post_id: str) -> tuple[Path | None, Path | None]: + """Return ``(asset_dir, note_dir)`` for the first media.json whose root or captured posts contain ``post_id``.""" + assets_root = VAULT / "assets" / "threads" assets_root = VAULT / "assets" / "threads" notes_root = VAULT / "archive" / "threads" if not assets_root.is_dir(): @@ -150,7 +153,9 @@ def run(cmd: list[str]) -> None: subprocess.check_call(cmd) -def cmd_locate(post_id: str) -> int: +def cmd_locate(args: argparse.Namespace) -> int: + """Print the on-disk paths (dump / scratch / assets / notes) for ``args.id``.""" + post_id = args.id dump = dump_dir(post_id) scratch = scratch_dir(post_id) assets, notes = locate(post_id) @@ -161,7 +166,9 @@ def cmd_locate(post_id: str) -> int: return 0 -def cmd_graph(post_id: str) -> int: +def cmd_graph(args: argparse.Namespace) -> int: + """Render the SSDL + ASCII graph for ``args.id`` into the scratch directory.""" + post_id = args.id src = scratch_dir(post_id) if not (src / "thread_data.json").is_file(): src = dump_dir(post_id) @@ -171,7 +178,9 @@ def cmd_graph(post_id: str) -> int: return 0 -def cmd_refetch(post_id: str) -> int: +def cmd_refetch(args: argparse.Namespace) -> int: + """gallery-dl a thread into the scratch dir, then run ``ingest_gallery.py`` to materialize thread_data.json.""" + post_id = args.id check_frozen(post_id) if not COOKIES.is_file(): raise SystemExit(f"missing {COOKIES}") @@ -227,7 +236,11 @@ def cmd_refetch(post_id: str) -> int: return 0 -def cmd_emit(post_id: str, tip: bool, slug: str | None) -> int: +def cmd_emit(args: argparse.Namespace) -> int: + """Run ``emit_archive.py`` for ``args.id`` and merge non-orig rows back into every per-author asset dir.""" + post_id = args.id + tip = args.tip + slug = args.slug check_frozen(post_id) src = scratch_dir(post_id) if not (src / "thread_data.json").is_file(): @@ -279,7 +292,10 @@ def cmd_emit(post_id: str, tip: bool, slug: str | None) -> int: return 0 -def cmd_lift(post_id: str, orig: bool) -> int: +def cmd_lift(args: argparse.Namespace) -> int: + """Invoke ``lift_catbox.py`` to swap fallback URLs into the notes for ``args.id`` (with ``--orig`` to restore).""" + post_id = args.id + orig = args.orig assets, notes = locate(post_id) if assets is None or notes is None: raise SystemExit(f"no vault thread for {post_id} — emit first") @@ -297,7 +313,10 @@ def cmd_lift(post_id: str, orig: bool) -> int: return 0 -def cmd_ocr(post_id: str, engine: str) -> int: +def cmd_ocr(args: argparse.Namespace) -> int: + """OCR every ``*_orig.{png,jpg}`` in the thread assets dir and write matching ``*_ocr.txt`` files.""" + post_id = args.id + engine = args.engine assets, _notes = locate(post_id) if assets is None: raise SystemExit(f"no assets thread for {post_id}") @@ -324,7 +343,9 @@ def cmd_ocr(post_id: str, engine: str) -> int: return 0 -def cmd_merge(post_id: str) -> int: +def cmd_merge(args: argparse.Namespace) -> int: + """Run ``media_merge.py`` for ``args.id`` to restore non-orig rows after an ``emit --force``.""" + post_id = args.id check_frozen(post_id) assets, _notes = locate(post_id) if assets is None: @@ -333,19 +354,21 @@ def cmd_merge(post_id: str) -> int: return 0 -def cmd_refresh(post_id: str, tip: bool, slug: str | None) -> int: - check_frozen(post_id) - cmd_refetch(post_id) - cmd_graph(post_id) - cmd_emit(post_id, tip=tip, slug=slug) +def cmd_refresh(args: argparse.Namespace) -> int: + """Convenience: ``refetch`` + ``graph`` + ``emit --force`` for ``args.id``.""" + check_frozen(args.id) + cmd_refetch(args) + cmd_graph(args) + cmd_emit(args) return 0 -def cmd_sync(handle: str | None) -> int: +def cmd_sync(args: argparse.Namespace) -> int: """Rebuild handle-index wikilinks from actual folder listing. Use after manual edits or a stale state. Idempotent. """ + handle = args.handle root = VAULT / "archive" / "threads" targets = [root / handle] if handle else None changed = 0 @@ -390,7 +413,9 @@ def cmd_sync(handle: str | None) -> int: return 0 -def cmd_publish(post_id: str) -> int: +def cmd_publish(args: argparse.Namespace) -> int: + """Flip ``draft: true`` to ``draft: false`` on the thread index (no-op if already published).""" + post_id = args.id check_frozen(post_id) _assets, notes = locate(post_id) if notes is None: @@ -413,8 +438,6 @@ def cmd_publish(post_id: str) -> int: def _now_iso() -> str: - from datetime import datetime, timezone - return ( datetime.now(timezone.utc) .replace(microsecond=0) @@ -423,7 +446,8 @@ def _now_iso() -> str: ) -def cmd_migrate_media(args) -> int: +def cmd_migrate_media(args: argparse.Namespace) -> int: + """Dispatch the per-thread ``migrate_legacy_thread`` (dry-run by default; ``--apply`` to write).""" if bool(args.id) == bool(args.all_root): raise SystemExit("migrate-media requires exactly one of --id or --all") if args.id: @@ -446,7 +470,8 @@ def cmd_migrate_media(args) -> int: return 0 -def cmd_audit_media(args) -> int: +def cmd_audit_media(args: argparse.Namespace) -> int: + """Run ``audit_thread`` for ``args.id`` and print issues (non-zero exit if any found).""" assets, note_dir = locate(args.id) if assets is None or note_dir is None: raise SystemExit(f"no vault thread for {args.id}") @@ -459,7 +484,8 @@ def cmd_audit_media(args) -> int: return 0 -def cmd_backup(args) -> int: +def cmd_backup(args: argparse.Namespace) -> int: + """Mirror ``args.id``'s asset dir into the configured destination via ``backup_thread``.""" check_frozen(args.id) assets, note_dir = locate(args.id) if assets is None or note_dir is None: @@ -477,7 +503,8 @@ def cmd_backup(args) -> int: return 0 if result.state == "synced" else 2 -def cmd_fallback(args) -> int: +def cmd_fallback(args: argparse.Namespace) -> int: + """Upload (or reuse) a fallback for ``args.id``/``args.media_id`` and rewrite the note references.""" if not args.confirm_origin_unavailable: raise SystemExit("fallback requires --confirm-origin-unavailable") check_frozen(args.id) @@ -508,7 +535,8 @@ def cmd_fallback(args) -> int: return 0 -def cmd_restore_origin(args) -> int: +def cmd_restore_origin(args: argparse.Namespace) -> int: + """Switch ``args.id``/``args.media_id`` back to the original X URL and rewrite the note references.""" check_frozen(args.id) assets, note_dir = locate(args.id) if assets is None or note_dir is None: @@ -592,41 +620,30 @@ def build_parser() -> argparse.ArgumentParser: return parser +_COMMANDS: dict[str, Callable[[argparse.Namespace], int]] = { + "locate": cmd_locate, + "graph": cmd_graph, + "refetch": cmd_refetch, + "emit": cmd_emit, + "lift": cmd_lift, + "ocr": cmd_ocr, + "merge": cmd_merge, + "refresh": cmd_refresh, + "publish": cmd_publish, + "sync": cmd_sync, + "migrate-media": cmd_migrate_media, + "audit-media": cmd_audit_media, + "backup": cmd_backup, + "fallback": cmd_fallback, + "restore-origin": cmd_restore_origin, +} + + def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) - post_id = getattr(args, "id", None) - if args.cmd == "locate": - return cmd_locate(post_id) - if args.cmd == "graph": - return cmd_graph(post_id) - if args.cmd == "refetch": - return cmd_refetch(post_id) - if args.cmd == "emit": - return cmd_emit(post_id, tip=args.tip, slug=args.slug) - if args.cmd == "lift": - return cmd_lift(post_id, orig=args.orig) - if args.cmd == "ocr": - return cmd_ocr(post_id, engine=args.engine) - if args.cmd == "merge": - return cmd_merge(post_id) - if args.cmd == "refresh": - return cmd_refresh(post_id, tip=args.tip, slug=args.slug) - if args.cmd == "publish": - return cmd_publish(post_id) - if args.cmd == "sync": - return cmd_sync(handle=args.handle) - if args.cmd == "migrate-media": - return cmd_migrate_media(args) - if args.cmd == "audit-media": - return cmd_audit_media(args) - if args.cmd == "backup": - return cmd_backup(args) - if args.cmd == "fallback": - return cmd_fallback(args) - if args.cmd == "restore-origin": - return cmd_restore_origin(args) - raise SystemExit(f"unknown {args.cmd}") + handler = _COMMANDS[args.cmd] + return handler(args) if __name__ == "__main__":