more branch scrape exploration.

scripts/twitter/README.md

@@ -8,6 +8,7 @@ From the vault root:
 python C:\projects\Threadwell\scripts\twitter\tw.py graph   --id <snowflake>
 python C:\projects\Threadwell\scripts\twitter\tw.py refresh --id <snowflake> --tip
 python C:\projects\Threadwell\scripts\twitter\tw.py refresh --id <spine-tip> --tip --branch <branch-tip>
+python C:\projects\Threadwell\scripts\twitter\tw.py add-branch --id <original-spine-tip> --from <reply-node>
 python C:\projects\Threadwell\scripts\twitter\tw.py audit-media --id <snowflake>
 python C:\projects\Threadwell\scripts\twitter\tw.py publish --id <snowflake>
 python C:\projects\Threadwell\scripts\twitter\tw.py sync    --handle <handle>
@@ -22,6 +23,8 @@ python C:\projects\Threadwell\scripts\twitter\tw.py migrate-media --all --apply
 
 `refresh` = refetch (cookies) + ingest + emit `--force` + media_merge. `--tip` treats `--id` as the tip and walks back to the OP regardless of handle. Repeatable `--branch <tip-id>` captures only those additional explicit tip paths and merges them into the same `thread_data.json`; it does not discover replies automatically. Captures run sequentially with gallery-dl retries disabled and five-second extractor/request pacing. Each thread emits exactly one archive directory owned by the OP (the post with `reply_to_id == None`); cross-author responders do not get a directory. Does not flip `draft`. Does not commit. Frozen ids abort.
 
+`add-branch` incrementally extends an already emitted thread. `--id` must be the original stored spine tip; repeatable `--from <reply-node>` values are each captured once. The command locally retains each node's missing attachment path and entire visible descendant subtree, merges only previously absent posts into the existing `thread_data.json`, downloads media for the new posts, and re-emits with the original spine unchanged. It preserves existing origin/local/fallback/derived/OCR media records and adds provider-origin plus local-copy records for new branch media. A visible leaf is only the deepest reply returned by that capture; the command does not query descendants individually or claim provider-wide completeness. Fallback upload remains a separate explicit action after origin unavailability is confirmed.
+
 Usual publish path after you like the note:
 
 1. `tw.py graph --id …` — confirm tip vs root.
