more scripts

scripts/twitter/do_not_refetch.txt

@@ -0,0 +1,4 @@
+# Frozen: onatt0 participation. Do not refetch or overwrite unless the user says so.
+1917646466417381426
+1917656437473399108
+1951638140600381942

scripts/twitter/emit_archive.py

@@ -17,7 +17,14 @@ 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_ids
+from twitter.tree import (
+    by_id,
+    branch_roots,
+    children_map,
+    descendants,
+    spine_from_tip,
+    spine_ids,
+)
 
 _POST_ID_LINE = re.compile(r'^post_id:\s*"?([^"\s]+)"?\s*$')
 _WIKILINK_ITEM = re.compile(r"^- \[\[([^\]|]+)\]\]\s*$")
@@ -125,7 +132,7 @@ 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]) -> str:
+def render_gaps(thread: ThreadData, spine: list[str], *, input_kind: str = "root") -> str:
     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]
@@ -138,7 +145,7 @@ def render_gaps(thread: ThreadData, spine: list[str]) -> str:
 
     lines = [
         "# gaps",
-        "input: root",
+        f"input: {input_kind}",
         f"suggested_tip: {suggested}",
     ]
     if quote_ids:
@@ -337,6 +344,7 @@ def emit(
     *,
     force: bool = False,
     reuse_dir: Path | None = None,
+    tip: str | None = None,
 ) -> EmitResult:
     src = input_dir / "thread_data.json"
     if not src.is_file():
@@ -346,7 +354,12 @@ def emit(
     if not thread.posts:
         raise SystemExit(f"no posts in {src}")
 
-    spine = spine_ids(thread)
+    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)
@@ -426,7 +439,7 @@ def emit(
         newline="\n",
     )
     (asset_dir / "gaps.md").write_text(
-        render_gaps(thread, spine),
+        render_gaps(thread, spine, input_kind=input_kind),
         encoding="utf-8",
         newline="\n",
     )
@@ -541,6 +554,11 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         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)
 
 
@@ -558,7 +576,14 @@ def main(argv: list[str] | None = None) -> int:
             f"written={written} errors={errors}"
         )
         return 0 if errors == 0 else 1
-    emit(args.input, args.vault, args.slug, args.archived, force=args.force)
+    emit(
+        args.input,
+        args.vault,
+        args.slug,
+        args.archived,
+        force=args.force,
+        tip=args.tip,
+    )
     return 0
 
 

scripts/twitter/graph_dry_run.py

@@ -0,0 +1,110 @@
+"""Print a tip-choice graph for a thread_data.json. No vault writes."""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from twitter.models import load_thread
+from twitter.slug import slugify
+from twitter.tree import branch_roots, by_id, children_map, descendants, spine_ids
+
+
+def _one_line(text: str, n: int = 56) -> str:
+    s = " ".join((text or "").split())
+    return s if len(s) <= n else s[: n - 1] + "…"
+
+
+def _node(post) -> str:
+    return f"[I:{post.handle}:{post.post_id}]"
+
+
+def render_ssdl(thread) -> str:
+    ids = by_id(thread)
+    kids = children_map(thread)
+    spine = spine_ids(thread)
+    lines: list[str] = []
+    lines.append("# SSDL (adapted): [I:handle:id] instruction, -> spine, => sibling fan-out")
+    lines.append("# [B:…] branch root off the spine. [T] last spine post = default suggested_tip")
+    spine_parts = [_node(ids[i]) for i in spine]
+    lines.append("SPINE: " + " -> ".join(spine_parts) + " -> [T]")
+    for sid in spine:
+        off = [c for c in kids.get(sid, []) if c not in set(spine)]
+        if not off:
+            continue
+        bits = []
+        for cid in off:
+            p = ids[cid]
+            desc = descendants(cid, kids)
+            extra = f" +{len(desc)}" if desc else ""
+            bits.append(f"[B:{p.handle}:{p.post_id}{extra}]")
+        lines.append(f"  fanout {sid}: " + " => ".join(bits))
+    return "\n".join(lines)
+
+
+def render_ascii(thread) -> str:
+    ids = by_id(thread)
+    kids = children_map(thread)
+    spine = spine_ids(thread)
+    spine_set = set(spine)
+    roots = branch_roots(thread, spine)
+    lines: list[str] = []
+    lines.append("# ASCII tree  time  handle  id  slug  first-line")
+    lines.append(f"# dump_root={thread.root_post_id}  suggested_tip={spine[-1]}  posts={len(thread.posts)}")
+    lines.append(f"# source={thread.source_url}")
+
+    def walk(pid: str, prefix: str, is_last: bool, on_spine: bool) -> None:
+        p = ids[pid]
+        mark = "S" if pid in spine_set else ("B" if pid in roots else " ")
+        branch = "└─" if is_last else "├─"
+        slug = slugify(p.text)
+        lines.append(
+            f"{prefix}{branch}[{mark}] {p.timestamp}  @{p.handle}  {p.post_id}  {slug}"
+        )
+        lines.append(f"{prefix}{'    ' if is_last else '│   '}    {_one_line(p.text)}")
+        childs = kids.get(pid, [])
+        nxt = prefix + ("    " if is_last else "│   ")
+        for i, c in enumerate(childs):
+            walk(c, nxt, i == len(childs) - 1, c in spine_set)
+
+    # forest: posts with no in-dump parent first
+    tops = [p.post_id for p in thread.posts if not p.reply_to_id or p.reply_to_id not in ids]
+    if thread.root_post_id in ids and thread.root_post_id not in tops:
+        tops.insert(0, thread.root_post_id)
+    seen: set[str] = set()
+    ordered = []
+    for t in tops:
+        if t not in seen:
+            ordered.append(t)
+            seen.add(t)
+    for i, t in enumerate(ordered):
+        walk(t, "", i == len(ordered) - 1, t in spine_set)
+    lines.append("")
+    lines.append("# copy-paste ids")
+    lines.append(f"SPINE_IDS={' '.join(spine)}")
+    lines.append(f"SUGGESTED_TIP={spine[-1]}")
+    lines.append("BRANCH_ROOTS=" + " ".join(roots))
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Dry-run thread graph for tip choice.")
+    parser.add_argument("--input", type=Path, required=True, help="dump dir or thread_data.json")
+    args = parser.parse_args(argv)
+    path = args.input
+    if path.is_dir():
+        path = path / "thread_data.json"
+    thread = load_thread(path)
+    text = render_ssdl(thread) + "\n\n" + render_ascii(thread) + "\n"
+    out = args.input if args.input.is_dir() else args.input.parent
+    dest = Path(r"C:\projects\Threadwell-ai\scratch") / f"graph_{thread.root_post_id}.txt"
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    dest.write_text(text, encoding="utf-8")
+    sys.stdout.buffer.write((str(dest) + "\n").encode("ascii", "replace"))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/twitter/ingest_gallery.py

