ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

feat(twitter_threads): gallery-dl media download + inline media embeds in Markdown

Browse Source 
download_media now shells out to gallery-dl (handles X.com auth/403/video->mp4) instead of urllib, returns the downloaded files, + media_files_for_post glob helper + --cookies CLI. render_markdown embeds images with ![](...) and videos with an HTML <video controls> tag (were plain links). extract_corpus uses the package download_media (one download path). Tests rewritten for the gallery-dl API. Corpus: 4 threads, 33 media (32 img embeds + 1 video), 0 missing.
This commit is contained in:
ed
2026-07-05 19:08:43 -04:00
parent 3f5a2b0659
commit 2c680e23e4
6 changed files with 92 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_twitter_threads_media.py
+44 -64
View File
@@ -1,82 +1,62 @@
"""Tests for scripts.twitter_threads.download_media: naming and HTTP download contract."""
from __future__ import annotations
"""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
from urllib.error import URLError
import pytest
from scripts.twitter_threads.error_types import PostMetrics, PostData, ThreadData
from scripts.twitter_threads.download_media import media_names_for_post, download_media
from scripts.twitter_threads.download_media import download_media, media_files_for_post
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, media_urls: tuple[str, ...]) -> PostData:
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=media_urls, reply_to_id=None, quote_of_id=None, metrics=metrics)
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 test_media_names_for_post() -> None:
post = _post("123", ("https://pbs.twimg.com/media/A?format=jpg&name=large", "https://pbs.twimg.com/media/B?format=png&name=large"))
assert media_names_for_post(post) == ["123_img1.jpg", "123_img2.png"]
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_media_names_video() -> None:
post = _post("123", ("https://video.twimg.com/ext_tw_video/1/pu/vid/1280x720/1234567890.mp4",))
assert media_names_for_post(post) == ["123_vid1.mp4"]
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_naming(tmp_path: Path) -> None:
post = _post("123", ("https://pbs.twimg.com/media/A?format=jpg&name=large", "https://pbs.twimg.com/media/B?format=png&name=large"))
thread = ThreadData(root_post_id="123", posts=(post,), source_url="https://x.com/x/x/123")
with patch("scripts.twitter_threads.download_media.urlopen", return_value=_FakeResp(b"bytes")):
result = download_media(thread, tmp_path)
assert result.is_ok
assert (tmp_path / "123_img1.jpg").exists()
assert (tmp_path / "123_img2.png").exists()
assert result.data is not None
names = [p.name for p in result.data]
assert "123_img1.jpg" in names
assert "123_img2.png" 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_video_naming(tmp_path: Path) -> None:
post = _post("123", ("https://video.twimg.com/ext_tw_video/1/pu/vid/1280x720/1234567890.mp4",))
thread = ThreadData(root_post_id="123", posts=(post,), source_url="https://x.com/x/x/123")
with patch("scripts.twitter_threads.download_media.urlopen", return_value=_FakeResp(b"vid")):
result = download_media(thread, tmp_path)
assert (tmp_path / "123_vid1.mp4").exists()
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_download_idempotent(tmp_path: Path) -> None:
post = _post("123", ("https://pbs.twimg.com/media/A?format=jpg&name=large",))
thread = ThreadData(root_post_id="123", posts=(post,), source_url="https://x.com/x/x/123")
name = media_names_for_post(post)[0]
(tmp_path / name).write_bytes(b"already")
mock_urlopen = MagicMock()
with patch("scripts.twitter_threads.download_media.urlopen", mock_urlopen):
result = download_media(thread, tmp_path)
assert result.is_ok
mock_urlopen.assert_not_called()
assert (tmp_path / name).read_bytes() == b"already"
def test_download_http_error(tmp_path: Path) -> None:
post = _post("123", ("https://pbs.twimg.com/media/A?format=jpg&name=large",))
thread = ThreadData(root_post_id="123", posts=(post,), source_url="https://x.com/x/x/123")
with patch("scripts.twitter_threads.download_media.urlopen", side_effect=URLError("boom")):
result = download_media(thread, tmp_path)
assert result.is_ok is False
assert result.error is not None
assert result.error.kind == "HttpError"
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"]
tests/test_twitter_threads_pipeline.py
+8 -6
View File
@@ -3,13 +3,13 @@ from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch
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 media_names_for_post, download_media, main as download_main
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
@@ -59,7 +59,7 @@ def test_pipeline_html_to_markdown(tmp_path: Path) -> None:
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}
media_names = {"10": ["10_img1.jpg"]}
out = tmp_path / "thread.md"
rr = render_markdown(td, media_names, out)
assert rr.is_ok
@@ -67,7 +67,7 @@ def test_pipeline_html_to_markdown(tmp_path: Path) -> None:
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
assert "![Media 1](./media/10_img1.jpg)" in text
def test_download_cli(tmp_path: Path) -> None:
@@ -76,10 +76,12 @@ def test_download_cli(tmp_path: Path) -> None:
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")):
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_img1.jpg").exists()
assert (mediadir / "5_1.jpg").exists()
def test_render_cli(tmp_path: Path) -> None:
tests/test_twitter_threads_render.py
+3 -3
View File
@@ -77,9 +77,9 @@ def test_render_media_links(tmp_path: Path) -> None:
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
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: