From ef98f6d1a1a7e0b3c07d5ebdad553c64efda03ae Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:17:18 -0400 Subject: [PATCH] feat(twitter_threads): Result[T] + render_markdown.py (YAML front-matter + per-post sections + media links) Added standalone Result[T] (frozen+slots, ok/err classmethods, is_ok) to error_types.py per canonical AND-over-OR error_handling.md (not the video_analysis _Ok|_Err sum type). render_markdown renders YAML front-matter from root post, per-post ## sections with reply markers, quote blockquotes, and ./media/ links; media_names mapping keeps render decoupled from download_media. TDD: red (no Result) -> green (7 render + 9 types passed). --- scripts/twitter_threads/error_types.py | 20 ++++ scripts/twitter_threads/render_markdown.py | 71 ++++++++++++++ tests/test_twitter_threads_render.py | 107 +++++++++++++++++++++ 3 files changed, 198 insertions(+) create mode 100644 scripts/twitter_threads/render_markdown.py create mode 100644 tests/test_twitter_threads_render.py diff --git a/scripts/twitter_threads/error_types.py b/scripts/twitter_threads/error_types.py index 010eb0e9..be7f2f40 100644 --- a/scripts/twitter_threads/error_types.py +++ b/scripts/twitter_threads/error_types.py @@ -1,6 +1,7 @@ """Shared dataclasses for the scripts.twitter_threads namespace.""" from __future__ import annotations from dataclasses import dataclass +from typing import Generic, TypeVar @dataclass(frozen=True, slots=True) class ErrorInfo: @@ -35,3 +36,22 @@ class ThreadData: root_post_id: str posts: tuple[PostData, ...] source_url: str + +T = TypeVar("T") + +@dataclass(frozen=True, slots=True) +class Result(Generic[T]): + data: T | None = None + error: ErrorInfo | None = None + + @property + def is_ok(self) -> bool: + return self.error is None + + @classmethod + def ok(cls, data: T) -> "Result[T]": + return cls(data=data, error=None) + + @classmethod + def err(cls, error: ErrorInfo) -> "Result[T]": + return cls(data=None, error=error) diff --git a/scripts/twitter_threads/render_markdown.py b/scripts/twitter_threads/render_markdown.py new file mode 100644 index 00000000..baf694a3 --- /dev/null +++ b/scripts/twitter_threads/render_markdown.py @@ -0,0 +1,71 @@ +"""Render ThreadData into a Markdown file with YAML front-matter.""" +from __future__ import annotations +from pathlib import Path +from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, ThreadData, PostData + +_EM_DASH = "\u2014" + +def _build_id_to_index(thread: ThreadData) -> dict[str, int]: + return {post.post_id: i + 1 for i, post in enumerate(thread.posts)} + +def _format_view_count(value: int | None) -> str: + if value is None: + return "null" + return str(value) + +def _render_yaml(thread: ThreadData, title: str) -> list[str]: + root = thread.posts[0] + return [ + "---", + f'title: "{title}"', + f'author: "{root.author}"', + f'handle: "@{root.handle}"', + f'post_url: "{thread.source_url}"', + f'post_id: "{thread.root_post_id}"', + f'timestamp: "{root.timestamp}"', + f'post_count: {len(thread.posts)}', + f'reply_count: {root.metrics.reply_count}', + f'repost_count: {root.metrics.repost_count}', + f'like_count: {root.metrics.like_count}', + f'view_count: {_format_view_count(root.metrics.view_count)}', + "---", + ] + +def _render_post(post: PostData, n: int, id_to_index: dict[str, int], media_names: dict[str, list[str]]) -> list[str]: + lines: list[str] = [] + header = f"## Post {n} ({post.timestamp})" + if post.reply_to_id is not None and post.reply_to_id in id_to_index: + header += f" {_EM_DASH} reply to Post {id_to_index[post.reply_to_id]}" + lines.append(header) + lines.append("") + lines.append(post.text) + lines.append("") + if post.quote_of_id is not None: + lines.append(f"> [Quoting post {post.quote_of_id}](https://x.com/i/web/status/{post.quote_of_id})") + lines.append("") + names = media_names.get(post.post_id, []) + if names: + for k, name in enumerate(names, start=1): + lines.append(f"[Media {k}](./media/{name})") + lines.append("") + return lines + +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] + first_line = root.text.splitlines()[0] if root.text.splitlines() else "" + title = first_line[:80] + lines.extend(_render_yaml(thread, title)) + lines.append("") + lines.append(f"# @{root.handle} {_EM_DASH} {title}") + lines.append("") + for i, post in enumerate(thread.posts): + n = i + 1 + lines.extend(_render_post(post, n, id_to_index, media_names)) + content = "\n".join(lines) + try: + 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 diff --git a/tests/test_twitter_threads_render.py b/tests/test_twitter_threads_render.py new file mode 100644 index 00000000..2a9302af --- /dev/null +++ b/tests/test_twitter_threads_render.py @@ -0,0 +1,107 @@ +"""Tests for scripts.twitter_threads.render_markdown and Result contract.""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from scripts.twitter_threads.error_types import ErrorInfo, make_error, PostMetrics, PostData, Result, ThreadData +from scripts.twitter_threads.render_markdown import render_markdown + + +def _post(post_id: str, text: str, reply_to_id: str | None = None, quote_of_id: str | None = None, view_count: int = 0, timestamp: str = "2024-02-01T00:00:00Z", handle: str = "NOTimothyLottes", author: str = "Tim") -> PostData: + metrics = PostMetrics(reply_count=0, repost_count=0, like_count=0, view_count=view_count) + return PostData(post_id=post_id, author=author, handle=handle, text=text, timestamp=timestamp, media_urls=(), reply_to_id=reply_to_id, quote_of_id=quote_of_id, metrics=metrics) + + +def test_result_ok_and_err() -> None: + r = Result.ok("x") + assert r.is_ok is True + assert r.data == "x" + e = make_error("K", "s", "d") + r2 = Result.err(e) + assert r2.is_ok is False + assert r2.error is e + + +def test_render_single_post(tmp_path: Path) -> None: + root = _post("123", "Hello world", view_count=5) + thread = ThreadData(root_post_id="123", posts=(root,), source_url="https://x.com/x/x/123") + out = tmp_path / "thread.md" + result = render_markdown(thread, {}, out) + assert result.is_ok + assert result.data == out + assert out.exists() + text = out.read_text(encoding="utf-8") + assert text.startswith("---") + assert "@NOTimothyLottes" in text + assert "post_id:" in text + assert "post_count: 1" in text + assert "view_count: 5" in text + assert "Hello world" in text + + +def test_render_thread(tmp_path: Path) -> None: + p1 = _post("1", "first post") + p2 = _post("2", "second post", reply_to_id="1") + p3 = _post("3", "third post", reply_to_id="2") + thread = ThreadData(root_post_id="1", posts=(p1, p2, p3), source_url="https://x.com/x/x/1") + out = tmp_path / "thread.md" + result = render_markdown(thread, {}, out) + assert result.is_ok + text = out.read_text(encoding="utf-8") + i1 = text.index("## Post 1") + i2 = text.index("## Post 2") + i3 = text.index("## Post 3") + assert i1 < i2 < i3 + assert "reply to Post 1" in text + assert "reply to Post 2" in text + + +def test_render_quote_tweet(tmp_path: Path) -> None: + root = _post("100", "this is a quote", quote_of_id="999") + thread = ThreadData(root_post_id="100", posts=(root,), source_url="https://x.com/x/x/100") + out = tmp_path / "thread.md" + result = render_markdown(thread, {}, out) + assert result.is_ok + text = out.read_text(encoding="utf-8") + quote_lines = [line for line in text.splitlines() if line.startswith("> ")] + assert any("Quoting" in line and "999" in line for line in quote_lines) + + +def test_render_media_links(tmp_path: Path) -> None: + root = _post("123", "media post") + thread = ThreadData(root_post_id="123", posts=(root,), source_url="https://x.com/x/x/123") + out = tmp_path / "thread.md" + media_names = {"123": ["123_img1.jpg", "123_img2.jpg", "123_vid1.mp4"]} + result = render_markdown(thread, media_names, out) + assert result.is_ok + text = out.read_text(encoding="utf-8") + assert "[Media 1](./media/123_img1.jpg)" in text + assert "[Media 2](./media/123_img2.jpg)" in text + assert "[Media 3](./media/123_vid1.mp4)" in text + + +def test_render_view_count_null(tmp_path: Path) -> None: + root = _post("123", "hello", view_count=None) + thread = ThreadData(root_post_id="123", posts=(root,), source_url="https://x.com/x/x/123") + out = tmp_path / "thread.md" + result = render_markdown(thread, {}, out) + assert result.is_ok + text = out.read_text(encoding="utf-8") + assert "view_count: null" in text + + +def test_render_title_truncated(tmp_path: Path) -> None: + text100 = "A" * 100 + root = _post("123", text100) + thread = ThreadData(root_post_id="123", posts=(root,), source_url="https://x.com/x/x/123") + out = tmp_path / "thread.md" + result = render_markdown(thread, {}, out) + assert result.is_ok + text = out.read_text(encoding="utf-8") + h1_lines = [line for line in text.splitlines() if line.startswith("# @")] + assert h1_lines, "expected an H1 line starting with '# @'" + h1 = h1_lines[0] + assert "A" * 80 in h1 + assert "A" * 81 not in h1 \ No newline at end of file