@@ -0,0 +1,129 @@
+"""Turn gallery-dl --dump-json into thread_data.json. No vault writes."""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from twitter.models import PostData, PostMetrics, ThreadData
+
+
+def _find_tweet_meta(entry: Any) -> dict[str, Any] | None:
+    if isinstance(entry, dict):
+        if "tweet_id" in entry or ("content" in entry and "author" in entry):
+            return entry
+        return None
+    if isinstance(entry, (list, tuple)):
+        for item in entry:
+            meta = _find_tweet_meta(item)
+            if meta is not None:
+                return meta
+    return None
+
+
+def _media_url_from_entry(entry: Any) -> str:
+    if isinstance(entry, (list, tuple)):
+        for item in entry:
+            if isinstance(item, str) and (
+                "pbs.twimg.com" in item or "video.twimg.com" in item
+            ):
+                return item
+    return ""
+
+
+def _post_from_meta(meta: dict[str, Any]) -> PostData:
+    author = meta.get("author") or {}
+    if isinstance(author, dict):
+        author_name = str(author.get("nick") or author.get("name") or "")
+        handle = str(author.get("name") or author.get("nick") or "")
+    else:
+        author_name = str(author)
+        handle = str(author)
+    view = meta.get("view_count")
+    reply = meta.get("reply_id")
+    quote = meta.get("quote_id")
+    return PostData(
+        post_id=str(meta.get("tweet_id") or meta.get("id") or ""),
+        author=author_name,
+        handle=handle,
+        text=str(meta.get("content") or ""),
+        timestamp=str(meta.get("date") or ""),
+        media_urls=(),
+        reply_to_id=str(reply) if reply else None,
+        quote_of_id=str(quote) if quote else None,
+        metrics=PostMetrics(
+            reply_count=int(meta.get("reply_count") or 0),
+            repost_count=int(meta.get("retweet_count") or 0),
+            like_count=int(meta.get("favorite_count") or 0),
+            view_count=int(view) if view is not None else None,
+        ),
+    )
+
+
+def posts_from_gallery(data: Any) -> list[PostData]:
+    by: dict[str, PostData] = {}
+    media: dict[str, list[str]] = {}
+    order: list[str] = []
+    entries = data if isinstance(data, list) else []
+    for entry in entries:
+        meta = _find_tweet_meta(entry)
+        if meta is None:
+            continue
+        tid = str(meta.get("tweet_id") or meta.get("id") or "")
+        if not tid:
+            continue
+        if tid not in by:
+            by[tid] = _post_from_meta(meta)
+            media[tid] = []
+            order.append(tid)
+        url = _media_url_from_entry(entry)
+        if url:
+            media[tid].append(url)
+    posts: list[PostData] = []
+    for tid in order:
+        base = by[tid]
+        posts.append(PostData(
+            post_id=base.post_id,
+            author=base.author,
+            handle=base.handle,
+            text=base.text,
+            timestamp=base.timestamp,
+            media_urls=tuple(media[tid]),
+            reply_to_id=base.reply_to_id,
+            quote_of_id=base.quote_of_id,
+            metrics=base.metrics,
+        ))
+    return posts
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--json", type=Path, required=True)
+    parser.add_argument("--out", type=Path, required=True)
+    parser.add_argument("--source-url", required=True)
+    parser.add_argument("--root", required=True)
+    args = parser.parse_args(argv)
+    data = json.loads(args.json.read_text(encoding="utf-8"))
+    posts = posts_from_gallery(data)
+    thread = ThreadData(
+        root_post_id=args.root,
+        posts=tuple(posts),
+        source_url=args.source_url,
+    )
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(
+        json.dumps(asdict(thread), indent=2, ensure_ascii=False) + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+    print(f"posts={len(posts)} -> {args.out}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/twitter/tree.py

@@ -18,6 +18,25 @@ def children_map(thread: ThreadData) -> dict[str, list[str]]:
     return kids
 
 
+def spine_from_tip(thread: ThreadData, tip_id: str) -> list[str]:
+    ids = by_id(thread)
+    if tip_id not in ids:
+        raise ValueError(f"tip {tip_id} not in dump")
+    handle = ids[tip_id].handle
+    chain = [tip_id]
+    cur = tip_id
+    while True:
+        parent = ids[cur].reply_to_id
+        if not parent or parent not in ids:
+            break
+        if ids[parent].handle != handle:
+            break
+        chain.append(parent)
+        cur = parent
+    chain.reverse()
+    return chain
+
+
 def spine_ids(thread: ThreadData) -> list[str]:
     ids = by_id(thread)
     kids = children_map(thread)