"""Emit Threadwell archive notes from a thread_data.json dump. No fetch, no lift.""" from __future__ import annotations import argparse import json import logging import re import shutil import sys from dataclasses import dataclass 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, merge_manifest_items, new_original_item, selected_url, ) from twitter.media_refs import remote_markup 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 ( by_id, branch_roots, children_map, descendants, spine_from_tip, spine_ids, ) 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, merge_manifest_items, new_original_item, selected_url, ) from media_refs import remote_markup 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 ( by_id, branch_roots, children_map, descendants, spine_from_tip, spine_ids, ) 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) if end < 0: return [], text return text[4:end].splitlines(), text[end + 4 :] 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: 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[MediaItem]) -> list[str]: missing: list[str] = [] for item in items: local = find_location(item, "local") 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.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 branch_count: int missing_media_count: int 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: return fmt.lstrip(".").lower() suffix = Path(unquote(parsed.path)).suffix if suffix: return suffix.lstrip(".").lower() return None def _media_files(media_dir: Path) -> list[Path]: if not media_dir.is_dir(): return [] return sorted(p for p in media_dir.iterdir() if p.is_file()) 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: return path return 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): return path return None def collect_media( thread: ThreadData, input_dir: Path, dest_dir: Path, *, now: str, ) -> 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[MediaItem] = [] for post in thread.posts: for index, url in enumerate(post.media_urls, start=1): media_id = _url_media_id(url, index) src = _take_preferred(unused, post.post_id, index) if src is None: src = _take_fallback(unused, post.post_id) if src is not None: unused.remove(src) ext = src.suffix.lstrip(".") or _url_ext(url) or "bin" 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, 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 def selected_media_by_post( 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.embed: continue url = selected_url(item) if url is not None: 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[MediaItem], ) -> list[MediaItem]: """Merge existing derived/fallback rows with fresh canonical items.""" old_path = dest_dir / "media.json" if not old_path.is_file(): return items try: old = json.loads(old_path.read_text(encoding="utf-8")) except json.JSONDecodeError: return items if old.get("schema_version") != 2: raise SystemExit( "legacy media.json requires: tw.py migrate-media --id --apply" ) existing = _from_wire_dict(old).items return list(merge_manifest_items(existing, tuple(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]: ids = by_id(thread) found: list[str] = [] seen: set[str] = set() for post in thread.posts: parent = post.reply_to_id if parent is None or parent in ids or parent in seen: continue seen.add(parent) found.append(parent) return found def empty_text_ids(thread: ThreadData) -> list[str]: return [p.post_id for p in thread.posts if not (p.text or "").strip()] 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] missing = [ p.post_id for p in thread.posts if p.reply_to_id is not None and p.reply_to_id not in ids ] empty = empty_text_ids(thread) lines = [ "# gaps", f"input: {input_kind}", f"suggested_tip: {suggested}", ] if quote_ids: lines.append("quote_of:") lines.extend(quote_ids) else: lines.append("quote_of: unset") lines.append("missing_reply_to:") lines.extend(missing) lines.append("empty_text:") lines.extend(empty) return "\n".join(lines) + "\n" 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:] if rest.startswith("\n"): rest = rest[1:] end = rest.find("\n---") if end < 0: return "" return rest[:end] 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: return match.group(1) return None def collect_existing_ids(vault: Path) -> tuple[set[str], dict[str, Path]]: """Return every archive post_id and spine dirs keyed by spine post_id.""" threads = vault / "archive" / "threads" ids: set[str] = set() spines: dict[str, Path] = {} if not threads.is_dir(): return ids, spines for md in threads.rglob("*.md"): pid = _frontmatter_post_id(md.read_text(encoding="utf-8")) if not pid: continue ids.add(pid) if md.name == "index.md" and md.parent.parent.parent == threads: spines[pid] = md.parent return ids, spines 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 while (parent / f"{base}-{n}").exists(): n += 1 return f"{base}-{n}" 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) if end < 0: return "", text closer = end + 4 fm = text[:closer] body = text[closer:] if body.startswith("\n"): fm += "\n" body = body[1:] elif not fm.endswith("\n"): fm += "\n" return fm, body def _upsert_wikilink(path: Path, target: str) -> None: """Insert or move a wikilink target on the handle index. If the handle folder contains a directory whose name matches `target`, any other wikilink whose basename does not match a real directory is dropped. This keeps the handle index in sync when a folder is reused with a new slug (e.g., a tip-climb refresh renames the folder). """ text = path.read_text(encoding="utf-8") fm, body = _split_frontmatter(text) if not fm.strip(): # Frontmatter got nuked by a hand edit. Don't silently emit a # bare wikilink list — fall back to a fresh standard frontmatter # so the file remains a complete page. handle = path.parent.name fm = ( f"---\n" f"title: {handle}\n" f"type: note\n" f"draft: false\n" f"description: Archived threads by {handle}.\n" f"tags:\n" f" - archive\n" f" - twitter\n" f" - {handle}\n" f"---\n" ) kept: list[str] = [] names: list[str] = [] handle_dir = path.parent real_dirs = {d.name for d in handle_dir.iterdir() if d.is_dir()} if handle_dir.is_dir() else set() for line in body.splitlines(): match = _WIKILINK_ITEM.match(line) if match: name = match.group(1) # wikilinks are vault-root paths; compare the basename to real_dirs if name.rsplit("/", 1)[-1] in real_dirs or not real_dirs: if name not in names: names.append(name) else: kept.append(line) if target not in names: names.append(target) names.sort() while kept and kept[0] == "": kept.pop(0) while kept and kept[-1] == "": kept.pop() parts: list[str] = [] if kept: parts.extend(kept) parts.append("") parts.extend(f"- [[{name}]]" for name in names) parts.append("") body_text = "\n".join(parts) path.write_text(fm.rstrip() + "\n\n" + body_text, encoding="utf-8", newline="\n") 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(): front = ( "---\n" f"title: {handle}\n" "type: note\n" "draft: false\n" f"description: Archived threads by {handle}.\n" "tags:\n" " - archive\n" " - twitter\n" f" - {handle}\n" "---\n" ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text( front + f"\n- [[{target}]]\n", encoding="utf-8", newline="\n", ) return _upsert_wikilink(path, target) 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) path.write_text( ( "---\n" "title: Threads\n" "type: note\n" "draft: false\n" "description: Archived Twitter/X threads and similar sequential posts.\n" "tags:\n" " - archive\n" "---\n" "\n" "Threads are filed by author handle, then by date and title. " "The main note in each folder is that author's own chain. " "Replies live beside it.\n" "\n" "How to read a folder: [[How threads are organized]].\n" "\n" "Example of the note shape (single spine, no branches): " "[[How to archive a thread]]\n" "\n" f"- [[archive/threads/{handle}]]\n" ), encoding="utf-8", newline="\n", ) return _upsert_wikilink(path, f"archive/threads/{handle}") 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() ] dumps.sort(key=lambda path: path.name) return dumps def _gap_line( post_id: str, handle: str, missing: list[str], empty: list[str], ) -> str: miss = ",".join(missing) if missing else "-" empt = ",".join(empty) if empty else "-" return f"{post_id} {handle} root {miss} {empt}" def emit( input_dir: Path, vault: Path, slug: str | None, archived: str, *, force: bool = False, reuse_dir: Path | None = None, tip: str | None = None, ) -> EmitResult: src = input_dir / "thread_data.json" if not src.is_file(): raise SystemExit(f"missing {src}") thread = load_thread(src) if not thread.posts: raise SystemExit(f"no posts in {src}") if tip: spine = spine_from_tip(thread, tip) input_kind = "tip" else: spine = spine_ids(thread) input_kind = "root" roots = branch_roots(thread, spine) ids = by_id(thread) kids = children_map(thread) first = ids[spine[0]] handle = first.handle first_line = (first.text or "").split("\n", 1)[0] if reuse_dir is None and force: _existing_ids, spines = collect_existing_ids(vault) # Only accept a candidate whose parent handle matches the # OP handle. A thread previously archived at a different # author's dir (e.g. an old tip-as-root archive of a cross- # author conversation) is treated as orphan, not a reuse. candidate = spines.get(thread.root_post_id) if candidate is not None and candidate.parent.name == handle: reuse_dir = candidate if reuse_dir is None: for post in thread.posts: candidate = spines.get(post.post_id) if candidate is not None and candidate.parent.name == handle: reuse_dir = candidate break if reuse_dir is None and slug: candidate = vault / "archive" / "threads" / handle / slug if candidate.is_dir(): reuse_dir = candidate if reuse_dir is not None: dir_name = reuse_dir.name handle = reuse_dir.parent.name note_dir = reuse_dir asset_dir = vault / "assets" / "threads" / handle / dir_name else: dir_name = thread_dir_name(date_prefix(first.timestamp), first_line, slug) if slug is None: dir_name = unique_dir_name( vault / "archive" / "threads" / handle, dir_name, ) note_dir = vault / "archive" / "threads" / handle / dir_name asset_dir = vault / "assets" / "threads" / handle / dir_name branch_names: dict[str, str] = {} for rid in roots: branch = ids[rid] fname = branch_file_name( date_prefix(branch.timestamp), branch.handle, branch.text, ) name = fname[:-3] if fname.endswith(".md") else fname branch_names[rid] = name note_dir.mkdir(parents=True, exist_ok=True) asset_dir.mkdir(parents=True, exist_ok=True) 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() } 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, rid, descendants(rid, kids), first.handle, archived, media_by_post, 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": [_item_to_wire(item) for item in items], "mirrors": [], } 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( gaps_text, encoding="utf-8", newline="\n", ) ensure_handle_index(vault, handle, dir_name) ensure_threads_index(vault, handle) return EmitResult( post_id=thread.root_post_id, handle=handle, dir_name=dir_name, branch_count=len(roots), missing_media_count=len(missing_keys), ) def emit_all( dumps_root: Path, vault: Path, archived: str, force: bool, ) -> tuple[int, int, int, int]: dumps = discover_dumps(dumps_root) existing_ids, spines = collect_existing_ids(vault) lines: list[str] = [] skipped = 0 written = 0 errors = 0 for dump in dumps: src = dump / "thread_data.json" try: thread = load_thread(src) if not thread.posts: raise ValueError(f"no posts in {src}") spine = spine_ids(thread) handle = by_id(thread)[spine[0]].handle missing = missing_parent_ids(thread) empty = empty_text_ids(thread) post_id = thread.root_post_id except Exception as exc: logger.warning("emit_all: skipping %s: %s", dump, exc) lines.append(f"{dump.name} ? error - -") errors += 1 continue lines.append(_gap_line(post_id, handle, missing, empty)) if post_id in existing_ids and not force: skipped += 1 continue reuse = spines.get(post_id) if force else None try: emit( dump, vault, None, archived, force=force, reuse_dir=reuse, ) except Exception as exc: logger.warning("emit_all: skipping %s: %s", dump, exc) errors += 1 continue written += 1 existing_ids.add(post_id) out = vault / "assets" / "threads" / "_batch_gaps.md" out.parent.mkdir(parents=True, exist_ok=True) out.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n") return len(dumps), skipped, written, errors def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Emit Threadwell archive notes from a thread_data.json dump. " "No fetch, no lift." ), ) source = parser.add_mutually_exclusive_group(required=True) source.add_argument( "--input", type=Path, help="Dump directory containing thread_data.json", ) source.add_argument( "--all", dest="all_root", type=Path, metavar="DUMPS_ROOT", help="Convert every dump directory under this root", ) parser.add_argument( "--vault", required=True, type=Path, help="Threadwell vault root", ) parser.add_argument( "--slug", default=None, help="Optional slug override for a single --input (ignored with --all)", ) parser.add_argument( "--archived", default=date.today().isoformat(), help="Archive date YYYY-MM-DD (default: today local)", ) parser.add_argument( "--force", action="store_true", help="Emit even if post_id already exists in the vault", ) parser.add_argument( "--tip", default=None, help="Climb reply_to from this post_id for the spine (same handle)", ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) if args.all_root is not None: seen, skipped, written, errors = emit_all( args.all_root, args.vault, args.archived, args.force, ) print( f"dumps seen={seen} skipped={skipped} " f"written={written} errors={errors}" ) return 0 if errors == 0 else 1 emit( args.input, args.vault, args.slug, args.archived, force=args.force, tip=args.tip, ) return 0 if __name__ == "__main__": raise SystemExit(main())