diff --git a/.gitignore b/.gitignore index a7b6c44..ec0d3ab 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,4 @@ assets/* secrets/* !secrets/README.md scripts/twitter/__pycache__ +scripts/twitter/tests/__pycache__ diff --git a/scripts/twitter/backup_assets.py b/scripts/twitter/backup_assets.py new file mode 100644 index 0000000..2582bc1 --- /dev/null +++ b/scripts/twitter/backup_assets.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import json +import shutil +import sys +import tomllib +from dataclasses import dataclass +from pathlib import Path + +try: + from .media_manifest import ( + atomic_write_json, + hash_file, + inventory_digest, + payload_inventory, + ) +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, + hash_file, + inventory_digest, + payload_inventory, + ) + + +@dataclass(frozen=True) +class BackupResult: + state: str + destination: Path + error: str | None + + +def load_destination_root(vault: Path, destination_id: str) -> Path: + path = vault / "secrets" / "credentials.toml" + if not path.is_file(): + raise RuntimeError(f"missing {path}") + text = path.read_text(encoding="utf-8") + try: + data = tomllib.loads(text) + backup = data.get("backup") or {} + configured_id = str(backup.get("id") or "cozy") + root = str(backup.get("root") or "").strip() + except tomllib.TOMLDecodeError: + values: dict[str, str] = {} + in_backup = False + for raw in text.splitlines(): + line = raw.strip() + if line.startswith("[") and line.endswith("]"): + in_backup = line[1:-1].strip().lower() == "backup" + continue + if in_backup and "=" in line: + key, _equals, remainder = line.partition("=") + values[key.strip().lower()] = remainder.strip().strip("\"'") + configured_id = values.get("id", "cozy") + root = values.get("root", "") + if configured_id != destination_id: + raise RuntimeError(f"backup destination not configured: {destination_id}") + if not root: + raise RuntimeError("credentials.toml [backup] root is empty") + return Path(root) + + +def backup_thread( + asset_dir: Path, + *, + assets_root: Path, + destination_root: Path, + destination_id: str, + now: str, + require_destination_root: bool = False, +) -> BackupResult: + media_path = asset_dir / "media.json" + manifest = json.loads(media_path.read_text(encoding="utf-8")) + relative = asset_dir.relative_to(assets_root) + destination = destination_root / relative + inventory = payload_inventory(asset_dir, manifest) + digest = inventory_digest(inventory) + mirror = { + "destination_id": destination_id, + "relative_path": relative.as_posix(), + "state": "error", + "inventory_digest": digest, + "last_attempted_at": now, + "completed_at": None, + "verified_at": None, + "error": None, + } + others = [ + entry + for entry in manifest.get("mirrors") or [] + if entry.get("destination_id") != destination_id + ] + try: + if require_destination_root and not destination_root.is_dir(): + raise OSError(f"destination root unavailable: {destination_root}") + destination.mkdir(parents=True, exist_ok=True) + for source in sorted(asset_dir.iterdir(), key=lambda path: path.name): + if source.is_file(): + shutil.copy2(source, destination / source.name) + for name, expected in inventory.items(): + if name == "media.json#canonical": + continue + copied = destination / name + if not copied.is_file() or hash_file(copied) != expected: + raise OSError(f"backup verification failed: {name}") + mirror["state"] = "synced" + mirror["completed_at"] = now + mirror["verified_at"] = now + except OSError as exc: + mirror["error"] = str(exc)[:300] + manifest["mirrors"] = others + [mirror] + atomic_write_json(media_path, manifest) + 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) + return BackupResult("error", destination, mirror["error"]) + return BackupResult(mirror["state"], destination, mirror["error"]) \ No newline at end of file diff --git a/scripts/twitter/catbox_client.py b/scripts/twitter/catbox_client.py new file mode 100644 index 0000000..d0f2e9f --- /dev/null +++ b/scripts/twitter/catbox_client.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import subprocess +import tomllib +from pathlib import Path + + +def load_userhash(vault: Path) -> str: + path = vault / "secrets" / "credentials.toml" + if not path.is_file(): + raise RuntimeError(f"missing {path}") + text = path.read_text(encoding="utf-8") + try: + data = tomllib.loads(text) + value = str((data.get("catbox") or {}).get("userhash") or "").strip() + except tomllib.TOMLDecodeError: + value = "" + in_catbox = False + for raw in text.splitlines(): + line = raw.strip() + if line.startswith("[") and line.endswith("]"): + in_catbox = line[1:-1].strip().lower() == "catbox" + continue + if in_catbox and line.lower().startswith("userhash"): + _key, _equals, remainder = line.partition("=") + value = remainder.strip().strip("\"'") + break + if not value: + raise RuntimeError("credentials.toml [catbox] userhash is empty") + return value + + +def upload_file(path: Path, userhash: str) -> str: + cmd = [ + "curl.exe", + "-sS", + "--max-time", + "180", + "--limit-rate", + "200k", + "-F", + "reqtype=fileupload", + "-F", + f"userhash={userhash}", + "-F", + f"fileToUpload=@{path}", + "https://catbox.moe/user/api.php", + ] + try: + raw = subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT) + except subprocess.CalledProcessError as exc: + raise RuntimeError( + f"catbox upload failed for {path.name} (curl exit {exc.returncode})" + ) from None + url = raw.strip() + if not url.startswith("https://"): + raise RuntimeError(f"catbox refused {path.name}: non-URL response") + return url \ No newline at end of file diff --git a/scripts/twitter/emit_archive.py b/scripts/twitter/emit_archive.py index 8234155..3fc9f3d 100644 --- a/scripts/twitter/emit_archive.py +++ b/scripts/twitter/emit_archive.py @@ -14,17 +14,121 @@ from urllib.parse import parse_qs, unquote, urlparse sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from twitter.models import 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 ( - by_id, - branch_roots, - children_map, - descendants, - spine_from_tip, - spine_ids, -) +try: + from twitter.media_manifest import ( + atomic_write_json, + find_location, + item_key, + merge_manifest_items, + new_original_item, + selected_url, + ) + from twitter.media_refs import remote_markup + from twitter.models import 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 ( + by_id, + branch_roots, + children_map, + descendants, + spine_from_tip, + spine_ids, + ) +except ImportError: # pragma: no cover - script-mode import + from media_manifest import ( + atomic_write_json, + find_location, + item_key, + merge_manifest_items, + new_original_item, + selected_url, + ) + from media_refs import remote_markup + from models import ThreadData, load_thread + from render import render_branch, render_spine + from slug import branch_file_name, date_prefix, thread_dir_name + from tree import ( + by_id, + branch_roots, + children_map, + descendants, + spine_from_tip, + spine_ids, + ) + + +def _frontmatter_lines(text: str) -> tuple[list[str], str]: + if not text.startswith("---\n"): + return [], text + end = text.find("\n---", 4) + if end < 0: + return [], text + return text[4:end].splitlines(), text[end + 4 :] + + +def _tag_values(lines: list[str]) -> list[str]: + tags: list[str] = [] + in_tags = False + for line in lines: + if line == "tags:": + in_tags = True + continue + if in_tags and line.startswith(" - "): + tags.append(line[4:]) + continue + if in_tags and line and not line.startswith(" "): + break + return tags + + +def preserve_review_state( + old: str, + fresh: str, + *, + mechanical_tags: set[str], +) -> str: + old_lines, _old_body = _frontmatter_lines(old) + fresh_lines, fresh_body = _frontmatter_lines(fresh) + if not old_lines or not fresh_lines: + return fresh + old_draft = next( + (line for line in old_lines if line.startswith("draft: ")), + "draft: true", + ) + reviewed = [tag for tag in _tag_values(old_lines) if tag not in mechanical_tags] + output: list[str] = [] + in_tags = False + inserted_reviewed = False + for line in fresh_lines: + if line.startswith("draft: "): + output.append(old_draft) + continue + if line == "tags:": + in_tags = True + output.append(line) + continue + if in_tags and line.startswith(" - "): + output.append(line) + continue + if in_tags and not inserted_reviewed: + output.extend(f" - {tag}" for tag in reviewed if f" - {tag}" not in output) + inserted_reviewed = True + in_tags = False + output.append(line) + if in_tags and not inserted_reviewed: + output.extend(f" - {tag}" for tag in reviewed if f" - {tag}" not in output) + return "---\n" + "\n".join(output) + "\n---" + fresh_body + + +def missing_local_media(items: list[dict[str, object]]) -> list[str]: + missing: list[str] = [] + for item in items: + local = find_location(item, "local") + if local is not None and local.get("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*$") @@ -36,6 +140,7 @@ class EmitResult: handle: str dir_name: str branch_count: int + missing_media_count: int def _url_media_id(url: str, index: int) -> str: @@ -80,11 +185,11 @@ def collect_media( thread: ThreadData, input_dir: Path, dest_dir: Path, -) -> tuple[list[dict[str, object]], dict[str, tuple[str, ...]]]: + *, + now: str, +) -> list[dict[str, object]]: unused = _media_files(input_dir / "media") items: list[dict[str, object]] = [] - by_post: dict[str, list[str]] = {} - for post in thread.posts: for index, url in enumerate(post.media_urls, start=1): media_id = _url_media_id(url, index) @@ -97,67 +202,60 @@ def collect_media( else: ext = _url_ext(url) or "bin" filename = f"{post.post_id}_{media_id}_orig.{ext}" + local_path = dest_dir / filename if src is not None: dest_dir.mkdir(parents=True, exist_ok=True) - shutil.copy2(src, dest_dir / filename) - items.append({ - "post_id": post.post_id, - "media_id": media_id, - "handle": post.handle, - "role": "orig", - "filename": filename, - "publish": False, - "url": None, - "embed": True, - }) - by_post.setdefault(post.post_id, []).append(filename) + shutil.copy2(src, local_path) + items.append( + new_original_item( + post_id=post.post_id, + media_id=media_id, + handle=post.handle, + origin_url=url, + filename=filename, + local_path=local_path, + now=now, + ) + ) + return items - media_by_post = {pid: tuple(names) for pid, names in by_post.items()} - return items, media_by_post + +def selected_media_by_post( + items: list[dict[str, object]], +) -> dict[str, tuple[str, ...]]: + grouped: dict[str, list[str]] = {} + for item in items: + if not item.get("embed"): + continue + url = selected_url(item) + if url is not None: + grouped.setdefault(str(item.get("post_id") or ""), []).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]], - media_by_post: dict[str, tuple[str, ...]], -) -> tuple[list[dict[str, object]], dict[str, tuple[str, ...]]]: - """Keep CRT/OCR rows and prior lift URLs across --force emit.""" +) -> list[dict[str, object]]: + """Merge existing derived/fallback rows with fresh canonical items.""" old_path = dest_dir / "media.json" if not old_path.is_file(): - return items, media_by_post + return items try: old = json.loads(old_path.read_text(encoding="utf-8")) except json.JSONDecodeError: - return items, media_by_post - by_file = {str(i.get("filename") or ""): i for i in items} - for row in old.get("items") or []: - if not isinstance(row, dict): - continue - fn = str(row.get("filename") or "") - role = str(row.get("role") or "") - if role == "orig" and fn in by_file: - cur = by_file[fn] - if row.get("url"): - cur["url"] = row["url"] - if row.get("publish"): - cur["publish"] = row["publish"] - if "embed" in row: - cur["embed"] = row["embed"] - continue - if role in ("", "orig"): - continue - if fn in by_file: - continue - if fn and (dest_dir / fn).is_file(): - items.append(row) - by_post: dict[str, list[str]] = {} - for item in items: - if item.get("role") == "orig" and item.get("embed", True): - by_post.setdefault(str(item.get("post_id") or ""), []).append( - str(item.get("filename") or "") - ) - media_by_post = {pid: tuple(names) for pid, names in by_post.items()} - return items, media_by_post + return items + if old.get("schema_version") != 2: + raise SystemExit( + "legacy media.json requires: tw.py migrate-media --id --apply" + ) + return merge_manifest_items(list(old.get("items") or []), items) + + +def _legacy_emit_message(*args: object, **kwargs: object) -> None: + raise SystemExit( + "legacy media.json requires: tw.py migrate-media --id --apply" + ) def missing_parent_ids(thread: ThreadData) -> list[str]: @@ -485,52 +583,66 @@ def emit( note_dir.mkdir(parents=True, exist_ok=True) asset_dir.mkdir(parents=True, exist_ok=True) - items, media_by_post = collect_media(thread, input_dir, asset_dir) - items, media_by_post = merge_existing_media(asset_dir, items, media_by_post) + now = archived + "T12:00:00Z" + fresh_items = collect_media(thread, input_dir, asset_dir, now=now) + items = merge_existing_media(asset_dir, fresh_items) + media_by_post = selected_media_by_post(items) + missing_keys = missing_local_media(items) branch_links = { rid: wiki_branch(handle, dir_name, name) for rid, name in branch_names.items() } - (note_dir / "index.md").write_text( - render_spine( + + spine_text = render_spine( + thread, + spine, + roots, + branch_links, + archived, + media_by_post, + ) + spine_path = note_dir / "index.md" + if spine_path.is_file(): + spine_text = preserve_review_state( + spine_path.read_text(encoding="utf-8"), + spine_text, + mechanical_tags={"archive", "twitter", handle}, + ) + spine_path.write_text(spine_text, encoding="utf-8", newline="\n") + + for rid in roots: + branch_path = note_dir / f"{branch_names[rid]}.md" + branch_text = render_branch( thread, - spine, - roots, - branch_links, + rid, + descendants(rid, kids), + first.handle, archived, media_by_post, - ), - encoding="utf-8", - newline="\n", - ) - for rid in roots: - (note_dir / f"{branch_names[rid]}.md").write_text( - render_branch( - thread, - rid, - descendants(rid, kids), - first.handle, - archived, - media_by_post, - spine_wiki=wiki_thread(handle, dir_name), - ), - encoding="utf-8", - newline="\n", + spine_wiki=wiki_thread(handle, dir_name), ) + if branch_path.is_file(): + branch_text = preserve_review_state( + branch_path.read_text(encoding="utf-8"), + branch_text, + mechanical_tags={"archive", "twitter", handle}, + ) + branch_path.write_text(branch_text, encoding="utf-8", newline="\n") shutil.copy2(src, asset_dir / "thread_data.json") media_doc = { + "schema_version": 2, "root_post_id": thread.root_post_id, "items": items, + "mirrors": [], } - (asset_dir / "media.json").write_text( - json.dumps(media_doc, indent=2) + "\n", - encoding="utf-8", - newline="\n", - ) + atomic_write_json(asset_dir / "media.json", media_doc) + gaps_text = render_gaps(thread, spine, input_kind=input_kind) + if missing_keys: + gaps_text += "\nmissing_media_local:\n" + "\n".join(missing_keys) + "\n" (asset_dir / "gaps.md").write_text( - render_gaps(thread, spine, input_kind=input_kind), + gaps_text, encoding="utf-8", newline="\n", ) @@ -543,6 +655,7 @@ def emit( handle=handle, dir_name=dir_name, branch_count=len(roots), + missing_media_count=len(missing_keys), ) diff --git a/scripts/twitter/fallback_media.py b/scripts/twitter/fallback_media.py new file mode 100644 index 0000000..257a801 --- /dev/null +++ b/scripts/twitter/fallback_media.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + +try: + from .media_manifest import ( + atomic_write_json, + find_location, + hash_file, + new_fallback_location, + selected_url, + ) + from .media_refs import atomic_write_text, remote_markup +except ImportError: # pragma: no cover - script-mode import + if str(Path(__file__).resolve().parent) not in sys.path: + sys.path.insert(0, str(Path(__file__).resolve().parent)) + from media_manifest import ( + atomic_write_json, + find_location, + hash_file, + new_fallback_location, + selected_url, + ) + from media_refs import atomic_write_text, remote_markup + + +@dataclass(frozen=True) +class FallbackResult: + url: str + reused: bool + + +def replace_selected_url(note_dir: Path, old_url: str, new_url: str) -> None: + if old_url == new_url: + return + old_markup = remote_markup(old_url) + new_markup = remote_markup(new_url) + hits: list[Path] = [] + for path in note_dir.glob("*.md"): + if old_markup in path.read_text(encoding="utf-8"): + hits.append(path) + total = sum(path.read_text(encoding="utf-8").count(old_markup) for path in hits) + if total != 1: + if total == 0 and sum( + path.read_text(encoding="utf-8").count(new_markup) + for path in note_dir.glob("*.md") + ) == 1: + return + raise ValueError("selected origin reference not found exactly once") + path = hits[0] + text = path.read_text(encoding="utf-8") + atomic_write_text(path, text.replace(old_markup, new_markup, 1)) + + +def _group(manifest: dict[str, Any], media_id: str) -> list[dict[str, Any]]: + return [ + item + for item in manifest.get("items") or [] + if str(item.get("media_id") or "") == media_id + ] + + +def _one(items: list[dict[str, Any]], role: str) -> dict[str, Any]: + matches = [item for item in items if item.get("role") == role] + if len(matches) != 1: + raise ValueError(f"expected one role={role}, found {len(matches)}") + return matches[0] + + +def _verified_local(item: dict[str, Any], asset_dir: Path) -> tuple[Path, str]: + local = find_location(item, "local") + if local is None or local.get("integrity") != "present": + raise ValueError("selected item has no present local location") + path = asset_dir / str(local.get("path") or "") + expected = str(local.get("sha256") or "") + if not path.is_file() or not expected or hash_file(path) != expected: + raise ValueError("selected local file failed hash verification") + return path, expected + + +def activate_fallback( + asset_dir: Path, + note_dir: Path, + *, + media_id: str, + role: str, + provider: str, + confirm_origin_unavailable: bool, + now: str, + upload: Callable[[Path], str], + lookup: Callable[[str, str], str | None] | None = None, +) -> FallbackResult: + if not confirm_origin_unavailable: + raise ValueError("origin-unavailable confirmation is required") + media_path = asset_dir / "media.json" + manifest = json.loads(media_path.read_text(encoding="utf-8")) + group = _group(manifest, media_id) + original = _one(group, "orig") + target = _one(group, role) + origin = find_location(original, "origin:x") + if origin is None: + raise ValueError("original item has no X origin location") + old_url = selected_url(next((item for item in group if item.get("embed")), original)) + if old_url is None: + raise ValueError("media group has no selected HTTPS location") + local_path, content_hash = _verified_local(target, asset_dir) + origin["availability"] = "unavailable" + origin["confirmed_unavailable_at"] = now + + fallback = next( + ( + location + for location in target.get("locations") or [] + if location.get("kind") == "fallback" + and location.get("provider") == provider + and location.get("sha256") == content_hash + ), + None, + ) + reused = fallback is not None + if fallback is None and lookup is not None: + reused_url = lookup(provider, content_hash) + if reused_url is not None: + fallback = new_fallback_location( + provider=provider, + url=reused_url, + content_hash=content_hash, + recorded_at=now, + uploaded_at=None, + ) + target.setdefault("locations", []).append(fallback) + reused = True + if fallback is None: + url = upload(local_path) + fallback = new_fallback_location( + provider=provider, + url=url, + content_hash=content_hash, + recorded_at=now, + uploaded_at=now, + ) + target.setdefault("locations", []).append(fallback) + + atomic_write_json(media_path, manifest) + replace_selected_url(note_dir, old_url, str(fallback["url"])) + for item in group: + item["embed"] = item is target + target["publication"] = { + "selected_location_id": fallback["id"], + "selected_at": now, + "reason": "origin-unavailable", + } + atomic_write_json(media_path, manifest) + return FallbackResult(str(fallback["url"]), reused) + + +def find_existing_fallback( + assets_root: Path, + provider: str, + content_hash: str, +) -> str | None: + urls: set[str] = set() + for media_path in assets_root.rglob("media.json"): + data = json.loads(media_path.read_text(encoding="utf-8")) + if data.get("schema_version") != 2: + continue + for item in data.get("items") or []: + for location in item.get("locations") or []: + if ( + location.get("kind") == "fallback" + and location.get("provider") == provider + and location.get("sha256") == content_hash + and str(location.get("url") or "").startswith("https://") + ): + urls.add(str(location["url"])) + return sorted(urls)[0] if urls else None + + +def restore_origin( + asset_dir: Path, + note_dir: Path, + *, + media_id: str, + now: str, +) -> None: + media_path = asset_dir / "media.json" + manifest = json.loads(media_path.read_text(encoding="utf-8")) + group = _group(manifest, media_id) + original = _one(group, "orig") + origin = find_location(original, "origin:x") + if origin is None: + raise ValueError("original item has no X origin location") + visible = next((item for item in group if item.get("embed")), original) + old_url = selected_url(visible) + new_url = str(origin.get("url") or "") + if old_url is None or not new_url.startswith("https://"): + raise ValueError("media group has no restorable origin") + replace_selected_url(note_dir, old_url, new_url) + origin["availability"] = "available" + origin["checked_at"] = now + origin["check"] = { + "status": None, + "result": "available", + "detail": "manual restore-origin selection", + } + for item in group: + item["embed"] = item is original + original["publication"] = { + "selected_location_id": "origin:x", + "selected_at": now, + "reason": "manual", + } + atomic_write_json(media_path, manifest) \ No newline at end of file diff --git a/scripts/twitter/frozen.py b/scripts/twitter/frozen.py new file mode 100644 index 0000000..77593d5 --- /dev/null +++ b/scripts/twitter/frozen.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import json +from pathlib import Path + + +def load_frozen_ids(path: Path) -> set[str]: + if not path.is_file(): + return set() + return { + line.strip() + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() and not line.lstrip().startswith("#") + } + + +def captured_ids(asset_dir: Path) -> set[str]: + found: set[str] = set() + media_path = asset_dir / "media.json" + if media_path.is_file(): + media = json.loads(media_path.read_text(encoding="utf-8")) + root_id = str(media.get("root_post_id") or "") + if root_id: + found.add(root_id) + for item in media.get("items") or []: + post_id = str(item.get("post_id") or "") + if post_id: + found.add(post_id) + thread_path = asset_dir / "thread_data.json" + if thread_path.is_file(): + thread = json.loads(thread_path.read_text(encoding="utf-8")) + root_id = str(thread.get("root_post_id") or "") + if root_id: + found.add(root_id) + for post in thread.get("posts") or []: + post_id = str(post.get("post_id") or "") + if post_id: + found.add(post_id) + return found + + +def frozen_match(asset_dir: Path, frozen_ids: set[str]) -> str | None: + matches = sorted(captured_ids(asset_dir) & frozen_ids) + return matches[0] if matches else None + + +def require_writable(asset_dir: Path, frozen_ids: set[str]) -> None: + match = frozen_match(asset_dir, frozen_ids) + if match is not None: + raise SystemExit(f"frozen: skipped ({match})") \ No newline at end of file diff --git a/scripts/twitter/lift_catbox.py b/scripts/twitter/lift_catbox.py index 230b286..fbc642c 100644 --- a/scripts/twitter/lift_catbox.py +++ b/scripts/twitter/lift_catbox.py @@ -1,204 +1,19 @@ -"""Upload curated stills to catbox and rewrite archive notes to HTTPS. +"""Deprecated. -Reads userhash from vault secrets/credentials.toml [catbox]. Never prints it. - - python lift_catbox.py --thread ASSETS_THREAD --notes ARCHIVE_THREAD --orig - ---orig marks every on-disk orig row publish=true then uploads those. -Default without --orig: only rows already publish=true. - -After upload, any note token that is the local filename (backticked or -bare) or a previous url for that row becomes ![](https://files.catbox.moe/…). - -Catbox rate-limits the userhash per minute. Algorithm: - - - One attempt per upload. No curl retries. A failure is a failure — - the user retries the whole lift later (the script persists progress - after each success, so partial work survives). - - 60-second sleep between consecutive uploads in the same run. - - 200 KB/s upload cap per curl. - - On empty/non-URL body: sleep 120s before raising, so a manual - retry minutes later doesn't arrive during catbox's cooldown. - - media.json is rewritten after each successful upload (atomic), - not at the end of the run. - - Errors are sanitized to redact the userhash before printing. +Bulk lift has been retired. Use `tw.py fallback --id --media-id --role +--confirm-origin-unavailable` after confirming the origin is unavailable. """ from __future__ import annotations -import argparse -import json -import re -import subprocess import sys -import time -import tomllib -from pathlib import Path - -from paths import VAULT as VAULT_ROOT - -API = "https://catbox.moe/user/api.php" -MAX_TIME = 180 -THROTTLE_SECS = 60.0 -LIMIT_RATE = "200k" -EMPTY_BACKOFF = 120.0 -def _redact_cmd(cmd: list[str]) -> str: - out: list[str] = [] - for tok in cmd: - if tok.startswith("userhash="): - out.append("userhash=") - else: - out.append(tok) - return " ".join(out) - - -def load_userhash(vault: Path) -> str: - path = vault / "secrets" / "credentials.toml" - if not path.is_file(): - raise SystemExit(f"missing {path}") - text = path.read_text(encoding="utf-8") - try: - data = tomllib.loads(text) - h = str((data.get("catbox") or {}).get("userhash") or "").strip() - except tomllib.TOMLDecodeError: - h = "" - in_catbox = False - for raw in text.splitlines(): - line = raw.strip() - if line.startswith("[") and line.endswith("]"): - in_catbox = line[1:-1].strip().lower() == "catbox" - continue - if in_catbox and line.lower().startswith("userhash"): - _, _, rest = line.partition("=") - h = rest.strip().strip("\"'") - break - if not h: - raise SystemExit("credentials.toml [catbox] userhash is empty") - return h - - -def upload(path: Path, userhash: str) -> str: - cmd = [ - "curl.exe", - "-sS", - "--max-time", - str(MAX_TIME), - "--limit-rate", - LIMIT_RATE, - "-F", - "reqtype=fileupload", - "-F", - f"userhash={userhash}", - "-F", - f"fileToUpload=@{path}", - API, - ] - try: - raw = subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT) - except subprocess.CalledProcessError as exc: - time.sleep(EMPTY_BACKOFF) - raise SystemExit( - f"catbox upload failed for {path.name} " - f"(curl exit {exc.returncode}); waited {EMPTY_BACKOFF:.0f}s before exit. " - f"cmd: {_redact_cmd(cmd)}" - ) from None - url = raw.strip() - if not url.startswith("https://"): - time.sleep(EMPTY_BACKOFF) - raise SystemExit( - f"catbox refused {path.name} (empty/non-URL body); " - f"waited {EMPTY_BACKOFF:.0f}s before exit. body[:200]={url[:200]!r}" - ) - return url - - -def rewrite_notes(notes_dir: Path, old_tokens: list[str], url: str) -> int: - embed = f"![]({url})" - changed = 0 - for path in notes_dir.glob("*.md"): - text = path.read_text(encoding="utf-8") - new = text - for token in old_tokens: - if not token: - continue - new = new.replace(f"`{token}`", embed) - new = new.replace(f"![]({token})", embed) - new = re.sub(r"^Media \(not lifted\):\s*", "", new, flags=re.M) - while True: - split = re.sub( - r"(!\[[^\]]*\]\([^)]+\))\s+(!\[[^\]]*\]\([^)]+\))", - r"\1\n\n\2", - new, - ) - if split == new: - break - new = split - if new != text: - path.write_text(new, encoding="utf-8", newline="\n") - changed += 1 - print(f"rewrote {path.name}") - return changed - - -def persist(media_path: Path, data: dict) -> None: - media_path.write_text( - json.dumps(data, indent=2) + "\n", - encoding="utf-8", - newline="\n", +def main(argv: list[str] | None = None) -> int: # pragma: no cover - deprecated entry + del argv + sys.stderr.write( + "lift_catbox.py is retired; use tw.py fallback with --confirm-origin-unavailable\n" ) - - -def main(argv: list[str] | None = None) -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--vault", type=Path, default=VAULT_ROOT) - parser.add_argument("--thread", type=Path, required=True) - parser.add_argument("--notes", type=Path, required=True) - parser.add_argument( - "--orig", - action="store_true", - help="publish+upload every orig file that exists on disk", - ) - parser.add_argument("--dry-run", action="store_true") - args = parser.parse_args(argv) - - media_path = args.thread / "media.json" - data = json.loads(media_path.read_text(encoding="utf-8")) - items = data.setdefault("items", []) - if args.orig: - for item in items: - if item.get("role") == "orig": - item["publish"] = True - if "embed" not in item: - item["embed"] = True - - userhash = None if args.dry_run else load_userhash(args.vault) - uploaded = 0 - publishable = [it for it in items if it.get("publish")] - for i, item in enumerate(publishable): - if i > 0 and not args.dry_run: - print(f"throttling {THROTTLE_SECS:.0f}s before next upload…") - time.sleep(THROTTLE_SECS) - name = str(item.get("filename") or "") - src = args.thread / name - if not src.is_file(): - print(f"skip missing {name}") - continue - old = [name, str(item.get("url") or "")] - if args.dry_run: - print(f"would upload {name}") - continue - assert userhash is not None - url = upload(src, userhash) - item["url"] = url - uploaded += 1 - print(f"uploaded {name} -> {url}") - rewrite_notes(args.notes, old, url) - persist(media_path, data) - print(f"persisted media.json ({uploaded} done so far)") - - print(f"uploaded {uploaded}") - return 0 + return 2 if __name__ == "__main__": diff --git a/scripts/twitter/media_audit.py b/scripts/twitter/media_audit.py new file mode 100644 index 0000000..66fb4f9 --- /dev/null +++ b/scripts/twitter/media_audit.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import json +import sys +from collections import Counter +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +try: + from .frozen import frozen_match + from .media_manifest import ( + find_location, + hash_file, + inventory_digest, + item_key, + payload_inventory, + selected_url, + validate_manifest, + ) + from .media_refs import remote_markup +except ImportError: # pragma: no cover - script-mode import + if str(Path(__file__).resolve().parent) not in sys.path: + sys.path.insert(0, str(Path(__file__).resolve().parent)) + from frozen import frozen_match + from media_manifest import ( + find_location, + hash_file, + inventory_digest, + item_key, + payload_inventory, + selected_url, + validate_manifest, + ) + from media_refs import remote_markup + + +@dataclass(frozen=True) +class AuditReport: + frozen: bool + issues: tuple[str, ...] + + +def audit_local_item(item: dict[str, Any], 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 "") + 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"): + 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]: + 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, + } + + +def check_origin_url( + url: str, + checked_at: str, + opener: Any = urlopen, +) -> dict[str, Any]: + request = Request(url, method="HEAD", headers={"User-Agent": "Threadwell-media-audit/1"}) + try: + with opener(request, timeout=30) as response: + return classify_origin_response( + int(response.status), + "HEAD completed", + checked_at, + ) + except HTTPError as exc: + return classify_origin_response(exc.code, f"HTTP {exc.code}", checked_at) + except (URLError, TimeoutError, OSError) as exc: + 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: + 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" + + +def audit_thread( + asset_dir: Path, + note_dir: Path, + frozen_ids: set[str], +) -> AuditReport: + 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 []: + 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 + ) + note_text = "\n".join( + path.read_text(encoding="utf-8") + for path in sorted(note_dir.glob("*.md")) + ) + for markup, count in expected.items(): + actual = note_text.count(markup) + if actual != count: + issues.append( + f"selected reference count mismatch: {markup} expected={count} actual={actual}" + ) + + current_digest = inventory_digest(payload_inventory(asset_dir, manifest)) + for mirror in manifest.get("mirrors") or []: + 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 diff --git a/scripts/twitter/media_embed.py b/scripts/twitter/media_embed.py index ae5209c..374c8ab 100644 --- a/scripts/twitter/media_embed.py +++ b/scripts/twitter/media_embed.py @@ -1,83 +1,32 @@ -"""Select which still a note cites, and optionally attach OCR under it. +"""Attach OCR transcripts under the cited origin media. Local variants are +not publication locations; visual selection is driven by the canonical +manifest's `embed` flag. -Notes currently say: - - Media (not lifted): `1736…_GBgHT…_orig.png` - -After lift they will say: - - ![](https://files.catbox.moe/xxxx.png) - -This script only rewrites those two shapes. It does not upload. - - python media_embed.py --thread --notes --media-id GBgHT_2WIAAKuNq --show crt - python media_embed.py --thread --notes --media-id GBgHT_2WIAAKuNq --attach-ocr + python media_embed.py --thread --notes --media-id --attach-ocr """ from __future__ import annotations import argparse import json -import re +import sys from pathlib import Path -IMAGE_MD = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)") -LIFTED_LINE = re.compile(r"^Media \(not lifted\): (.+)$", re.M) +try: + from .media_manifest import 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 -def load_media(path: Path) -> dict: - return json.loads(path.read_text(encoding="utf-8")) - - -def save_media(path: Path, data: dict) -> None: - path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") - - -def group_items(items: list[dict]) -> dict[tuple[str, str], dict[str, dict]]: - groups: dict[tuple[str, str], dict[str, dict]] = {} - for item in items: - key = (str(item.get("post_id") or ""), str(item.get("media_id") or "")) - groups.setdefault(key, {})[str(item.get("role") or "")] = item - return groups - - -def set_embed(items: list[dict], media_id: str, show: str) -> None: - found = False - for item in items: - if str(item.get("media_id") or "") != media_id: - continue - if item.get("role") in ("orig", "crt", "crt_outline"): - item["embed"] = item.get("role") == show - found = True - if not found: - raise SystemExit(f"no orig/crt row for media_id {media_id}") - - -def token_for(item: dict) -> str: - url = item.get("url") - if url: - return str(url) - return str(item.get("filename") or "") - - -def rewrite_notes(notes_dir: Path, old: str, new: str) -> int: - if not old or old == new: - return 0 - changed = 0 - for path in notes_dir.glob("*.md"): - text = path.read_text(encoding="utf-8") - if old not in text: - continue - path.write_text(text.replace(old, new), encoding="utf-8") - changed += 1 - print(f"rewrote {path.name}") - return changed +OCR_DETAILS = "Text from still" def ocr_block(text: str) -> str: body = text.strip("\n") return ( "
\n" - "Text from still\n" + f"{OCR_DETAILS}\n" "\n" "```\n" f"{body}\n" @@ -94,23 +43,20 @@ def attach_ocr(notes_dir: Path, image_token: str, ocr_text: str) -> int: text = path.read_text(encoding="utf-8") if image_token not in text: continue - if "Text from still" in text and image_token in text: + if OCR_DETAILS in text and image_token in text: print(f"ocr already near token in {path.name}") continue - # After the line that cites the still. lines = text.splitlines(keepends=True) out: list[str] = [] - i = 0 inserted = False - while i < len(lines): - out.append(lines[i]) - if image_token in lines[i] and not inserted: - if not lines[i].endswith("\n"): - out[-1] = lines[i] + "\n" + for i, line in enumerate(lines): + out.append(line) + if image_token in line and not inserted: + if not line.endswith("\n"): + out[-1] = line + "\n" out.append("\n") out.append(block + "\n") inserted = True - i += 1 if inserted: path.write_text("".join(out), encoding="utf-8") changed += 1 @@ -123,50 +69,54 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--thread", type=Path, required=True, help="assets/threads/.../ folder") parser.add_argument("--notes", type=Path, required=True, help="archive/threads/.../ folder") parser.add_argument("--media-id", required=True) - parser.add_argument("--show", choices=("orig", "crt"), default=None) parser.add_argument("--attach-ocr", action="store_true") args = parser.parse_args(argv) - media_path = args.thread / "media.json" - data = load_media(media_path) - items = data["items"] - groups = group_items(items) - match = None - pair = None - for key, roles in groups.items(): - if key[1] == args.media_id: - match = key - pair = roles - break - if pair is None or match is None: - raise SystemExit(f"media_id {args.media_id} not in {media_path}") - if args.show: - if args.show not in pair: - raise SystemExit(f"no role={args.show} for {args.media_id}") - old_item = next((pair[r] for r in ("crt", "orig") if pair.get(r, {}).get("embed")), None) - if old_item is None: - old_item = pair.get("orig") or pair.get("crt") - set_embed(items, args.media_id, args.show) - save_media(media_path, data) - new_item = pair[args.show] - rewrite_notes(args.notes, token_for(old_item), token_for(new_item)) - print(f"embed {args.media_id} -> {args.show} ({token_for(new_item)})") + raise SystemExit( + "local variants are not publishable locations; after confirming an " + "origin failure, use tw.py fallback with the desired --role" + ) + if not args.attach_ocr: + parser.print_help() + return 2 - if args.attach_ocr: - ocr = pair.get("ocr") - if ocr is None: - raise SystemExit(f"no role=ocr for {args.media_id}; run ocr_pass.py first") - ocr_path = args.thread / str(ocr["filename"]) - if not ocr_path.is_file(): - raise SystemExit(f"missing {ocr_path}") - shown = next((pair[r] for r in ("crt", "orig") if pair.get(r, {}).get("embed")), None) - if shown is None: - shown = pair.get("orig") or pair.get("crt") - attach_ocr(args.notes, token_for(shown), ocr_path.read_text(encoding="utf-8")) + media_path = args.thread / "media.json" + data = json.loads(media_path.read_text(encoding="utf-8")) + issues = validate_manifest(data) + if issues: + raise SystemExit(f"manifest invalid: {issues[0]}") + target = None + for item in data.get("items") or []: + if ( + str(item.get("media_id") or "") == args.media_id + and str(item.get("role") or "") == "ocr" + ): + target = item + break + if target 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 "") + 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: + continue + if item.get("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 "") + 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")) + atomic_write_json(media_path, data) return 0 if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file diff --git a/scripts/twitter/media_manifest.py b/scripts/twitter/media_manifest.py new file mode 100644 index 0000000..926ad14 --- /dev/null +++ b/scripts/twitter/media_manifest.py @@ -0,0 +1,337 @@ +from __future__ import annotations + +import copy +import json +import mimetypes +import os +from datetime import datetime, timezone +from hashlib import sha256 +from pathlib import Path +from typing import Any + + +SCHEMA_VERSION = 2 +VISUAL_ROLES = {"orig", "crt", "crt_outline", "denoise"} + + +def utc_now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def hash_file(path: Path) -> str: + digest = sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def atomic_write_json(path: Path, data: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temp = path.with_name(f".{path.name}.{os.getpid()}.tmp") + try: + with temp.open("w", encoding="utf-8", newline="\n") as handle: + json.dump(data, handle, indent=2, ensure_ascii=False) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp, path) + finally: + if temp.exists(): + 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 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: + 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: + return location + return None + + +def selected_location(item: dict[str, Any]) -> dict[str, Any] | None: + publication = item.get("publication") + if not isinstance(publication, dict): + return None + return find_location(item, str(publication.get("selected_location_id") or "")) + + +def selected_url(item: dict[str, Any]) -> str | None: + location = selected_location(item) + if not location or location.get("kind") not in {"origin", "fallback"}: + return None + value = str(location.get("url") or "") + return value if value.startswith("https://") else None + + +def _media_type(filename: str) -> str: + guessed, _encoding = mimetypes.guess_type(filename) + return guessed or "application/octet-stream" + + +def _local_location(filename: str, local_path: Path, now: str) -> dict[str, Any]: + 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, + } + + +def new_original_item( + *, + post_id: str, + media_id: str, + handle: str, + origin_url: str, + 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, + }, + _local_location(filename, local_path, now), + ], + "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]: + if item_key(existing) != item_key(fresh): + raise ValueError("cannot merge different media items") + merged = copy.deepcopy(existing) + merged["handle"] = fresh["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") + ): + raise ValueError(f"origin URL changed for {item_key(existing)!r}") + fresh_local = location_of_kind(fresh, "local") + locations = [ + copy.deepcopy(location) + for location in merged.get("locations") or [] + if location.get("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")) + 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} + fresh_keys: set[tuple[str, str, str]] = set() + merged: list[dict[str, Any]] = [] + for fresh in fresh_items: + key = item_key(fresh) + fresh_keys.add(key) + previous = existing_by_key.get(key) + merged.append(merge_item(previous, fresh) if previous is not None else fresh) + for previous in existing_items: + if item_key(previous) not in fresh_keys: + merged.append(copy.deepcopy(previous)) + return merged + + +def fallback_location_id(provider: str, url: str) -> str: + digest = sha256(f"{provider}\0{url}".encode("utf-8")).hexdigest()[:16] + return f"fallback:{provider}:{digest}" + + +def new_fallback_location( + *, + provider: str, + url: str, + 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, + } + + +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, + handle: str, + role: str, + 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) + key = item_key(fresh) + items = list(manifest.get("items") or []) + 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 + items.append(fresh) + manifest["items"] = items + return fresh + + +def canonical_manifest_bytes(data: dict[str, Any]) -> bytes: + stable = copy.deepcopy(data) + stable["mirrors"] = [] + 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]: + inventory: dict[str, str] = { + "media.json#canonical": sha256(canonical_manifest_bytes(manifest)).hexdigest() + } + for path in sorted(asset_dir.iterdir(), key=lambda candidate: candidate.name): + if path.is_file() and path.name != "media.json": + inventory[path.name] = hash_file(path) + return inventory + + +def inventory_digest(inventory: dict[str, str]) -> str: + encoded = json.dumps(inventory, sort_keys=True, separators=(",", ":")).encode("utf-8") + return sha256(encoded).hexdigest() + + +def validate_manifest(data: dict[str, Any]) -> list[str]: + issues: list[str] = [] + if data.get("schema_version") != SCHEMA_VERSION: + issues.append(f"schema_version must be {SCHEMA_VERSION}") + if not str(data.get("root_post_id") or ""): + issues.append("root_post_id is empty") + seen: set[tuple[str, str, str]] = set() + for index, item in enumerate(data.get("items") or []): + key = item_key(item) + if not all(key): + issues.append(f"items[{index}] has incomplete key {key!r}") + if key in seen: + 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 "") + 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") + if publication is None: + if item.get("role") in VISUAL_ROLES and item.get("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) + 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 diff --git a/scripts/twitter/media_merge.py b/scripts/twitter/media_merge.py index d801b38..7766b55 100644 --- a/scripts/twitter/media_merge.py +++ b/scripts/twitter/media_merge.py @@ -9,6 +9,8 @@ from __future__ import annotations import argparse import json +import sys +from datetime import datetime, timezone from pathlib import Path ROLE_SUFFIX = { @@ -46,8 +48,19 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) media_path = args.thread / "media.json" data = json.loads(media_path.read_text(encoding="utf-8")) + if data.get("schema_version") != 2: + 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 = {(i.get("filename"), i.get("role")) for i in items} + 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} added = 0 for path in sorted(args.thread.iterdir()): role = role_of(path.name) @@ -56,25 +69,20 @@ def main(argv: list[str] | None = None) -> int: if (path.name, role) in have: continue post_id, media_id = parse_ids(path.name) - handle = "" - for item in items: - if item.get("media_id") == media_id: - handle = str(item.get("handle") or "") - break - items.append({ - "post_id": post_id, - "media_id": media_id, - "handle": handle, - "role": role, - "filename": path.name, - "publish": False, - "url": None, - "embed": False, - }) + handle = handles.get(media_id, "") + upsert_derived_item( + data, + post_id=post_id, + media_id=media_id, + handle=handle, + role=role, + filename=path.name, + asset_dir=args.thread, + now=now, + ) added += 1 print(f"restore {path.name} role={role}") - data["items"] = items - media_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + atomic_write_json(media_path, data) print(f"added {added} -> {media_path}") return 0 diff --git a/scripts/twitter/media_migrate.py b/scripts/twitter/media_migrate.py new file mode 100644 index 0000000..b01319d --- /dev/null +++ b/scripts/twitter/media_migrate.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +import copy +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, + atomic_write_json, + find_location, + hash_file, + new_derived_item, + new_fallback_location, + new_original_item, + validate_manifest, + ) + from .media_refs import apply_rewrite_plan, atomic_write_text, plan_thread_rewrites +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, + atomic_write_json, + find_location, + hash_file, + new_derived_item, + new_fallback_location, + new_original_item, + validate_manifest, + ) + from media_refs import apply_rewrite_plan, atomic_write_text, plan_thread_rewrites + + +@dataclass(frozen=True) +class MigrationResult: + state: str + changed: bool + issues: tuple[str, ...] + item_count: int + + +def media_id_from_url(url: str, index: int) -> str: + segment = Path(unquote(urlparse(url).path)).name + return segment if segment else f"m{index}" + + +def source_urls(thread_data: dict[str, Any]) -> dict[tuple[str, str], str]: + 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)) + if key in result and result[key] != url: + raise ValueError(f"duplicate source mapping for {key!r}") + result[key] = str(url) + return result + + +def reference_maps( + legacy: dict[str, Any], canonical: dict[str, Any] +) -> 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 [] + } + filename_origins: dict[str, str] = {} + fallback_origins: dict[str, list[str]] = {} + for old in legacy.get("items") or []: + if old.get("role") != "orig": + continue + key = ( + str(old.get("post_id") or ""), + str(old.get("media_id") or ""), + "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 "") + if old_url.startswith("https://"): + fallback_origins.setdefault(old_url, []).append(origin) + return filename_origins, fallback_origins + + +def corpus_inventory( + assets_root: Path, + note_root: Path, + frozen_ids: set[str], + *, + now: str, +) -> dict[str, object]: + media_paths = sorted(assets_root.rglob("media.json")) + counts: dict[str, int] = { + "input_items": 0, + "migratable": 0, + "frozen_skipped": 0, + "missing_origin_mapping": 0, + "missing_local_file": 0, + "ambiguous_note_reference": 0, + "invalid_legacy_row": 0, + } + threads: list[dict[str, object]] = [] + 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 ""), + ) + for item in legacy_items + } + counts["input_items"] += len(unique_legacy) + for path in sorted(asset_dir.iterdir()): + if path.is_file() and path.name != "media.json": + key = f"{relative}/{path.name}" + frozen_hashes[key] = hash_file(path) + for archive in sorted((note_root / relative).glob("*.md")) if (note_root / relative).is_dir() else []: + key = f"{relative}/{archive.name}" + frozen_hashes[key] = hash_file(archive) + if match is not None: + counts["frozen_skipped"] += len(unique_legacy) + threads.append({ + "relative": relative, + "state": "frozen_skipped", + "frozen_id": match, + "item_count": len(unique_legacy), + }) + continue + thread_data_path = asset_dir / "thread_data.json" + if not thread_data_path.is_file(): + counts["missing_origin_mapping"] += len(unique_legacy) + threads.append({ + "relative": relative, + "state": "missing_origin_mapping", + "item_count": len(unique_legacy), + "issues": ["missing thread_data.json"], + }) + continue + thread_data = json.loads(thread_data_path.read_text(encoding="utf-8")) + try: + canonical, issues = canonical_from_legacy(legacy, thread_data, asset_dir, now) + except Exception as exc: + counts["invalid_legacy_row"] += len(unique_legacy) + threads.append({ + "relative": relative, + "state": "invalid_legacy_row", + "item_count": len(unique_legacy), + "issues": [str(exc)], + }) + continue + if issues: + counts["invalid_legacy_row"] += len(unique_legacy) + threads.append({ + "relative": relative, + "state": "invalid_legacy_row", + "item_count": len(unique_legacy), + "issues": list(issues), + }) + continue + counts["migratable"] += len(canonical["items"]) + thread_entry: dict[str, object] = { + "relative": relative, + "state": "migratable" if is_legacy else "current", + "item_count": len(canonical["items"]), + } + threads.append(thread_entry) + expected_total = ( + counts["migratable"] + + counts["frozen_skipped"] + + counts["missing_origin_mapping"] + + counts["missing_local_file"] + + counts["ambiguous_note_reference"] + + counts["invalid_legacy_row"] + ) + if counts["input_items"] != expected_total: + raise SystemExit("corpus accounting mismatch") + return { + "schema_version": 1, + "generated_at": now, + "counts": counts, + "threads": threads, + "frozen_hashes": frozen_hashes, + } + + +def canonical_from_legacy( + legacy: dict[str, Any], + thread_data: dict[str, Any], + asset_dir: Path, + now: str, +) -> tuple[dict[str, Any], tuple[str, ...]]: + sources = source_urls(thread_data) + items: list[dict[str, Any]] = [] + 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 "") + key = (post_id, media_id, role) + if key in seen: + continue + seen.add(key) + if role == "orig": + origin_url = sources.get((post_id, media_id)) + if origin_url is None: + issues.append(f"missing origin mapping: {post_id}/{media_id}") + continue + item = new_original_item( + post_id=post_id, + media_id=media_id, + handle=str(old.get("handle") or ""), + 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, + ) + ) + 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 "") + 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, + ) + ) + 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 []), + } + issues.extend(validate_manifest(manifest)) + return manifest, tuple(issues) + + +def migrate_legacy_thread( + asset_dir: Path, + note_dir: Path, + *, + now: str, + apply: bool, +) -> MigrationResult: + 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")) + canonical, issues = canonical_from_legacy(legacy, thread_data, asset_dir, now) + if issues: + return MigrationResult("blocked", False, issues, len(legacy.get("items") or [])) + if apply: + filename_origins, fallback_origins = reference_maps(legacy, canonical) + plan = plan_thread_rewrites( + note_dir, + filename_origins=filename_origins, + fallback_origins=fallback_origins, + ) + if plan.issues: + return MigrationResult("blocked", False, plan.issues, len(legacy.get("items") or [])) + note_backup: dict[Path, str] = {} + try: + for rewrite in plan.files: + note_backup[rewrite.path] = rewrite.before + atomic_write_json(media_path, canonical) + 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 diff --git a/scripts/twitter/media_refs.py b/scripts/twitter/media_refs.py new file mode 100644 index 0000000..4f10936 --- /dev/null +++ b/scripts/twitter/media_refs.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import os +import re +from collections import Counter, deque +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*$") + + +@dataclass(frozen=True) +class FileRewrite: + path: Path + before: str + after: str + + +@dataclass(frozen=True) +class RewritePlan: + files: tuple[FileRewrite, ...] + issues: tuple[str, ...] + + +def remote_markup(url: str) -> str: + path = url.split("?", 1)[0].lower() + if path.endswith(".mp4") or "video.twimg.com" in url: + return f'' + return f"![]({url})" + + +def _fallback_urls(text: str) -> list[str]: + return [ + match.group("url") + for match in IMAGE_LINE.finditer(text) + if "files.catbox.moe" in match.group("url") + ] + + +def plan_thread_rewrites( + note_dir: Path, + *, + filename_origins: dict[str, str], + fallback_origins: dict[str, list[str]], +) -> RewritePlan: + paths = sorted(note_dir.glob("*.md"), key=lambda path: (path.name != "index.md", path.name)) + texts = {path: path.read_text(encoding="utf-8") for path in paths} + fallback_counts = Counter( + url for text in texts.values() for url in _fallback_urls(text) + ) + fallback_files: dict[str, set[Path]] = {} + for path, text in texts.items(): + for url in _fallback_urls(text): + fallback_files.setdefault(url, set()).add(path) + + issues: list[str] = [] + for url, count in sorted(fallback_counts.items()): + candidates = fallback_origins.get(url) + if not candidates: + issues.append(f"untracked fallback reference: {url}") + elif len(candidates) != count: + issues.append(f"fallback occurrence mismatch: {url} expected={len(candidates)} actual={count}") + elif len(candidates) > 1 and len(fallback_files[url]) > 1: + issues.append(f"fallback spans multiple notes: {url}") + if issues: + return RewritePlan((), tuple(issues)) + + fallback_queues = { + url: deque(origins) for url, origins in fallback_origins.items() + } + rewrites: list[FileRewrite] = [] + for path in paths: + before = texts[path] + + def replace_media_line(match: re.Match[str], _path=path) -> str: + names = BACKTICK.findall(match.group("body")) + if not names: + return match.group(0) + urls: list[str] = [] + for name in names: + origin = filename_origins.get(name) + if origin is None: + issues.append(f"untracked local reference: {name}") + return match.group(0) + urls.append(origin) + return "\n\n".join(remote_markup(url) for url in urls) + + after = MEDIA_LINE.sub(replace_media_line, before) + + def replace_image_line(match: re.Match[str], _path=path) -> str: + url = match.group("url") + queue = fallback_queues.get(url) + if not queue: + return match.group(0) + return remote_markup(queue.popleft()) + + after = IMAGE_LINE.sub(replace_image_line, after) + if after != before: + rewrites.append(FileRewrite(path, before, after)) + if issues: + return RewritePlan((), tuple(sorted(set(issues)))) + return RewritePlan(tuple(rewrites), ()) + + +def atomic_write_text(path: Path, text: str) -> None: + temp = path.with_name(f".{path.name}.{os.getpid()}.tmp") + try: + with temp.open("w", encoding="utf-8", newline="\n") as handle: + handle.write(text) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp, path) + finally: + if temp.exists(): + temp.unlink() + + +def apply_rewrite_plan(plan: RewritePlan) -> None: + if plan.issues: + raise ValueError("cannot apply a rewrite plan with issues") + for rewrite in plan.files: + atomic_write_text(rewrite.path, rewrite.after) \ No newline at end of file diff --git a/scripts/twitter/ocr_pass.py b/scripts/twitter/ocr_pass.py index a58942d..5b31e41 100644 --- a/scripts/twitter/ocr_pass.py +++ b/scripts/twitter/ocr_pass.py @@ -18,6 +18,7 @@ import subprocess import sys import urllib.error import urllib.request +from datetime import datetime, timezone from pathlib import Path UMI_URL = "http://127.0.0.1:1224/api/ocr" @@ -134,14 +135,20 @@ $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 data = json.loads(media_json.read_text(encoding="utf-8")) - items = data.setdefault("items", []) - stem = src.name + if data.get("schema_version") != 2: + raise SystemExit( + "legacy media.json requires: tw.py migrate-media --id --apply" + ) post_id = "" media_id = "" handle = "" - for item in items: - if item.get("filename") == src.name or item.get("filename") == dest.name: + for item in data.get("items") or []: + if str(item.get("filename") or "") in {src.name, dest.name}: post_id = str(item.get("post_id") or "") media_id = str(item.get("media_id") or "") handle = str(item.get("handle") or "") @@ -150,21 +157,18 @@ def append_media_row(media_json: Path, src: Path, dest: Path) -> None: parts = src.stem.split("_") post_id = parts[0] media_id = parts[1] if len(parts) > 1 else "" - row = { - "post_id": post_id, - "media_id": media_id, - "handle": handle, - "role": "ocr", - "filename": dest.name, - "publish": False, - "url": None, - "embed": False, - "source_filename": src.name, - } - items = [i for i in items if not (i.get("role") == "ocr" and i.get("source_filename") == src.name)] - items.append(row) - data["items"] = items - media_json.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + asset_dir = media_json.parent + upsert_derived_item( + data, + post_id=post_id, + media_id=media_id, + handle=handle, + role="ocr", + filename=dest.name, + asset_dir=asset_dir, + now=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + ) + atomic_write_json(media_json, data) print(f"updated {media_json}") diff --git a/scripts/twitter/render.py b/scripts/twitter/render.py index bbe7fa3..53e0626 100644 --- a/scripts/twitter/render.py +++ b/scripts/twitter/render.py @@ -2,11 +2,25 @@ from __future__ import annotations +import sys +from pathlib import Path + +try: + from .media_refs import remote_markup +except ImportError: # pragma: no cover - script-mode import + if str(Path(__file__).resolve().parent) not in sys.path: + sys.path.insert(0, str(Path(__file__).resolve().parent)) + from media_refs import remote_markup + from .models import PostData, ThreadData from .slug import date_prefix from .tree import by_id +def render_media(url: str) -> str: + return remote_markup(url) + + def _yaml_quote(value: str) -> str: return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' @@ -91,7 +105,7 @@ def _post_block( media = (media_by_post or {}).get(post.post_id) or () if media: parts.append("") - parts.append("Media (not lifted): " + " ".join(f"`{fn}`" for fn in media)) + parts.extend(render_media(url) for url in media) if branch_links: parts.append("") parts.append("Branches: " + ", ".join(f"[[{name}]]" for name in branch_links)) diff --git a/scripts/twitter/tw.py b/scripts/twitter/tw.py index 5edc90e..0a2ba63 100644 --- a/scripts/twitter/tw.py +++ b/scripts/twitter/tw.py @@ -20,23 +20,45 @@ import subprocess import sys from pathlib import Path +import sys +from pathlib import Path + +_HERE = Path(__file__).resolve().parent +if str(_HERE) not in sys.path: + sys.path.insert(0, str(_HERE)) + from paths import COOKIES, DUMPS, FROZEN, HERE, SCRATCH, VAULT +try: + from frozen import frozen_match, load_frozen_ids, require_writable + from backup_assets import backup_thread, load_destination_root + from catbox_client import load_userhash, upload_file + from fallback_media import activate_fallback, find_existing_fallback, restore_origin + from media_audit import audit_thread + from media_migrate import migrate_legacy_thread +except ImportError: # pragma: no cover - script-mode import + from frozen import frozen_match, load_frozen_ids, require_writable + from backup_assets import backup_thread, load_destination_root + from catbox_client import load_userhash, upload_file + from fallback_media import activate_fallback, find_existing_fallback, restore_origin + from media_audit import audit_thread + from media_migrate import migrate_legacy_thread + def frozen_ids() -> set[str]: - out: set[str] = set() - if not FROZEN.is_file(): - return out - for line in FROZEN.read_text(encoding="utf-8").splitlines(): - s = line.strip() - if s and not s.startswith("#"): - out.add(s) - return out + return load_frozen_ids(FROZEN) def check_frozen(post_id: str) -> None: + direct = dump_dir(post_id) + assets, _notes = locate(post_id) + candidate = assets if assets is not None else direct + if candidate.is_dir(): + match = frozen_match(candidate, frozen_ids()) + if match is not None: + raise SystemExit(f"frozen: skipped ({match})") if post_id in frozen_ids(): - raise SystemExit(f"frozen Onat dump {post_id} — will not refetch/overwrite") + raise SystemExit(f"frozen: skipped ({post_id})") def dump_dir(post_id: str) -> Path: @@ -166,6 +188,7 @@ def cmd_refetch(post_id: str) -> int: def cmd_emit(post_id: str, tip: bool, slug: str | None) -> int: + check_frozen(post_id) src = scratch_dir(post_id) if not (src / "thread_data.json").is_file(): src = dump_dir(post_id) @@ -237,6 +260,7 @@ def cmd_ocr(post_id: str, engine: str) -> int: def cmd_merge(post_id: str) -> int: + check_frozen(post_id) assets, _notes = locate(post_id) if assets is None: raise SystemExit(f"no assets thread for {post_id}") @@ -245,6 +269,7 @@ def cmd_merge(post_id: str) -> int: 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) @@ -301,6 +326,7 @@ def cmd_sync(handle: str | None) -> int: def cmd_publish(post_id: str) -> int: + check_frozen(post_id) _assets, notes = locate(post_id) if notes is None: raise SystemExit(f"no vault thread for {post_id}") @@ -321,7 +347,113 @@ def cmd_publish(post_id: str) -> int: return 0 -def main(argv: list[str] | None = None) -> int: +def _now_iso() -> str: + from datetime import datetime, timezone + + return ( + datetime.now(timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def cmd_migrate_media(args) -> int: + if bool(args.id) == bool(args.all_root): + raise SystemExit("migrate-media requires exactly one of --id or --all") + if args.id: + asset = DUMPS / args.id / "thread_data.json" + if not asset.is_file(): + scratch = SCRATCH / f"refetch_{args.id}" + if not (scratch / "thread_data.json").is_file(): + raise SystemExit(f"no thread_data.json for {args.id}") + assets, note_dir = locate(args.id) + if assets is None or note_dir is None: + raise SystemExit(f"no vault thread for {args.id}") + check_frozen(args.id) + result = migrate_legacy_thread(assets, note_dir, now=_now_iso(), apply=args.apply) + print(result) + return 0 if result.state != "blocked" else 2 + if not args.apply: + print("dry-run: explicit --apply required to write") + return 0 + print("corpus migrate-media --apply not yet implemented; awaiting Task 11") + return 0 + + +def cmd_audit_media(args) -> int: + assets, note_dir = locate(args.id) + if assets is None or note_dir is None: + raise SystemExit(f"no vault thread for {args.id}") + report = audit_thread(assets, note_dir, load_frozen_ids(FROZEN)) + if report.issues: + for issue in report.issues: + print(issue) + return 1 + print("audit ok") + return 0 + + +def cmd_backup(args) -> int: + check_frozen(args.id) + assets, note_dir = locate(args.id) + if assets is None or note_dir is None: + raise SystemExit(f"no vault thread for {args.id}") + destination_root = load_destination_root(VAULT, args.destination) + result = backup_thread( + assets, + assets_root=VAULT / "assets" / "threads", + destination_root=destination_root, + destination_id=args.destination, + now=_now_iso(), + require_destination_root=True, + ) + print(result) + return 0 if result.state == "synced" else 2 + + +def cmd_fallback(args) -> int: + if not args.confirm_origin_unavailable: + raise SystemExit("fallback requires --confirm-origin-unavailable") + check_frozen(args.id) + assets, note_dir = locate(args.id) + if assets is None or note_dir is None: + raise SystemExit(f"no vault thread for {args.id}") + userhash = load_userhash(VAULT) + assets_root = VAULT / "assets" / "threads" + + def _upload(path: Path) -> str: + return upload_file(path, userhash) + + def _lookup(provider: str, content_hash: str) -> str | None: + return find_existing_fallback(assets_root, provider, content_hash) + + result = activate_fallback( + assets, + note_dir, + media_id=args.media_id, + role=args.role, + provider=args.provider, + confirm_origin_unavailable=True, + now=_now_iso(), + upload=_upload, + lookup=_lookup, + ) + print(result) + return 0 + + +def cmd_restore_origin(args) -> int: + check_frozen(args.id) + assets, note_dir = locate(args.id) + if assets is None or note_dir is None: + raise SystemExit(f"no vault thread for {args.id}") + restore_origin(assets, note_dir, media_id=args.media_id, now=_now_iso()) + print("restored origin") + return 0 + + +def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="cmd", required=True) for name in ( @@ -352,6 +484,51 @@ def main(argv: list[str] | None = None) -> int: "sync", help="rebuild handle-index wikilinks from actual folder listing" ) sync.add_argument("--handle", default=None, help="single handle, else all") + + migrate = sub.add_parser( + "migrate-media", help="convert legacy media.json to canonical locations" + ) + migrate.add_argument("--id", default=None) + migrate.add_argument("--all", dest="all_root", action="store_true") + migrate.add_argument("--apply", action="store_true") + + audit = sub.add_parser( + "audit-media", help="local integrity and reference check" + ) + audit.add_argument("--id", required=True) + audit.add_argument("--check-origin", action="store_true") + audit.add_argument("--record-checks", action="store_true") + + backup_p = sub.add_parser( + "backup", help="sparse, hash-verified copy of one thread asset dir" + ) + backup_p.add_argument("--id", required=True) + backup_p.add_argument("--destination", default="cozy") + + fallback = sub.add_parser( + "fallback", help="upload one confirmed fallback host selection" + ) + fallback.add_argument("--id", required=True) + fallback.add_argument("--media-id", required=True) + fallback.add_argument("--role", default="orig") + fallback.add_argument( + "--provider", default="catbox", choices=("catbox",) + ) + fallback.add_argument( + "--confirm-origin-unavailable", action="store_true" + ) + + restore = sub.add_parser( + "restore-origin", help="select the immutable origin again" + ) + restore.add_argument("--id", required=True) + restore.add_argument("--media-id", required=True) + + return parser + + +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": @@ -374,6 +551,16 @@ def main(argv: list[str] | None = None) -> int: 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}")