feat(twitter_threads): thread_from_dict + download_media/render_markdown CLIs (end-to-end -m pipeline)

thread_from_dict (JSON wire boundary) round-trips thread_data.json; download_media/render_markdown gain argparse main() + __main__ so the plan Task 5.3 pipeline runs via -m; dual-import added so all modules run standalone or via -m (G5). Pipeline integration test (html->names->render) + CLI tests cover the glue without network. TDD: red -> green (4 pipeline + 29 regression = 33 passed).

scripts/twitter_threads/download_media.py

@@ -1,10 +1,16 @@
 """Download media URLs from ThreadData via stdlib urllib into a local directory."""
 from __future__ import annotations
+import argparse
+import json
+import sys
 from pathlib import Path
 from urllib.request import urlopen
 from urllib.error import URLError
 from urllib.parse import urlparse, parse_qs
-from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, PostData, ThreadData
+try:
+ from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, PostData, ThreadData, thread_from_dict
+except ImportError:
+ from error_types import ErrorInfo, make_error, Result, PostData, ThreadData, thread_from_dict
 
 def _classify(url: str) -> tuple[str, str]:
  parsed = urlparse(url)
@@ -54,3 +60,30 @@ def download_media(thread: ThreadData, output_dir: Path) -> Result[list[Path]]:
     return Result.err(make_error("WriteError", "download_media", str(e)))
    paths.append(target)
  return Result.ok(paths)
+
+def main(argv: list[str] | None = None) -> int:
+ parser = argparse.ArgumentParser(description="Download media for a Twitter/X thread_data.json.")
+ parser.add_argument("--input", type=Path, required=True, help="thread_data.json path")
+ parser.add_argument("--output", type=Path, required=True, help="media output directory")
+ args = parser.parse_args(argv)
+ try:
+  raw = args.input.read_text(encoding="utf-8")
+ except OSError as e:
+  sys.stderr.write(f"error: ReadError: {e}\n")
+  return 1
+ try:
+  data = json.loads(raw)
+ except json.JSONDecodeError as e:
+  sys.stderr.write(f"error: JsonParseError: {e}\n")
+  return 1
+ thread = thread_from_dict(data)
+ result = download_media(thread, args.output)
+ if not result.is_ok:
+  sys.stderr.write(f"error: {result.error.kind}: {result.error.detail}\n")
+  return 1
+ print(f"downloaded {len(result.data)} file(s) to {args.output}")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

scripts/twitter_threads/error_types.py

@@ -1,7 +1,7 @@
 """Shared dataclasses for the scripts.twitter_threads namespace."""
 from __future__ import annotations
 from dataclasses import dataclass
-from typing import Generic, TypeVar
+from typing import Any, Generic, TypeVar
 
 @dataclass(frozen=True, slots=True)
 class ErrorInfo:
@@ -55,3 +55,26 @@ class Result(Generic[T]):
  @classmethod
  def err(cls, error: ErrorInfo) -> "Result[T]":
   return cls(data=None, error=error)
+
+def thread_from_dict(d: dict[str, Any]) -> ThreadData:
+ posts: list[PostData] = []
+ for pd in d.get("posts", []):
+  m = pd.get("metrics", {})
+  metrics = PostMetrics(
+   reply_count=int(m.get("reply_count", 0)),
+   repost_count=int(m.get("repost_count", 0)),
+   like_count=int(m.get("like_count", 0)),
+   view_count=m.get("view_count"),
+  )
+  posts.append(PostData(
+   post_id=pd.get("post_id", ""),
+   author=pd.get("author", ""),
+   handle=pd.get("handle", ""),
+   text=pd.get("text", ""),
+   timestamp=pd.get("timestamp", ""),
+   media_urls=tuple(pd.get("media_urls", [])),
+   reply_to_id=pd.get("reply_to_id"),
+   quote_of_id=pd.get("quote_of_id"),
+   metrics=metrics,
+  ))
+ return ThreadData(root_post_id=d.get("root_post_id", ""), posts=tuple(posts), source_url=d.get("source_url", ""))

scripts/twitter_threads/render_markdown.py

@@ -1,7 +1,13 @@
 """Render ThreadData into a Markdown file with YAML front-matter."""
 from __future__ import annotations
+import argparse
+import json
+import sys
 from pathlib import Path
-from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, ThreadData, PostData
+try:
+ from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, ThreadData, PostData, thread_from_dict
+except ImportError:
+ from error_types import ErrorInfo, make_error, Result, ThreadData, PostData, thread_from_dict
 
 _EM_DASH = "\u2014"
 
