more script features

scripts/twitter/README.md

@@ -25,7 +25,7 @@ python scripts/twitter/tw.py relabel --all --apply
 
 ## Capture
 
-`refresh` is refetch + ingest + emit `--force` + media merge. `--tip` treats `--id` as the tip and walks back to the opening post. If that tip is the OP, same-handle self-replies stay on the spine and foreign replies are branches. Repeat `--branch <tip-id>` to capture extra tips into the same `thread_data.json`. It does not invent replies. Captures run one at a time. gallery-dl retries are off. Extractor and request sleeps are five seconds. Re-emit deletes branch notes that are no longer roots.
+`refresh` is refetch + ingest + emit `--force` + media merge. `--tip` treats `--id` as the tip and walks back to the opening post. If that tip is the OP, same-handle self-replies stay on the spine and foreign replies are branches. Repeat `--branch <tip-id>` to capture extra tips into the same `thread_data.json`. `--attach <child>:<parent>` sets one hop before the walk. A walk that stops on a missing parent aborts unless `--allow-broken-walk`. A leftover dir under another handle aborts unless `--retire-old-dir`. It does not invent replies. Captures run one at a time. gallery-dl retries are off. Extractor and request sleeps are five seconds. Re-emit deletes branch notes that are no longer roots.
 
 Each thread writes one archive folder. The owner is the opening post (`reply_to_id` is empty). Other authors stay in that folder. New notes set `draft: false`. The command does not commit. Frozen ids abort.
 

scripts/twitter/emit_archive.py

@@ -9,7 +9,7 @@ import os
 import re
 import shutil
 import sys
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass, replace
 from datetime import date
 from pathlib import Path
 from urllib.parse import parse_qs, unquote, urlparse
@@ -463,6 +463,89 @@ def wiki_branch(handle: str, dir_name: str, name: str) -> str:
     return f"archive/threads/{handle}/{dir_name}/{name}"
 
 
