feat(twitter_threads): full-thread extraction (conversations) + front-matter from target + corpus dedupe

fetch_thread_from_url now passes -o conversations=true (full threads, not single tweets), sorts posts chronologically, and sets root_post_id to the URL's target tweet. render_markdown builds front-matter from the target (root_post_id) post, not the earliest conversation tweet (fixes wrong @handle). Added dedupe_corpus.py: keeps deepest thread per conversation, deletes duplicates, keeps+marks unique branches (branch_of) + writes threads_index.json. 33 tests pass.

conductor/tracks/twitter_threads_extraction_20260705/dedupe_corpus.py

@@ -0,0 +1,100 @@
+"""Deduplicate the docs/twitter corpus: keep the deepest thread per conversation, delete
+shallower duplicates, and KEEP + MARK unique branches. Part of the twitter_threads_extraction_20260705 track.
+
+A "conversation group" = extracted threads whose post-id sets overlap. Within a group the
+PRIMARY is the extraction with the most posts (deepest). For every other member:
+  - if its target tweet is contained in the PRIMARY's post set  -> DUPLICATE (delete its dir)
+  - else                                                        -> UNIQUE BRANCH (keep; mark branch_of)
+
+Branch tracking: kept branches get `branch_of` written into their thread_data.json, and every
+kept thread is recorded in docs/twitter/threads_index.json.
+
+Usage:
+  uv run python conductor/tracks/twitter_threads_extraction_20260705/dedupe_corpus.py          # dry run
+  uv run python conductor/tracks/twitter_threads_extraction_20260705/dedupe_corpus.py --apply  # execute
+"""
+from __future__ import annotations
+
+import json
+import shutil
+import sys
+from pathlib import Path
+
+CORPUS = Path("docs/twitter")
+
+
+def _load() -> dict[str, dict]:
+ info: dict[str, dict] = {}
+ for d in sorted(CORPUS.glob("*")):
+  j = d / "thread_data.json"
+  if not j.exists():
+   continue
+  t = json.loads(j.read_text(encoding="utf-8"))
+  tid = t["root_post_id"]
+  posts = {p["post_id"] for p in t["posts"]}
+  handle = next((p["handle"] for p in t["posts"] if p["post_id"] == tid), "?")
+  info[tid] = {"dir": d, "posts": posts, "n": len(posts), "handle": handle, "url": t.get("source_url", ""), "json": j}
+ return info
+
+
+def _groups(info: dict[str, dict]) -> list[list[str]]:
+ ids = list(info)
+ parent = {i: i for i in ids}
+ def find(x: str) -> str:
+  while parent[x] != x:
+   parent[x] = parent[parent[x]]
+   x = parent[x]
+  return x
+ for i in range(len(ids)):
+  for k in range(i + 1, len(ids)):
+   a, b = ids[i], ids[k]
+   if info[a]["posts"] & info[b]["posts"]:
+    parent[find(a)] = find(b)
+ out: dict[str, list[str]] = {}
+ for i in ids:
+  out.setdefault(find(i), []).append(i)
+ return list(out.values())
+
+
+def main() -> int:
+ apply = "--apply" in sys.argv
+ info = _load()
+ if not info:
+  print("no corpus found under docs/twitter/")
+  return 1
+ index: list[dict] = []
+ to_delete: list[str] = []
+ branches: list[tuple[str, str]] = []
+ for members in _groups(info):
+  members.sort(key=lambda m: (info[m]["n"], int(m)), reverse=True)
+  primary = members[0]
+  print(f"\nGROUP @{info[primary]['handle']} size={len(members)} primary={primary} (n={info[primary]['n']})")
+  for m in members:
+   if m == primary:
+    role = "PRIMARY(deepest)"
+    index.append({"target": m, "handle": info[m]["handle"], "posts": info[m]["n"], "role": "primary", "branch_of": None, "url": info[m]["url"]})
+   elif m in info[primary]["posts"]:
+    role = "DUPLICATE -> delete"
+    to_delete.append(m)
+   else:
+    role = f"UNIQUE BRANCH of {primary} -> keep + mark"
+    branches.append((m, primary))
+    index.append({"target": m, "handle": info[m]["handle"], "posts": info[m]["n"], "role": "branch", "branch_of": primary, "url": info[m]["url"]})
+   print(f"   {m} n={info[m]['n']} @{info[m]['handle']} -> {role}")
+ print(f"\nPLAN: keep {len(index)} threads, delete {len(to_delete)} duplicates, mark {len(branches)} branches. apply={apply}")
+ if apply:
+  for m, primary in branches:
+   j = info[m]["json"]
+   t = json.loads(j.read_text(encoding="utf-8"))
+   t["branch_of"] = primary
+   j.write_text(json.dumps(t, indent=2, ensure_ascii=False), encoding="utf-8")
+  for m in to_delete:
+   shutil.rmtree(info[m]["dir"], ignore_errors=True)
+   print(f"deleted {info[m]['dir']}")
+  (CORPUS / "threads_index.json").write_text(json.dumps(sorted(index, key=lambda e: e["target"]), indent=2, ensure_ascii=False), encoding="utf-8")
+  print(f"wrote {CORPUS / 'threads_index.json'}")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())