@@ -69,3 +75,37 @@ def render_markdown(thread: ThreadData, media_names: dict[str, list[str]], outpu
  except OSError as e:
   return Result.err(make_error("WriteError", "render_markdown", str(e)))
  return Result.ok(output)
+
+def main(argv: list[str] | None = None) -> int:
+ parser = argparse.ArgumentParser(description="Render a Twitter/X thread_data.json to Markdown.")
+ parser.add_argument("--input", type=Path, required=True, help="thread_data.json path")
+ parser.add_argument("--media-dir", type=Path, required=True, help="media directory to link")
+ parser.add_argument("--output", type=Path, required=True, help="output .md path")
+ args = parser.parse_args(argv)
+ try:
+  raw = args.input.read_text(encoding="utf-8")
+ except OSError as e:
+  sys.stderr.write(f"error: ReadError: {e}\n")
+  return 1
+ try:
+  data = json.loads(raw)
+ except json.JSONDecodeError as e:
+  sys.stderr.write(f"error: JsonParseError: {e}\n")
+  return 1
+ thread = thread_from_dict(data)
+ media_names: dict[str, list[str]] = {}
+ for post in thread.posts:
+  if args.media_dir.exists():
+   files = sorted(f.name for f in args.media_dir.glob(f"{post.post_id}_*"))
+   if files:
+    media_names[post.post_id] = files
+ result = render_markdown(thread, media_names, args.output)
+ if not result.is_ok:
+  sys.stderr.write(f"error: {result.error.kind}: {result.error.detail}\n")
+  return 1
+ print(str(args.output))
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

tests/test_twitter_threads_pipeline.py

@@ -0,0 +1,99 @@
+"""RED tests for the twitter_threads pipeline: thread_from_dict, download_main, render_main."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from scripts.twitter_threads.error_types import PostMetrics, PostData, ThreadData, thread_from_dict
+from scripts.twitter_threads.fetch_thread import fetch_thread_from_html, thread_to_dict
+from scripts.twitter_threads.download_media import media_names_for_post, download_media, main as download_main
+from scripts.twitter_threads.render_markdown import render_markdown, main as render_main
+
+
+class _FakeResp:
+ def __init__(self, data: bytes) -> None:
+  self._data = data
+
+ def read(self) -> bytes:
+  return self._data
+
+ def __enter__(self) -> "_FakeResp":
+  return self
+
+ def __exit__(self, *a: object) -> bool:
+  return False
+
+
+def _post(post_id: str, text: str = "hi", media_urls: tuple[str, ...] = (), reply_to_id: str | None = None) -> PostData:
+ metrics = PostMetrics(reply_count=1, repost_count=2, like_count=3, view_count=None)
+ return PostData(post_id=post_id, author="Tim", handle="NOTimothyLottes", text=text, timestamp="2024-02-01T00:00:00Z", media_urls=media_urls, reply_to_id=reply_to_id, quote_of_id=None, metrics=metrics)
+
+
+def test_thread_json_roundtrip() -> None:
+ p1 = _post("1")
+ p2 = _post("2", media_urls=("https://pbs.twimg.com/media/AAA.jpg",), reply_to_id="1")
+ t = ThreadData(root_post_id="1", posts=(p1, p2), source_url="https://x.com/x/x/1")
+ d = thread_to_dict(t)
+ t2 = thread_from_dict(json.loads(json.dumps(d)))
+ assert t2 == t
+
+
+def test_pipeline_html_to_markdown(tmp_path: Path) -> None:
+ html = """<html><body>
+<article data-post-id="10" data-author="Tim" data-handle="NOTimothyLottes">
+ <time datetime="2024-02-01T00:00:00Z">d1</time>
+ <div data-testid="tweetText">first</div>
+ <img src="https://pbs.twimg.com/media/ZZZ.jpg">
+</article>
+<article data-post-id="11" data-author="Tim" data-handle="NOTimothyLottes" data-reply-to="10">
+ <time datetime="2024-02-01T00:00:00Z">d2</time>
+ <div data-testid="tweetText">second</div>
+</article>
+</body></html>"""
+ path = tmp_path / "thread.html"
+ path.write_text(html, encoding="utf-8")
+ res = fetch_thread_from_html(path)
+ assert res.is_ok
+ td = res.data
+ assert td is not None
+ media_names = {p.post_id: media_names_for_post(p) for p in td.posts if p.media_urls}
+ out = tmp_path / "thread.md"
+ rr = render_markdown(td, media_names, out)
+ assert rr.is_ok
+ text = out.read_text(encoding="utf-8")
+ assert "## Post 1" in text
+ assert "## Post 2" in text
+ assert "reply to Post 1" in text
+ assert "[Media 1](./media/10_img1.jpg)" in text
+
+
+def test_download_cli(tmp_path: Path) -> None:
+ p1 = _post("5", media_urls=("https://pbs.twimg.com/media/BBB.jpg",))
+ t = ThreadData(root_post_id="5", posts=(p1,), source_url="https://x.com/x/x/5")
+ thread_data_json = tmp_path / "thread_data.json"
+ thread_data_json.write_text(json.dumps(thread_to_dict(t)), encoding="utf-8")
+ mediadir = tmp_path / "media"
+ with patch("scripts.twitter_threads.download_media.urlopen", return_value=_FakeResp(b"x")):
+  code = download_main(["--input", str(thread_data_json), "--output", str(mediadir)])
+ assert code == 0
+ assert (mediadir / "5_img1.jpg").exists()
+
+
+def test_render_cli(tmp_path: Path) -> None:
+ p1 = _post("7", text="body", media_urls=("https://pbs.twimg.com/media/CCC.jpg",))
+ t = ThreadData(root_post_id="7", posts=(p1,), source_url="https://x.com/x/x/7")
+ thread_data_json = tmp_path / "thread_data.json"
+ thread_data_json.write_text(json.dumps(thread_to_dict(t)), encoding="utf-8")
+ mediadir = tmp_path / "media"
+ mediadir.mkdir()
+ (mediadir / "7_img1.jpg").write_bytes(b"x")
+ out = tmp_path / "thread.md"
+ code = render_main(["--input", str(thread_data_json), "--media-dir", str(mediadir), "--output", str(out)])
+ assert code == 0
+ assert out.exists()
+ text = out.read_text(encoding="utf-8")
+ assert "[Media 1](./media/7_img1.jpg)" in text
+ assert "body" in text