@@ -43,7 +46,7 @@ Frozen threads (ids listed in `do_not_refetch.txt`, matched against every captur
 
 | Script | Job |
 |---|---|
-| `tw.py` | locate / graph / refetch / emit / refresh / ocr / merge / sync / publish / migrate-media / backup / fallback / restore-origin |
+| `tw.py` | locate / graph / refetch / add-branch / emit / refresh / ocr / merge / sync / publish / migrate-media / backup / fallback / restore-origin |
 | `graph_dry_run.py` | tip graph, no vault writes |
 | `ingest_gallery.py` | gallery-dl JSON → `thread_data.json` |
 | `emit_archive.py` | JSON → notes + assets. Uses canonical media locations. `--force` preserves derived/fallback rows |

scripts/twitter/tests/test_tw_media_commands.py

@@ -2,18 +2,51 @@ from __future__ import annotations
 
 import json
 import unittest
+from argparse import Namespace
+from contextlib import redirect_stdout
+from dataclasses import asdict, replace
+from io import StringIO
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from unittest.mock import patch
 
+from twitter import tw as tw_module
+from twitter.models import PostData, PostMetrics, ThreadData, load_thread
 from twitter.tw import (
     _capture_ids,
+    _copy_selected_media,
     _gallery_base_args,
+    _merge_branch_posts,
     _merge_gallery_files,
+    _select_branch_capture,
     _validate_capture_ids,
     build_parser,
+    cmd_add_branch,
 )
 
 
+def make_post(post_id: str, reply_to_id: str | None) -> PostData:
+    return PostData(
+        post_id=post_id,
+        author=post_id,
+        handle=post_id,
+        text=post_id,
+        timestamp=f"2026-01-01 00:00:{int(post_id) % 60:02d}",
+        media_urls=(),
+        reply_to_id=reply_to_id,
+        quote_of_id=None,
+        metrics=PostMetrics(0, 0, 0, None),
+    )
+
+
+def write_test_thread(path: Path, thread: ThreadData) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(asdict(thread), indent=2) + "\n",
+        encoding="utf-8",
+    )
+
+
 class TwMediaCommandTests(unittest.TestCase):
     def test_refetch_accepts_repeatable_branch(self) -> None:
         args = build_parser().parse_args([
@@ -82,6 +115,224 @@ class TwMediaCommandTests(unittest.TestCase):
             with self.assertRaisesRegex(SystemExit, "300"):
                 _validate_capture_ids(path, ("100", "300"))
 
+    def test_select_branch_capture_keeps_entire_visible_subtree(self) -> None:
+        existing = ThreadData(
+            root_post_id="100",
+            posts=(make_post("100", None),),
+            source_url="https://x.com/i/status/200",
+        )
+        captured = ThreadData(
+            root_post_id="300",
+            posts=(
+                make_post("100", None),
+                make_post("300", "100"),
+                make_post("400", "300"),
+                make_post("500", "300"),
+                make_post("600", "100"),
+            ),
+            source_url="https://x.com/i/status/300",
+        )
+        selected, leaves = _select_branch_capture(existing, captured, "300")
+        self.assertEqual(
+            tuple(post.post_id for post in selected),
+            ("300", "400", "500"),
+        )
+        self.assertEqual(leaves, ("400", "500"))
+
+    def test_select_branch_capture_keeps_missing_attachment_path(self) -> None:
+        existing = ThreadData(
+            root_post_id="100",
+            posts=(make_post("100", None),),
+            source_url="https://x.com/i/status/200",
+        )
+        captured = ThreadData(
+            root_post_id="300",
+            posts=(
+                make_post("100", None),
+                make_post("250", "100"),
+                make_post("300", "250"),
+                make_post("400", "300"),
+            ),
+            source_url="https://x.com/i/status/300",
+        )
+        selected, leaves = _select_branch_capture(existing, captured, "300")
+        self.assertEqual(
+            tuple(post.post_id for post in selected),
+            ("250", "300", "400"),
+        )
+        self.assertEqual(leaves, ("400",))
+
+    def test_select_branch_capture_rejects_unattached_capture(self) -> None:
+        existing = ThreadData(
+            root_post_id="100",
+            posts=(make_post("100", None),),
+            source_url="https://x.com/i/status/200",
+        )
+        captured = ThreadData(
+            root_post_id="999",
+            posts=(make_post("999", None), make_post("300", "999")),
+            source_url="https://x.com/i/status/300",
+        )
+        with self.assertRaisesRegex(SystemExit, "does not attach"):
+            _select_branch_capture(existing, captured, "300")
+
+    def test_merge_branch_posts_preserves_existing_and_first_new_post(self) -> None:
+        existing_100 = make_post("100", None)
+        existing_200 = make_post("200", "100")
+        existing = ThreadData(
+            root_post_id="100",
+            posts=(existing_100, existing_200),
+            source_url="https://x.com/i/status/200",
+        )
+        first_300 = make_post("300", "100")
+        second_300 = replace(first_300, text="replacement must not win")
+        merged, added = _merge_branch_posts(
+            existing,
+            [
+                (replace(existing_200, text="fresh duplicate"), first_300),
+                (second_300, make_post("400", "300")),
+            ],
+        )
+        self.assertEqual(merged.posts, (
+            existing_100,
+            existing_200,
+            first_300,
+            make_post("400", "300"),
+        ))
+        self.assertEqual(merged.root_post_id, existing.root_post_id)
+        self.assertEqual(merged.source_url, existing.source_url)
+        self.assertEqual(added, ("300", "400"))
+
+    def test_copy_selected_media_copies_only_selected_post_files(self) -> None:
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "source"
+            destination = root / "destination"
+            (source / "media").mkdir(parents=True)
+            (destination / "media").mkdir(parents=True)
+            (source / "media" / "300_1.jpg").write_bytes(b"300")
+            (source / "media" / "400_1.png").write_bytes(b"400")
+            (source / "media" / "600_1.jpg").write_bytes(b"600")
+            (destination / "media" / "existing.jpg").write_bytes(b"old")
+            copied = _copy_selected_media(
+                source,
+                destination,
+                {"300", "400"},
+            )
+            self.assertEqual(
+                tuple(path.name for path in copied),
+                ("300_1.jpg", "400_1.png"),
+            )
+            self.assertEqual(
+                sorted(path.name for path in (destination / "media").iterdir()),
+                ["300_1.jpg", "400_1.png", "existing.jpg"],
+            )
+
+    def test_add_branch_accepts_repeatable_from_nodes(self) -> None:
+        args = build_parser().parse_args([
+            "add-branch",
+            "--id", "200",
+            "--from", "300",
+            "--from", "400",
+        ])
+        self.assertEqual(args.id, "200")
+        self.assertEqual(args.from_ids, ["300", "400"])
+
+    def test_add_branch_stages_selected_subtree_and_keeps_spine_tip(self) -> None:
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            assets = root / "assets"
+            notes = root / "notes"
+            notes.mkdir()
+            existing = ThreadData(
+                root_post_id="200",
+                posts=(make_post("100", None), make_post("200", "100")),
+                source_url="https://x.com/i/status/200",
+            )
+            existing_path = assets / "thread_data.json"
+            write_test_thread(existing_path, existing)
+            existing_bytes = existing_path.read_bytes()
+
+            branch_scratch = root / "refetch_300"
+            captured = ThreadData(
+                root_post_id="300",
+                posts=(
+                    make_post("100", None),
+                    make_post("300", "100"),
+                    make_post("400", "300"),
+                    make_post("500", "300"),
+                    make_post("600", "100"),
+                ),
+                source_url="https://x.com/i/status/300",
+            )
+            write_test_thread(branch_scratch / "thread_data.json", captured)
+            (branch_scratch / "media").mkdir()
+            for post_id in ("300", "400", "500", "600"):
+                (branch_scratch / "media" / f"{post_id}_1.jpg").write_bytes(
+                    post_id.encode("ascii"),
+                )
+
+            def fake_scratch(post_id: str) -> Path:
+                return root / f"refetch_{post_id}"
+
+            with (
+                patch.object(tw_module, "locate", return_value=(assets, notes)),
+                patch.object(tw_module, "scratch_dir", side_effect=fake_scratch),
+                patch.object(tw_module, "check_frozen"),
+                patch.object(tw_module, "cmd_refetch") as refetch,
+                patch.object(tw_module, "cmd_emit") as emit,
+            ):
+                with redirect_stdout(StringIO()):
+                    result = cmd_add_branch(
+                        Namespace(id="200", from_ids=["300"]),
+                    )
+
+            self.assertEqual(result, 0)
+            refetch_args = refetch.call_args.args[0]
+            self.assertEqual(refetch_args.id, "300")
+            self.assertEqual(refetch_args.branch, [])
+            emit_args = emit.call_args.args[0]
+            self.assertEqual(emit_args.id, "200")
+            self.assertTrue(emit_args.tip)
+            self.assertIsNone(emit_args.slug)
+
+            staged = load_thread(root / "refetch_200" / "thread_data.json")
+            self.assertEqual(
+                tuple(post.post_id for post in staged.posts),
+                ("100", "200", "300", "400", "500"),
+            )
+            self.assertEqual(
+                sorted(path.name for path in (root / "refetch_200" / "media").iterdir()),
+                ["300_1.jpg", "400_1.jpg", "500_1.jpg"],
+            )
+            self.assertEqual(existing_path.read_bytes(), existing_bytes)
+
+    def test_add_branch_rejects_id_that_is_not_stored_spine_tip(self) -> None:
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            assets = root / "assets"
+            notes = root / "notes"
+            notes.mkdir()
+            write_test_thread(
+                assets / "thread_data.json",
+                ThreadData(
+                    root_post_id="100",
+                    posts=(make_post("100", None), make_post("200", "100")),
+                    source_url="https://x.com/i/status/100",
+                ),
+            )
+            with (
+                patch.object(tw_module, "locate", return_value=(assets, notes)),
+                patch.object(tw_module, "check_frozen"),
+                patch.object(
+                    tw_module,
+                    "cmd_refetch",
+                    side_effect=AssertionError("must reject before fetch"),
+                ),
+            ):
+                with self.assertRaisesRegex(SystemExit, "stored spine tip"):
+                    cmd_add_branch(Namespace(id="200", from_ids=["300"]))
+
     def test_migrate_defaults_to_dry_run(self) -> None:
         args = build_parser().parse_args(["migrate-media", "--id", "100"])
         self.assertEqual(args.cmd, "migrate-media")

scripts/twitter/tw.py

@@ -4,6 +4,7 @@ From the vault root:
 
   python scripts/twitter/tw.py graph --id 1692565070583136348
   python scripts/twitter/tw.py refresh --id 1692565070583136348 --tip
+  python scripts/twitter/tw.py add-branch --id 1692565070583136348 --from <reply-node>
   python scripts/twitter/tw.py lift --id 1692565070583136348 --orig
   python scripts/twitter/tw.py ocr --id 1692565070583136348
   python scripts/twitter/tw.py locate --id 1692565070583136348
@@ -16,9 +17,11 @@ from __future__ import annotations
 
 import argparse
 import json
+import shutil
 import subprocess
 import sys
 from collections.abc import Callable
+from dataclasses import asdict
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -27,7 +30,7 @@ if str(_HERE) not in sys.path:
     sys.path.insert(0, str(_HERE))
 
 from paths import COOKIES, DUMPS, FROZEN, HERE, SCRATCH, VAULT
-from models import load_thread
+from models import PostData, ThreadData, load_thread
 
 try:
     from frozen import frozen_match, load_frozen_ids, require_writable
@@ -227,6 +230,132 @@ def _validate_capture_ids(path: Path, capture_ids: tuple[str, ...]) -> None:
         raise SystemExit(f"requested tips missing from capture: {', '.join(missing)}")
 
 
+def _select_branch_capture(
+    existing: ThreadData,
+    captured: ThreadData,
+    from_id: str,
+) -> tuple[tuple[PostData, ...], tuple[str, ...]]:
+    """Select one reply node's attachment path and visible subtree."""
+    captured_by = {post.post_id: post for post in captured.posts}
+    if from_id not in captured_by:
+        raise SystemExit(f"reply node {from_id} missing from capture")
+
+    children: dict[str, list[str]] = {
+        post.post_id: [] for post in captured.posts
+    }
+    for post in captured.posts:
+        if post.reply_to_id in captured_by:
+            children[post.reply_to_id].append(post.post_id)
+
+    subtree_ids: set[str] = set()
+    stack = [from_id]
+    while stack:
+        post_id = stack.pop()
+        if post_id in subtree_ids:
+            continue
+        subtree_ids.add(post_id)
+        stack.extend(reversed(children.get(post_id, [])))
+
+    leaves = tuple(
+        post.post_id
+        for post in captured.posts
+        if post.post_id in subtree_ids
+        and not any(
+            child_id in subtree_ids
+            for child_id in children.get(post.post_id, [])
+        )
+    )
+
+    existing_ids = {post.post_id for post in existing.posts}
+    attachment_ids: set[str] = set()
+    attachment_seen: set[str] = set()
+    current = from_id
+    while current not in existing_ids:
+        if current in attachment_seen:
+            raise SystemExit(
+                f"reply node {from_id} does not attach to existing thread"
+            )
+        attachment_seen.add(current)
+        post = captured_by.get(current)
+        if post is None:
+            raise SystemExit(
+                f"reply node {from_id} does not attach to existing thread"
+            )
+        attachment_ids.add(current)
+        if post.reply_to_id is None:
+            raise SystemExit(
+                f"reply node {from_id} does not attach to existing thread"
+            )
+        current = post.reply_to_id
+
+    selected_ids = subtree_ids | attachment_ids
+    selected = tuple(
+        post for post in captured.posts if post.post_id in selected_ids
+    )
+    return selected, leaves
+
+
+def _merge_branch_posts(
+    existing: ThreadData,
+    groups: list[tuple[PostData, ...]],
+) -> tuple[ThreadData, tuple[str, ...]]:
+    """Append previously absent selected posts without replacing old data."""
+    posts = list(existing.posts)
+    seen = {post.post_id for post in posts}
+    added: list[str] = []
+    for group in groups:
+        for post in group:
+            if post.post_id in seen:
+                continue
+            seen.add(post.post_id)
+            posts.append(post)
+            added.append(post.post_id)
+    return (
+        ThreadData(
+            root_post_id=existing.root_post_id,
+            posts=tuple(posts),
+            source_url=existing.source_url,
+        ),
+        tuple(added),
+    )
+
+
+def _copy_selected_media(
+    source: Path,
+    destination: Path,
+    post_ids: set[str],
+) -> tuple[Path, ...]:
+    """Copy media whose filename begins with a selected post id."""
+    source_media = source / "media"
+    destination_media = destination / "media"
+    if not source_media.is_dir() or not post_ids:
+        return ()
+    if source_media.resolve() == destination_media.resolve():
+        return ()
+    prefixes = tuple(f"{post_id}_" for post_id in sorted(post_ids))
+    destination_media.mkdir(parents=True, exist_ok=True)
+    copied: list[Path] = []
+    for path in sorted(source_media.iterdir()):
+        if not path.is_file() or not path.name.startswith(prefixes):
+            continue
+        output = destination_media / path.name
+        shutil.copy2(path, output)
+        copied.append(output)
+    return tuple(copied)
+
+
+def _write_thread(path: Path, thread: ThreadData) -> None:
+    """Atomically write one typed thread to scratch JSON."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    partial = path.with_suffix(path.suffix + ".part")
+    partial.write_text(
+        json.dumps(asdict(thread), indent=2, ensure_ascii=False) + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+    partial.replace(path)
+
+
 def cmd_refetch(args: argparse.Namespace) -> int:
     """Capture one spine tip plus explicit branch tips into one dump."""
     post_id = args.id
@@ -278,6 +407,57 @@ def cmd_refetch(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_add_branch(args: argparse.Namespace) -> int:
+    """Merge visible reply subtrees into an existing emitted thread."""
+    post_id = args.id
+    check_frozen(post_id)
+    assets, notes = locate(post_id)
+    if assets is None or notes is None:
+        raise SystemExit(f"no emitted thread for {post_id} — refresh first")
+    existing_path = assets / "thread_data.json"
+    if not existing_path.is_file():
+        raise SystemExit(f"missing {existing_path}")
+    existing = load_thread(existing_path)
+    if existing.root_post_id != post_id:
+        raise SystemExit(
+            f"{post_id} is not the stored spine tip {existing.root_post_id}"
+        )
+    if post_id not in {post.post_id for post in existing.posts}:
+        raise SystemExit(f"spine tip {post_id} not present in existing thread")
+
+    from_ids = tuple(dict.fromkeys(args.from_ids))
+    selected_groups: list[tuple[PostData, ...]] = []
+    capture_sources: list[tuple[Path, tuple[PostData, ...]]] = []
+    leaf_groups: list[tuple[str, tuple[str, ...]]] = []
+    for from_id in from_ids:
+        cmd_refetch(argparse.Namespace(id=from_id, branch=[]))
+        capture_dir = scratch_dir(from_id)
+        captured = load_thread(capture_dir / "thread_data.json")
+        selected, leaves = _select_branch_capture(existing, captured, from_id)
+        selected_groups.append(selected)
+        capture_sources.append((capture_dir, selected))
+        leaf_groups.append((from_id, leaves))
+
+    merged, added_ids = _merge_branch_posts(existing, selected_groups)
+    added_set = set(added_ids)
+    output_dir = scratch_dir(post_id)
+    for capture_dir, selected in capture_sources:
+        selected_added = {
+            post.post_id for post in selected if post.post_id in added_set
+        }
+        _copy_selected_media(capture_dir, output_dir, selected_added)
+    _write_thread(output_dir / "thread_data.json", merged)
+    cmd_emit(argparse.Namespace(id=post_id, tip=True, slug=None))
+
+    added_text = ",".join(added_ids) if added_ids else "-"
+    leaf_text = ";".join(
+        f"{from_id}:{','.join(leaves) if leaves else '-'}"
+        for from_id, leaves in leaf_groups
+    )
+    print(f"add-branch added={added_text} visible_leaves={leaf_text}")
+    return 0
+
+
 def cmd_emit(args: argparse.Namespace) -> int:
     """Run ``emit_archive.py`` for ``args.id`` and merge non-orig rows back into every per-author asset dir."""
     post_id = args.id
@@ -667,6 +847,20 @@ def build_parser() -> argparse.ArgumentParser:
     restore.add_argument("--id", required=True)
     restore.add_argument("--media-id", required=True)
 
+    add_branch = sub.add_parser(
+        "add-branch",
+        help="merge visible reply subtrees into an existing emitted thread",
+    )
+    add_branch.add_argument("--id", required=True)
+    add_branch.add_argument(
+        "--from",
+        dest="from_ids",
+        action="append",
+        required=True,
+        metavar="REPLY_ID",
+        help="capture this reply node's visible subtree; repeatable",
+    )
+
     return parser
 
 
@@ -674,6 +868,7 @@ _COMMANDS: dict[str, Callable[[argparse.Namespace], int]] = {
     "locate": cmd_locate,
     "graph": cmd_graph,
     "refetch": cmd_refetch,
+    "add-branch": cmd_add_branch,
     "emit": cmd_emit,
     "lift": cmd_lift,
     "ocr": cmd_ocr,

scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/root_tweet_detail.json

@@ -536,4 +536,4 @@
       ]
     }
   }
-}
+}