From ed86731701eb264bac5f97de190acafe77bd4cfc Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:48:07 -0400 Subject: [PATCH] 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 | 35 +++++++- scripts/twitter_threads/error_types.py | 25 +++++- scripts/twitter_threads/render_markdown.py | 44 +++++++++- tests/test_twitter_threads_pipeline.py | 99 ++++++++++++++++++++++ 4 files changed, 199 insertions(+), 4 deletions(-) create mode 100644 tests/test_twitter_threads_pipeline.py diff --git a/scripts/twitter_threads/download_media.py b/scripts/twitter_threads/download_media.py index 3a042951..0754e849 100644 --- a/scripts/twitter_threads/download_media.py +++ b/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()) diff --git a/scripts/twitter_threads/error_types.py b/scripts/twitter_threads/error_types.py index be7f2f40..e5a3bd6e 100644 --- a/scripts/twitter_threads/error_types.py +++ b/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", "")) diff --git a/scripts/twitter_threads/render_markdown.py b/scripts/twitter_threads/render_markdown.py index baf694a3..7b3bcf5b 100644 --- a/scripts/twitter_threads/render_markdown.py +++ b/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" @@ -68,4 +74,38 @@ def render_markdown(thread: ThreadData, media_names: dict[str, list[str]], outpu output.write_text(content, encoding="utf-8") except OSError as e: return Result.err(make_error("WriteError", "render_markdown", str(e))) - return Result.ok(output) \ No newline at end of file + 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()) \ No newline at end of file diff --git a/tests/test_twitter_threads_pipeline.py b/tests/test_twitter_threads_pipeline.py new file mode 100644 index 00000000..49f5ec2b --- /dev/null +++ b/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 = """ +
+ +
first
+ +
+
+ +
second
+
+""" + 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