Merge remote-tracking branch 'tier2-clone/tier2/twitter_threads_extraction_20260705'

This commit is contained in:
ed
2026-07-05 19:34:31 -04:00
58 changed files with 2789 additions and 70 deletions
+133
View File
@@ -0,0 +1,133 @@
"""RED tests for scripts.twitter_threads.fetch_thread (normalize URL + HTML to ThreadData)."""
from __future__ import annotations
from pathlib import Path
import pytest
from scripts.twitter_threads.error_types import PostData, ThreadData
from scripts.twitter_threads.fetch_thread import fetch_thread_from_html, _normalize_url
def test_normalize_url_strips_query() -> None:
assert _normalize_url("https://x.com/VPCOMPRESSB/status/1991383117571957052?s=20") == "https://x.com/VPCOMPRESSB/status/1991383117571957052"
assert _normalize_url("https://x.com/NOTimothyLottes/status/123") == "https://x.com/NOTimothyLottes/status/123"
def test_parse_html_single_post(tmp_path: Path) -> None:
html = """<html><body>
<link rel="canonical" href="https://x.com/NOTimothyLottes/status/1757198624818168210">
<article data-post-id="1757198624818168210" data-author="Timothy Lottes" data-handle="NOTimothyLottes" data-reply-count="5" data-repost-count="10" data-like-count="100" data-view-count="5000">
<time datetime="2024-02-13T12:00:00Z">Feb 13</time>
<div data-testid="tweetText">Hello thread</div>
<img src="https://pbs.twimg.com/media/AAA.jpg">
</article>
</body></html>"""
p = tmp_path / "x.html"
p.write_text(html, encoding="utf-8")
result = fetch_thread_from_html(p)
assert result.is_ok
td = result.data
assert td is not None
assert td.root_post_id == "1757198624818168210"
assert len(td.posts) == 1
post = td.posts[0]
assert post.post_id == "1757198624818168210"
assert post.author == "Timothy Lottes"
assert post.handle == "NOTimothyLottes"
assert post.timestamp == "2024-02-13T12:00:00Z"
assert "Hello thread" in post.text
assert post.media_urls == ("https://pbs.twimg.com/media/AAA.jpg",)
assert post.metrics.reply_count == 5
assert post.metrics.repost_count == 10
assert post.metrics.like_count == 100
assert post.metrics.view_count == 5000
assert td.source_url == "https://x.com/NOTimothyLottes/status/1757198624818168210"
def test_parse_html_thread(tmp_path: Path) -> None:
html = """<html><body>
<article data-post-id="1" data-author="A" data-handle="a">
<time datetime="2024-01-01T00:00:00Z">d1</time>
<div data-testid="tweetText">p1</div>
</article>
<article data-post-id="2" data-author="A" data-handle="a" data-reply-to="1">
<time datetime="2024-01-02T00:00:00Z">d2</time>
<div data-testid="tweetText">p2</div>
</article>
<article data-post-id="3" data-author="A" data-handle="a" data-reply-to="2">
<time datetime="2024-01-03T00:00:00Z">d3</time>
<div data-testid="tweetText">p3</div>
</article>
</body></html>"""
p = tmp_path / "x.html"
p.write_text(html, encoding="utf-8")
result = fetch_thread_from_html(p)
assert result.is_ok
td = result.data
assert td is not None
assert len(td.posts) == 3
ids = [pt.post_id for pt in td.posts]
assert ids == ["1", "2", "3"]
assert td.posts[1].reply_to_id == "1"
assert td.posts[2].reply_to_id == "2"
assert td.posts[0].reply_to_id is None
def test_parse_html_no_media(tmp_path: Path) -> None:
html = """<html><body>
<article data-post-id="1" data-author="A" data-handle="a">
<time datetime="2024-01-01T00:00:00Z">d</time>
<div data-testid="tweetText">no media here</div>
</article>
</body></html>"""
p = tmp_path / "x.html"
p.write_text(html, encoding="utf-8")
result = fetch_thread_from_html(p)
assert result.is_ok
td = result.data
assert td is not None
assert td.posts[0].media_urls == ()
def test_parse_html_quote(tmp_path: Path) -> None:
html = """<html><body>
<article data-post-id="1" data-author="A" data-handle="a" data-quote-of="999">
<time datetime="2024-01-01T00:00:00Z">d</time>
<div data-testid="tweetText">quoting</div>
</article>
</body></html>"""
p = tmp_path / "x.html"
p.write_text(html, encoding="utf-8")
result = fetch_thread_from_html(p)
assert result.is_ok
td = result.data
assert td is not None
assert td.posts[0].quote_of_id == "999"
def test_parse_html_malformed(tmp_path: Path) -> None:
html = "<html><body><p>no articles here</p></body></html>"
p = tmp_path / "x.html"
p.write_text(html, encoding="utf-8")
result = fetch_thread_from_html(p)
assert result.is_ok is False
assert result.error is not None
assert result.error.kind == "ParseError"
def test_parse_html_view_count_absent(tmp_path: Path) -> None:
html = """<html><body>
<article data-post-id="1" data-author="A" data-handle="a">
<time datetime="2024-01-01T00:00:00Z">d</time>
<div data-testid="tweetText">x</div>
</article>
</body></html>"""
p = tmp_path / "x.html"
p.write_text(html, encoding="utf-8")
result = fetch_thread_from_html(p)
assert result.is_ok
td = result.data
assert td is not None
assert td.posts[0].metrics.view_count is None
assert td.posts[0].metrics.reply_count == 0
+62
View File
@@ -0,0 +1,62 @@
"""Tests for scripts.twitter_threads.download_media gallery-dl subprocess contract."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
from scripts.twitter_threads.error_types import PostMetrics, PostData, ThreadData
from scripts.twitter_threads.download_media import download_media, media_files_for_post
def _post(post_id: str) -> PostData:
metrics = PostMetrics(reply_count=0, repost_count=0, like_count=0, view_count=None)
return PostData(post_id=post_id, author="Tim", handle="NOTimothyLottes", text="t", timestamp="2024-02-01T00:00:00Z", media_urls=(), reply_to_id=None, quote_of_id=None, metrics=metrics)
def _thread(post_id: str) -> ThreadData:
return ThreadData(root_post_id=post_id, posts=(_post(post_id),), source_url=f"https://x.com/x/status/{post_id}")
def test_download_collects_matching_files(tmp_path: Path) -> None:
mediadir = tmp_path / "m"
mediadir.mkdir()
(mediadir / "123_1.jpg").write_bytes(b"x")
(mediadir / "123_2.png").write_bytes(b"x")
(mediadir / "999_1.jpg").write_bytes(b"x")
with patch("scripts.twitter_threads.download_media.subprocess.run", MagicMock(return_value=MagicMock(returncode=0))):
r = download_media(_thread("123"), mediadir)
assert r.is_ok
names = [p.name for p in r.data]
assert "123_1.jpg" in names
assert "123_2.png" in names
assert "999_1.jpg" not in names
def test_download_invokes_gallery_dl_with_url_and_cookies(tmp_path: Path) -> None:
mediadir = tmp_path / "m"
ck = tmp_path / "cookies.txt"
ck.write_text("x", encoding="utf-8")
with patch("scripts.twitter_threads.download_media.subprocess.run", MagicMock(return_value=MagicMock(returncode=0))) as mock:
download_media(_thread("55"), mediadir, ck)
assert mock.called
args = mock.call_args.args[0]
assert "gallery-dl" in args[0] or args[0] == "gallery-dl"
assert "https://x.com/x/status/55" in args
assert "--cookies" in args
assert str(ck) in args
def test_download_gallery_dl_error(tmp_path: Path) -> None:
with patch("scripts.twitter_threads.download_media.subprocess.run", MagicMock(return_value=MagicMock(returncode=1, stderr="boom"))):
r = download_media(_thread("123"), tmp_path / "m")
assert r.is_ok is False
assert r.error.kind == "GalleryDlError"
def test_media_files_for_post(tmp_path: Path) -> None:
d = tmp_path / "m"
d.mkdir()
(d / "123_1.jpg").write_bytes(b"x")
(d / "123_2.png").write_bytes(b"x")
(d / "999_1.jpg").write_bytes(b"x")
assert media_files_for_post(d, "123") == ["123_1.jpg", "123_2.png"]
+101
View File
@@ -0,0 +1,101 @@
"""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, MagicMock
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 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 = {"10": ["10_img1.jpg"]}
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"
mediadir.mkdir()
(mediadir / "5_1.jpg").write_bytes(b"x")
with patch("scripts.twitter_threads.download_media.subprocess.run", MagicMock(return_value=MagicMock(returncode=0))):
code = download_main(["--input", str(thread_data_json), "--output", str(mediadir)])
assert code == 0
assert (mediadir / "5_1.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
+107
View File
@@ -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 '<video controls src="./media/123_vid1.mp4"></video>' 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
+127
View File
@@ -0,0 +1,127 @@
"""Tests for scripts.twitter_threads.error_types dataclass contract."""
from __future__ import annotations
import dataclasses
import pytest
from scripts.twitter_threads.error_types import ErrorInfo, make_error, PostMetrics, PostData, ThreadData
def test_error_info_fields() -> None:
e = ErrorInfo(kind="ParseError", source="fetch", detail="bad html")
assert e.kind == "ParseError"
assert e.source == "fetch"
assert e.detail == "bad html"
def test_make_error_factory() -> None:
e = make_error("HttpError", "download_media", "404")
assert isinstance(e, ErrorInfo)
assert e.kind == "HttpError"
assert e.source == "download_media"
assert e.detail == "404"
def test_error_info_frozen() -> None:
e = ErrorInfo(kind="x", source="y", detail="z")
with pytest.raises(dataclasses.FrozenInstanceError):
e.kind = "w"
def test_error_info_slots() -> None:
e = ErrorInfo(kind="x", source="y", detail="z")
assert not hasattr(e, "__dict__")
def test_post_metrics() -> None:
m = PostMetrics(reply_count=1, repost_count=2, like_count=3, view_count=None)
assert m.reply_count == 1
assert m.repost_count == 2
assert m.like_count == 3
assert m.view_count is None
m2 = PostMetrics(reply_count=10, repost_count=20, like_count=30, view_count=100)
assert m2.view_count == 100
def test_post_data_fields() -> None:
metrics = PostMetrics(reply_count=1, repost_count=2, like_count=42, view_count=None)
p = PostData(
post_id="123",
author="Tim",
handle="NOTimothyLottes",
text="hi",
timestamp="2024-02-01T00:00:00Z",
media_urls=("https://x/i.jpg",),
reply_to_id=None,
quote_of_id=None,
metrics=metrics,
)
assert p.post_id == "123"
assert p.author == "Tim"
assert p.handle == "NOTimothyLottes"
assert p.text == "hi"
assert p.timestamp == "2024-02-01T00:00:00Z"
assert isinstance(p.media_urls, tuple)
assert p.media_urls == ("https://x/i.jpg",)
assert p.reply_to_id is None
assert p.quote_of_id is None
assert p.metrics.like_count == 42
def test_post_data_slots_frozen() -> None:
metrics = PostMetrics(reply_count=1, repost_count=2, like_count=3, view_count=None)
p = PostData(
post_id="123",
author="Tim",
handle="NOTimothyLottes",
text="hi",
timestamp="2024-02-01T00:00:00Z",
media_urls=(),
reply_to_id=None,
quote_of_id=None,
metrics=metrics,
)
assert not hasattr(p, "__dict__")
with pytest.raises(dataclasses.FrozenInstanceError):
p.post_id = "456"
def test_thread_data_fields() -> None:
metrics = PostMetrics(reply_count=1, repost_count=2, like_count=3, view_count=None)
p = PostData(
post_id="123",
author="Tim",
handle="NOTimothyLottes",
text="hi",
timestamp="2024-02-01T00:00:00Z",
media_urls=(),
reply_to_id=None,
quote_of_id=None,
metrics=metrics,
)
t = ThreadData(root_post_id="123", posts=(p,), source_url="https://x.com/NOTimothyLottes/status/123")
assert t.root_post_id == "123"
assert isinstance(t.posts, tuple)
assert len(t.posts) == 1
assert t.posts[0] is p
assert t.source_url == "https://x.com/NOTimothyLottes/status/123"
def test_thread_data_slots_frozen() -> None:
metrics = PostMetrics(reply_count=1, repost_count=2, like_count=3, view_count=None)
p = PostData(
post_id="123",
author="Tim",
handle="NOTimothyLottes",
text="hi",
timestamp="2024-02-01T00:00:00Z",
media_urls=(),
reply_to_id=None,
quote_of_id=None,
metrics=metrics,
)
t = ThreadData(root_post_id="123", posts=(p,), source_url="https://x.com/x/x/123")
assert not hasattr(t, "__dict__")
with pytest.raises(dataclasses.FrozenInstanceError):
t.source_url = "https://other"