conductor/tracks/twitter_threads_extraction_20260705/run_corpus.py

@@ -53,7 +53,7 @@ def main() -> int:
   (base / "thread_data.json").write_text(json.dumps(thread_to_dict(td), indent=2, ensure_ascii=False), encoding="utf-8")
   media_names = {p.post_id: media_names_for_post(p) for p in td.posts if p.media_urls}
   rr = render_markdown(td, media_names, base / "thread.md")
-  handle = td.posts[0].handle if td.posts else "?"
+  handle = next((p.handle for p in td.posts if p.post_id == td.root_post_id), td.posts[0].handle if td.posts else "?")
   status = "OK" if rr.is_ok else f"RENDER_FAIL {rr.error.kind}"
   print(f"{status} {td.root_post_id}: posts={len(td.posts)} media={n_media} @{handle} -> {base / 'thread.md'}")
   if rr.is_ok:

scripts/twitter_threads/fetch_thread.py

@@ -207,7 +207,7 @@ def _posts_from_gallery_dl(data: Any) -> list[PostData]:
 
 def fetch_thread_from_url(url: str, cookies_path: Path | None = None) -> Result[ThreadData]:
  clean = _normalize_url(url)
- args = ["gallery-dl", "--dump-json", "-o", "text-tweets=true"]
+ args = ["gallery-dl", "--dump-json", "-o", "text-tweets=true", "-o", "conversations=true"]
  if cookies_path is not None:
   args += ["--cookies", str(cookies_path)]
  args.append(clean)
@@ -224,7 +224,10 @@ def fetch_thread_from_url(url: str, cookies_path: Path | None = None) -> Result[
  posts = _posts_from_gallery_dl(data)
  if not posts:
   return Result.err(make_error("ParseError", "fetch_thread_from_url", "no tweet metadata in gallery-dl output"))
- return Result.ok(ThreadData(root_post_id=posts[0].post_id, posts=tuple(posts), source_url=clean))
+ posts.sort(key=lambda p: int(p.post_id) if p.post_id.isdigit() else 0)
+ target_id = clean.rstrip("/").split("/")[-1]
+ root_id = target_id if target_id.isdigit() else posts[0].post_id
+ return Result.ok(ThreadData(root_post_id=root_id, posts=tuple(posts), source_url=clean))
 
 
 def thread_to_dict(thread: ThreadData) -> dict[str, Any]:

scripts/twitter_threads/render_markdown.py

@@ -19,8 +19,14 @@ def _format_view_count(value: int | None) -> str:
   return "null"
  return str(value)
 
+def _root_post(thread: ThreadData) -> PostData:
+ for p in thread.posts:
+  if p.post_id == thread.root_post_id:
+   return p
+ return thread.posts[0]
+
 def _render_yaml(thread: ThreadData, title: str) -> list[str]:
- root = thread.posts[0]
+ root = _root_post(thread)
  return [
   "---",
   f'title: "{title}"',
@@ -59,7 +65,7 @@ def _render_post(post: PostData, n: int, id_to_index: dict[str, int], media_name
 def render_markdown(thread: ThreadData, media_names: dict[str, list[str]], output: Path) -> Result[Path]:
  lines: list[str] = []
  id_to_index = _build_id_to_index(thread)
- root = thread.posts[0]
+ root = _root_post(thread)
  first_line = root.text.splitlines()[0] if root.text.splitlines() else ""
  title = first_line[:80]
  lines.extend(_render_yaml(thread, title))