+def parse_attach(raw: str) -> tuple[str, str]:
+    """Parse ``child:parent``. Ids have no colon."""
+    if raw.count(":") != 1:
+        raise SystemExit(f"attach must be child:parent, got {raw}")
+    child_id, parent_id = raw.split(":")
+    if not child_id or not parent_id:
+        raise SystemExit(f"attach must be child:parent, got {raw}")
+    return child_id, parent_id
+
+
+def apply_attaches(
+    thread: ThreadData,
+    attaches: tuple[tuple[str, str], ...],
+) -> ThreadData:
+    """Set each child's ``reply_to_id`` to the named parent. Both ids must exist."""
+    by_post = {post.post_id: post for post in thread.posts}
+    for child_id, parent_id in attaches:
+        if child_id not in by_post:
+            raise SystemExit(f"attach child missing: {child_id}")
+        if parent_id not in by_post:
+            raise SystemExit(f"attach parent missing: {parent_id}")
+        by_post[child_id] = replace(by_post[child_id], reply_to_id=parent_id)
+        print(f"attach {child_id} -> {parent_id}")
+    return ThreadData(
+        root_post_id=thread.root_post_id,
+        posts=tuple(by_post[post.post_id] for post in thread.posts),
+        source_url=thread.source_url,
+    )
+
+
+def write_thread_json(path: Path, thread: ThreadData) -> None:
+    """Write one thread_data.json from a typed thread."""
+    payload = {
+        "root_post_id": thread.root_post_id,
+        "source_url": thread.source_url,
+        "posts": [asdict(post) for post in thread.posts],
+    }
+    path.write_text(
+        json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+
+
+def foreign_owner_dirs(
+    thread: ThreadData,
+    handle: str,
+    spines: dict[str, Path],
+) -> tuple[Path, ...]:
+    """Archive dirs that hold this thread's posts under a different handle."""
+    found: list[Path] = []
+    seen: set[Path] = set()
+    for post in thread.posts:
+        candidate = spines.get(post.post_id)
+        if candidate is None or candidate.parent.name == handle:
+            continue
+        if candidate in seen:
+            continue
+        seen.add(candidate)
+        found.append(candidate)
+    return tuple(found)
+
+
+def rewrite_handle_index_from_dirs(path: Path) -> None:
+    """Rebuild handle-index wikilinks from remaining thread folders."""
+    if not path.is_file():
+        return
+    handle = path.parent.name
+    real = sorted(item.name for item in path.parent.iterdir() if item.is_dir())
+    text = path.read_text(encoding="utf-8")
+    fm, body = _split_frontmatter(text)
+    kept = [
+        line for line in body.splitlines() if not _WIKILINK_ITEM.match(line)
+    ]
+    links = [f"- [[{wiki_thread(handle, name)}]]" for name in real]
+    parts = [line for line in kept if line.strip()] + links
+    path.write_text(
+        fm.rstrip() + "\n\n" + "\n".join(parts) + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+
+
 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"
@@ -556,6 +639,9 @@ def emit(
     tip: str | None = None,
     reconcile_scratch: Path = SCRATCH,
     frozen_ids: set[str] | None = None,
+    attaches: tuple[tuple[str, str], ...] = (),
+    allow_broken_walk: bool = False,
+    retire_old_dir: bool = False,
 ) -> EmitResult:
     src = input_dir / "thread_data.json"
     if not src.is_file():
@@ -564,10 +650,20 @@ def emit(
     thread = load_thread(src)
     if not thread.posts:
         raise SystemExit(f"no posts in {src}")
+    if attaches:
+        thread = apply_attaches(thread, attaches)
+        write_thread_json(src, thread)
 
     if tip:
         spine = spine_from_tip(thread, tip)
         input_kind = "tip"
+        print("spine-walk " + " <- ".join(reversed(spine)))
+        missing = by_id(thread)[spine[0]].reply_to_id
+        if missing:
+            message = f"spine-walk stops missing_parent={missing}"
+            if not allow_broken_walk:
+                raise SystemExit(message)
+            print(message)
     else:
         spine = spine_ids(thread)
         input_kind = "root"
@@ -579,8 +675,24 @@ def emit(
     rendered_title = title_text(first.text)
     desired_dir_name = thread_dir_name(date_prefix(first.timestamp), rendered_title, slug)
 
-    if reuse_dir is None and force:
+    _existing_ids, spines = collect_existing_ids(vault)
+    leftovers = foreign_owner_dirs(thread, handle, spines)
+    if leftovers:
+        listed = ", ".join(str(path) for path in leftovers)
+        if not retire_old_dir:
+            raise SystemExit(f"leftover dir {listed}; pass --retire-old-dir")
+        for old in leftovers:
+            asset_old = (
+                vault / "assets" / "threads" / old.parent.name / old.name
+            )
+            shutil.rmtree(old)
+            if asset_old.is_dir():
+                shutil.rmtree(asset_old)
+            rewrite_handle_index_from_dirs(old.parent / "index.md")
+            print(f"retire-old-dir {old}")
         _existing_ids, spines = collect_existing_ids(vault)
+
+    if reuse_dir is None and force:
         # 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-
@@ -1461,6 +1573,23 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         default=None,
         help="Climb reply_to from this post_id for the spine (same handle)",
     )
+    parser.add_argument(
+        "--attach",
+        action="append",
+        default=[],
+        metavar="CHILD:PARENT",
+        help="set child.reply_to_id to parent before the spine walk; repeatable",
+    )
+    parser.add_argument(
+        "--allow-broken-walk",
+        action="store_true",
+        help="emit even if the tip walk stops on a missing parent",
+    )
+    parser.add_argument(
+        "--retire-old-dir",
+        action="store_true",
+        help="delete archive/asset dirs for this thread under another handle",
+    )
     return parser.parse_args(argv)
 
 
@@ -1485,6 +1614,9 @@ def main(argv: list[str] | None = None) -> int:
         args.archived,
         force=args.force,
         tip=args.tip,
+        attaches=tuple(parse_attach(raw) for raw in args.attach),
+        allow_broken_walk=args.allow_broken_walk,
+        retire_old_dir=args.retire_old_dir,
     )
     return 0
 

scripts/twitter/tests/test_emit_attach.py

@@ -0,0 +1,199 @@
+"""Emit --attach, broken-walk abort, and leftover-dir retire."""
+
+from __future__ import annotations
+
+import json
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.emit_archive import emit
+from twitter.models import PostData, PostMetrics, ThreadData
+
+
+def post(
+    post_id: str,
+    handle: str,
+    text: str,
+    reply_to: str | None,
+    timestamp: str = "2025-02-25 12:00:00",
+) -> dict:
+    return {
+        "post_id": post_id,
+        "author": handle,
+        "handle": handle,
+        "text": text,
+        "timestamp": timestamp,
+        "media_urls": [],
+        "reply_to_id": reply_to,
+        "quote_of_id": None,
+        "metrics": {
+            "reply_count": 0,
+            "repost_count": 0,
+            "like_count": 0,
+            "view_count": 0,
+        },
+    }
+
+
+class EmitAttachTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp(prefix="emit_attach_"))
+        self.input_dir = self.tmp / "input"
+        self.input_dir.mkdir()
+        self.vault = self.tmp / "vault"
+        (self.vault / "archive" / "threads").mkdir(parents=True)
+        (self.vault / "assets" / "threads").mkdir(parents=True)
+        self._write(
+            [
+                post("1", "levelsio", "IT WORKS", None, "2025-02-25 16:00:00"),
+                post(
+                    "2",
+                    "Jonathan_Blow",
+                    "overview",
+                    "deleted",
+                    "2025-02-25 22:00:00",
+                ),
+                post(
+                    "3",
+                    "Jonathan_Blow",
+                    "udp nat",
+                    "2",
+                    "2025-02-25 22:10:00",
+                ),
+            ]
+        )
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _write(self, posts: list[dict]) -> None:
+        (self.input_dir / "thread_data.json").write_text(
+            json.dumps(
+                {
+                    "root_post_id": "3",
+                    "source_url": "https://x.com/i/status/3",
+                    "posts": posts,
+                },
+                indent=2,
+            ),
+            encoding="utf-8",
+            newline="\n",
+        )
+
+    def test_attach_walks_tip_to_op(self) -> None:
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-25",
+            force=True,
+            tip="3",
+            attaches=(("2", "1"),),
+        )
+        note = (
+            self.vault
+            / "archive"
+            / "threads"
+            / "levelsio"
+            / "2025-02-25-it-works"
+            / "index.md"
+        )
+        text = note.read_text(encoding="utf-8")
+        self.assertIn("**1/** **@levelsio** ^1", text)
+        self.assertIn("**2/** **@Jonathan_Blow** ^2", text)
+        self.assertIn("**3/** **@Jonathan_Blow** ^3", text)
+
+    def test_broken_walk_aborts_without_override(self) -> None:
+        with self.assertRaisesRegex(SystemExit, "missing_parent=deleted"):
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug=None,
+                archived="2026-08-25",
+                force=True,
+                tip="3",
+            )
+
+    def test_allow_broken_walk_emits_short_spine(self) -> None:
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-25",
+            force=True,
+            tip="3",
+            allow_broken_walk=True,
+        )
+        note = (
+            self.vault
+            / "archive"
+            / "threads"
+            / "Jonathan_Blow"
+            / "2025-02-25-overview"
+            / "index.md"
+        )
+        text = note.read_text(encoding="utf-8")
+        self.assertIn("**1/** **@Jonathan_Blow** ^2", text)
+        self.assertNotIn("@levelsio", text)
+
+    def test_leftover_dir_aborts_without_override(self) -> None:
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-25",
+            force=True,
+            tip="3",
+            allow_broken_walk=True,
+        )
+        with self.assertRaisesRegex(SystemExit, "leftover dir"):
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug=None,
+                archived="2026-08-25",
+                force=True,
+                tip="3",
+                attaches=(("2", "1"),),
+            )
+
+    def test_retire_old_dir_removes_foreign_handle(self) -> None:
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-25",
+            force=True,
+            tip="3",
+            allow_broken_walk=True,
+        )
+        old = (
+            self.vault
+            / "archive"
+            / "threads"
+            / "Jonathan_Blow"
+            / "2025-02-25-overview"
+        )
+        self.assertTrue(old.is_dir())
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-25",
+            force=True,
+            tip="3",
+            attaches=(("2", "1"),),
+            retire_old_dir=True,
+        )
+        self.assertFalse(old.is_dir())
+        note = (
+            self.vault
+            / "archive"
+            / "threads"
+            / "levelsio"
+            / "2025-02-25-it-works"
+            / "index.md"
+        )
+        self.assertTrue(note.is_file())

