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/<name> links; media_names mapping keeps render decoupled from download_media. TDD: red (no Result) -> green (7 render + 9 types passed).
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user