scripts/twitter/tests/test_emit_op_owned.py

@@ -487,34 +487,21 @@ class OrphansNotReusedTests(unittest.TestCase):
             newline="\n",
         )
 
-        emit(
-            input_dir=self.input_dir,
-            vault=self.vault,
-            slug=None,
-            archived="2026-08-24",
-            force=True,
-            tip="1651282559287042048",
-        )
+        with self.assertRaisesRegex(SystemExit, "leftover dir"):
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug=None,
+                archived="2026-08-24",
+                force=True,
+                tip="1651282559287042048",
+            )
 
         archive_root = self.vault / "archive" / "threads"
-        # The OP-owned archive lives under kenpex, not NOTimothyLottes
-        self.assertTrue(
-            (archive_root / "kenpex").is_dir(),
-            "OP-owned archive must live under the OP handle's dir",
-        )
-        kenpex_thread_dirs = sorted(
-            d.name
-            for d in (archive_root / "kenpex").iterdir()
-            if d.is_dir()
-        )
-        self.assertEqual(len(kenpex_thread_dirs), 1)
-
-        # The legacy NOTimothyLottes dir is left as orphan; this
-        # test does not auto-remove it (manual cleanup is operator
-        # responsibility).
         self.assertTrue(
             (archive_root / "NOTimothyLottes" / "2023-04-26-legacy-slug").is_dir(),
         )
+        self.assertFalse((archive_root / "kenpex").exists())
 
 
 class SpinePrefersSameAuthorTests(unittest.TestCase):

scripts/twitter/tests/test_tw_media_commands.py

@@ -387,6 +387,26 @@ class TwMediaCommandTests(unittest.TestCase):
         self.assertTrue(emit_args.preserve_existing)
         self.assertTrue(refresh_args.preserve_existing)
 
+    def test_emit_and_refresh_accept_attach_and_overrides(self) -> None:
+        emit_args = build_parser().parse_args([
+            "emit", "--id", "3", "--tip",
+            "--attach", "2:1",
+            "--allow-broken-walk",
+            "--retire-old-dir",
+        ])
+        refresh_args = build_parser().parse_args([
+            "refresh", "--id", "3", "--tip",
+            "--attach", "2:1",
+            "--allow-broken-walk",
+            "--retire-old-dir",
+        ])
+        self.assertEqual(emit_args.attach, ["2:1"])
+        self.assertTrue(emit_args.allow_broken_walk)
+        self.assertTrue(emit_args.retire_old_dir)
+        self.assertEqual(refresh_args.attach, ["2:1"])
+        self.assertTrue(refresh_args.allow_broken_walk)
+        self.assertTrue(refresh_args.retire_old_dir)
+
     def test_emit_preserve_existing_merges_before_running_emitter(self) -> None:
         with TemporaryDirectory() as tmp:
             root = Path(tmp)

scripts/twitter/tw.py

@@ -537,6 +537,12 @@ def cmd_emit(args: argparse.Namespace) -> int:
         cmd.extend(["--tip", post_id])
     if slug:
         cmd.extend(["--slug", slug])
+    for raw in getattr(args, "attach", []) or []:
+        cmd.extend(["--attach", raw])
+    if getattr(args, "allow_broken_walk", False):
+        cmd.append("--allow-broken-walk")
+    if getattr(args, "retire_old_dir", False):
+        cmd.append("--retire-old-dir")
     run(cmd)
     # Cross-author threads emit one asset dir per author. The tip's
     # post_id is only in the tip's author's asset dir; the other
@@ -870,6 +876,23 @@ def build_parser() -> argparse.ArgumentParser:
                 action="store_true",
                 help="retain archived posts omitted by the fresh capture",
             )
+            p.add_argument(
+                "--attach",
+                action="append",
+                default=[],
+                metavar="CHILD:PARENT",
+                help="set child.reply_to_id to parent before the spine walk",
+            )
+            p.add_argument(
+                "--allow-broken-walk",
+                action="store_true",
+                help="emit even if the tip walk stops on a missing parent",
+            )
+            p.add_argument(
+                "--retire-old-dir",
+                action="store_true",
+                help="delete this thread's archive under another handle",
+            )
         if name == "lift":
             p.add_argument("--orig", action="store_true")
         if name == "ocr":