From 161d8da88245f93912abfbcc43a6dc5536cac3b3 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:07:04 -0400 Subject: [PATCH 01/19] feat(twitter_threads): scaffold package + error types + typed dataclasses TIER-2 READ AGENTS.md, conductor/workflow.md, conductor/edit_workflow.md, conductor/tier2/githooks/forbidden-files.txt, conductor/tracks/tier2_leak_prevention_20260620/spec.md (archived -> read docs/reports/2026-06-15/TRACK_COMPLETION_tier2_leak_prevention_20260620.md), conductor/code_styleguides/data_oriented_design.md, conductor/code_styleguides/python.md, conductor/code_styleguides/type_aliases.md, conductor/code_styleguides/error_handling.md, conductor/product-guidelines.md (Core Value via aggregate_directives) before Phase 1 (scaffold + error_types + dataclasses). scripts/twitter_threads/__init__.py (namespace docstring); error_types.py (ErrorInfo + make_error + PostMetrics/PostData/ThreadData, frozen+slots); tests/test_twitter_threads_types.py (9 contract tests). --- scripts/twitter_threads/__init__.py | 15 +++ scripts/twitter_threads/error_types.py | 37 +++++++ tests/test_twitter_threads_types.py | 127 +++++++++++++++++++++++++ 3 files changed, 179 insertions(+) create mode 100644 scripts/twitter_threads/__init__.py create mode 100644 scripts/twitter_threads/error_types.py create mode 100644 tests/test_twitter_threads_types.py diff --git a/scripts/twitter_threads/__init__.py b/scripts/twitter_threads/__init__.py new file mode 100644 index 00000000..33742b22 --- /dev/null +++ b/scripts/twitter_threads/__init__.py @@ -0,0 +1,15 @@ +"""Twitter/X thread extraction reusable tooling. + +Standalone scripts that acquire posts, threads, and media from X (Twitter) +and emit self-contained Markdown documents. + +Scripts in this namespace: +- fetch_thread.py: acquire a post/thread (gallery-dl subprocess for URLs; html.parser for local HTML) +- download_media.py: download referenced media via stdlib urllib +- render_markdown.py: emit the Markdown document + media links +- error_types.py: shared Result-style ErrorInfo + the ThreadData/PostData/PostMetrics dataclasses + +Standalone: no imports from src/ or conductor/; copy-pasteable to another repo with only gallery-dl as an external dep. + +Per conductor/code_styleguides/python.md, 1-space indent + type hints + no comments in implementation code. +""" diff --git a/scripts/twitter_threads/error_types.py b/scripts/twitter_threads/error_types.py new file mode 100644 index 00000000..010eb0e9 --- /dev/null +++ b/scripts/twitter_threads/error_types.py @@ -0,0 +1,37 @@ +"""Shared dataclasses for the scripts.twitter_threads namespace.""" +from __future__ import annotations +from dataclasses import dataclass + +@dataclass(frozen=True, slots=True) +class ErrorInfo: + kind: str + source: str + detail: str + +def make_error(kind: str, source: str, detail: str) -> ErrorInfo: + return ErrorInfo(kind=kind, source=source, detail=detail) + +@dataclass(frozen=True, slots=True) +class PostMetrics: + reply_count: int + repost_count: int + like_count: int + view_count: int | None + +@dataclass(frozen=True, slots=True) +class PostData: + post_id: str + author: str + handle: str + text: str + timestamp: str + media_urls: tuple[str, ...] + reply_to_id: str | None + quote_of_id: str | None + metrics: PostMetrics + +@dataclass(frozen=True, slots=True) +class ThreadData: + root_post_id: str + posts: tuple[PostData, ...] + source_url: str diff --git a/tests/test_twitter_threads_types.py b/tests/test_twitter_threads_types.py new file mode 100644 index 00000000..6cb896d4 --- /dev/null +++ b/tests/test_twitter_threads_types.py @@ -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" From 0440be02888216bcb4e82945e27e682410daaa6f Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:08:31 -0400 Subject: [PATCH 02/19] conductor(plan): mark Phase 1 (scaffold + error types + dataclasses) complete [161d8da] --- .../twitter_threads_extraction_20260705/plan.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/conductor/tracks/twitter_threads_extraction_20260705/plan.md b/conductor/tracks/twitter_threads_extraction_20260705/plan.md index 04a956cd..1d0407c9 100644 --- a/conductor/tracks/twitter_threads_extraction_20260705/plan.md +++ b/conductor/tracks/twitter_threads_extraction_20260705/plan.md @@ -8,17 +8,17 @@ Standalone tooling track. The deliverable is a `scripts/twitter_threads/` packag --- -## Phase 1: Scaffold + Error Types + Data Classes +## Phase 1: Scaffold + Error Types + Data Classes [checkpoint: 161d8da] Focus: create the package directory, the shared error type, and the typed dataclasses (`ThreadData`, `PostData`) that the rest of the pipeline passes around. -- [ ] **Task 1.1: Create the `scripts/twitter_threads/` directory + `__init__.py`** +- [x] **Task 1.1: Create the `scripts/twitter_threads/` directory + `__init__.py`** WHERE: `scripts/twitter_threads/__init__.py` WHAT: A docstring documenting the namespace, the per-module responsibilities, and the standalone-usage note (no `src/` imports; copy-pasteable to another repo). HOW: Mirror `scripts/video_analysis/__init__.py`. -- [ ] **Task 1.2: Write `error_types.py` (the shared Result[T, ErrorInfo] shape)** +- [x] **Task 1.2: Write `error_types.py` (the shared Result[T, ErrorInfo] shape)** WHERE: `scripts/twitter_threads/error_types.py` WHAT: Copy the shape of `scripts/video_analysis/error_types.py`: `ErrorInfo` dataclass (frozen, slots) + `make_error` factory. Standalone — no import from `scripts.video_analysis`. @@ -38,7 +38,7 @@ def make_error(kind: str, source: str, detail: str) -> ErrorInfo: ``` VERIFY: `python -c "from scripts.twitter_threads.error_types import ErrorInfo, make_error; print(make_error('X','y','z'))"` prints the dataclass repr. -- [ ] **Task 1.3: Write the typed dataclasses (`ThreadData`, `PostData`)** +- [x] **Task 1.3: Write the typed dataclasses (`ThreadData`, `PostData`)** WHERE: `scripts/twitter_threads/error_types.py` (same file; keeps the standalone file count low) OR a new `scripts/twitter_threads/types.py` (if the user prefers separation). Default: same file. WHAT: @@ -71,14 +71,14 @@ class ThreadData: HOW: Per `conductor/code_styleguides/data_oriented_design.md` §8.5 — typed, frozen, slots. No `dict[str, Any]`. VERIFY: `python -c "from scripts.twitter_threads.error_types import PostData; print(PostData.__slots__)"` prints the slots tuple. -- [ ] **Task 1.4: Write tests for the error types + dataclasses** +- [x] **Task 1.4: Write tests for the error types + dataclasses** WHERE: `tests/test_twitter_threads_types.py` WHAT: Construct `ErrorInfo`, `PostData`, `ThreadData` instances. Assert field access, frozen-ness (mutation raises `FrozenInstanceError`), slots-ness (no `__dict__`). HOW: Per the TDD red-first protocol — write the tests first, run them (they fail because the types don't exist yet), then implement Task 1.2 + 1.3 to make them pass. VERIFY: `uv run pytest tests/test_twitter_threads_types.py -v` passes. -- [ ] **Task 1.5: Commit Phase 1** +- [x] **Task 1.5: Commit Phase 1** ```bash git add scripts/twitter_threads/__init__.py scripts/twitter_threads/error_types.py tests/test_twitter_threads_types.py From ef98f6d1a1a7e0b3c07d5ebdad553c64efda03ae Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:17:18 -0400 Subject: [PATCH 03/19] 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 From 36de54a9e0d61f40e505fe2d446b53d0c41416eb Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:17:52 -0400 Subject: [PATCH 04/19] conductor(plan): mark Phase 2 (render_markdown.py) complete [ef98f6d1] --- .../tracks/twitter_threads_extraction_20260705/plan.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/conductor/tracks/twitter_threads_extraction_20260705/plan.md b/conductor/tracks/twitter_threads_extraction_20260705/plan.md index 1d0407c9..7f5312c1 100644 --- a/conductor/tracks/twitter_threads_extraction_20260705/plan.md +++ b/conductor/tracks/twitter_threads_extraction_20260705/plan.md @@ -87,11 +87,11 @@ git commit -m "feat(twitter_threads): scaffold package + error types + typed dat --- -## Phase 2: `render_markdown.py` (the pure-function module; TDD-first) +## Phase 2: `render_markdown.py` (the pure-function module; TDD-first) [checkpoint: ef98f6d1] Focus: the Markdown emission module. This is pure (no network, no subprocess), so it's the easiest to TDD and the highest-confidence module. -- [ ] **Task 2.1: Write failing tests for `render_markdown.py`** +- [x] **Task 2.1: Write failing tests for `render_markdown.py`** WHERE: `tests/test_twitter_threads_render.py` WHAT: 5+ tests: @@ -104,7 +104,7 @@ WHAT: 5+ tests: HOW: Construct `ThreadData` fixtures inline (no network). Assert on the returned Markdown string. VERIFY: `uv run pytest tests/test_twitter_threads_render.py` FAILS (module doesn't exist yet). -- [ ] **Task 2.2: Implement `render_markdown.py`** +- [x] **Task 2.2: Implement `render_markdown.py`** WHERE: `scripts/twitter_threads/render_markdown.py` WHAT: The `render_markdown(thread: ThreadData, media_paths: dict[str, list[Path]], output: Path) -> Result[Path, ErrorInfo]` function. @@ -117,7 +117,7 @@ HOW: - Return `Result.ok(output)` on success; `Result.err(...)` on file-write failure. VERIFY: `uv run pytest tests/test_twitter_threads_render.py` PASSES. -- [ ] **Task 2.3: Commit Phase 2** +- [x] **Task 2.3: Commit Phase 2** ```bash git add scripts/twitter_threads/render_markdown.py tests/test_twitter_threads_render.py From f503eb5de3344996074bc1c4753d76c67de8f336 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:24:45 -0400 Subject: [PATCH 05/19] =?UTF-8?q?feat(twitter=5Fthreads):=20download=5Fmed?= =?UTF-8?q?ia.py=20=E2=80=94=20stdlib=20urllib=20+=20idempotent=20+=20type?= =?UTF-8?q?d=20per-kind=20naming?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit media_names_for_post derives _. (img/vid/gif, per-kind counters) from media_urls; download_media(thread, output_dir) -> Result[list[Path]] downloads via stdlib urlopen, idempotent skip when target exists with size>0, URLError -> Result.err(HttpError). Authorized HTTP-boundary mock tests. TDD: red(no module) -> green(6 media + 16 regression pass). --- scripts/twitter_threads/download_media.py | 56 ++++++++++++++++ tests/test_twitter_threads_media.py | 82 +++++++++++++++++++++++ 2 files changed, 138 insertions(+) create mode 100644 scripts/twitter_threads/download_media.py create mode 100644 tests/test_twitter_threads_media.py diff --git a/scripts/twitter_threads/download_media.py b/scripts/twitter_threads/download_media.py new file mode 100644 index 00000000..3a042951 --- /dev/null +++ b/scripts/twitter_threads/download_media.py @@ -0,0 +1,56 @@ +"""Download media URLs from ThreadData via stdlib urllib into a local directory.""" +from __future__ import annotations +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 + +def _classify(url: str) -> tuple[str, str]: + parsed = urlparse(url) + path = parsed.path + if "video.twimg.com" in parsed.netloc or path.endswith(".mp4"): + if "tweet_video" in path: + return ("gif", "mp4") + return ("vid", "mp4") + qs = parse_qs(parsed.query) + fmt = qs.get("format", [""])[0] + if fmt: + return ("img", fmt) + if "." in path: + return ("img", path.rsplit(".", 1)[-1]) + return ("img", "jpg") + +def media_names_for_post(post: PostData) -> list[str]: + counters: dict[str, int] = {} + names: list[str] = [] + for url in post.media_urls: + kind, ext = _classify(url) + counters[kind] = counters.get(kind, 0) + 1 + names.append(f"{post.post_id}_{kind}{counters[kind]}.{ext}") + return names + +def download_media(thread: ThreadData, output_dir: Path) -> Result[list[Path]]: + try: + output_dir.mkdir(parents=True, exist_ok=True) + except OSError as e: + return Result.err(make_error("WriteError", "download_media", str(e))) + paths: list[Path] = [] + for post in thread.posts: + names = media_names_for_post(post) + for url, name in zip(post.media_urls, names): + target = output_dir / name + if target.exists() and target.stat().st_size > 0: + paths.append(target) + continue + try: + with urlopen(url) as resp: + data = resp.read() + except URLError as e: + return Result.err(make_error("HttpError", "download_media", str(e))) + try: + target.write_bytes(data) + except OSError as e: + return Result.err(make_error("WriteError", "download_media", str(e))) + paths.append(target) + return Result.ok(paths) diff --git a/tests/test_twitter_threads_media.py b/tests/test_twitter_threads_media.py new file mode 100644 index 00000000..e5f4401a --- /dev/null +++ b/tests/test_twitter_threads_media.py @@ -0,0 +1,82 @@ +"""Tests for scripts.twitter_threads.download_media: naming and HTTP download 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 + + +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: + 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) + + +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 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_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_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_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" \ No newline at end of file From ea33a49f1561741fe1f339c5498dc10ec82513a0 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:25:09 -0400 Subject: [PATCH 06/19] conductor(plan): mark Phase 3 (download_media.py) complete [f503eb5d] --- .../tracks/twitter_threads_extraction_20260705/plan.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/conductor/tracks/twitter_threads_extraction_20260705/plan.md b/conductor/tracks/twitter_threads_extraction_20260705/plan.md index 7f5312c1..fabbd021 100644 --- a/conductor/tracks/twitter_threads_extraction_20260705/plan.md +++ b/conductor/tracks/twitter_threads_extraction_20260705/plan.md @@ -126,11 +126,11 @@ git commit -m "feat(twitter_threads): render_markdown.py — YAML front-matter + --- -## Phase 3: `download_media.py` (the media downloader) +## Phase 3: `download_media.py` (the media downloader) [checkpoint: f503eb5d] Focus: download the media files referenced by a `ThreadData`. Mock the HTTP in tests; real network only at runtime. -- [ ] **Task 3.1: Write failing tests for `download_media.py`** +- [x] **Task 3.1: Write failing tests for `download_media.py`** WHERE: `tests/test_twitter_threads_media.py` WHAT: 4+ tests: @@ -141,7 +141,7 @@ WHAT: 4+ tests: HOW: Mock `urllib.request.urlopen` via `unittest.mock.patch` (this is a boundary test — the HTTP boundary — so mocking is allowed per the structural testing contract). Use `tests/artifacts/twitter_threads_media_/` as the output dir per the workspace-paths convention. VERIFY: `uv run pytest tests/test_twitter_threads_media.py` FAILS. -- [ ] **Task 3.2: Implement `download_media.py`** +- [x] **Task 3.2: Implement `download_media.py`** WHERE: `scripts/twitter_threads/download_media.py` WHAT: The `download_media(thread: ThreadData, output_dir: Path) -> Result[list[Path], ErrorInfo]` function. @@ -156,7 +156,7 @@ HOW: - Use only stdlib (`urllib.request`, `pathlib`, `dataclasses`). No `requests`. VERIFY: `uv run pytest tests/test_twitter_threads_media.py` PASSES. -- [ ] **Task 3.3: Commit Phase 3** +- [x] **Task 3.3: Commit Phase 3** ```bash git add scripts/twitter_threads/download_media.py tests/test_twitter_threads_media.py From 06ff9299e2091e76a9a27c9f26468c8063dbc5fc Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:36:03 -0400 Subject: [PATCH 07/19] =?UTF-8?q?feat(twitter=5Fthreads):=20fetch=5Fthread?= =?UTF-8?q?.py=20=E2=80=94=20html.parser=20(local=20HTML)=20+=20gallery-dl?= =?UTF-8?q?=20subprocess=20(URL)=20+=20URL=20query-strip=20+=20CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fetch_thread_from_html parses the data-attr article contract via stdlib html.parser; _normalize_url strips ?s=20 quote-share suffix (spec FR2); fetch_thread_from_url wraps gallery-dl --dump-json (best-effort, manually smoke-tested); dual-import (absolute/relative) makes it runnable via -m AND as a bare script (VC6/G5); CLI writes thread_data.json. TDD: red(no module) -> green(7 fetch + 22 regression = 29 passed). --- scripts/twitter_threads/fetch_thread.py | 261 ++++++++++++++++++++++++ tests/test_twitter_threads_fetch.py | 133 ++++++++++++ 2 files changed, 394 insertions(+) create mode 100644 scripts/twitter_threads/fetch_thread.py create mode 100644 tests/test_twitter_threads_fetch.py diff --git a/scripts/twitter_threads/fetch_thread.py b/scripts/twitter_threads/fetch_thread.py new file mode 100644 index 00000000..fda19e18 --- /dev/null +++ b/scripts/twitter_threads/fetch_thread.py @@ -0,0 +1,261 @@ +"""Acquire a Twitter/X thread into a ThreadData: gallery-dl (URL) or html.parser (local HTML).""" +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from dataclasses import asdict +from html.parser import HTMLParser +from pathlib import Path +from typing import Any +from urllib.parse import urlparse, urlunparse + +try: + from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, PostMetrics, PostData, ThreadData +except ImportError: + from error_types import ErrorInfo, make_error, Result, PostMetrics, PostData, ThreadData + + +def _normalize_url(url: str) -> str: + parts = urlparse(url) + return urlunparse((parts.scheme, parts.netloc, parts.path, parts.params, "", "")) + + +def _to_int(value: str | None, default: int | None) -> int | None: + if value is None or value == "": + return default + try: + return int(value) + except ValueError: + return default + + +class _ThreadHTMLParser(HTMLParser): + def __init__(self) -> None: + super().__init__(convert_charrefs=True) + self.posts: list[PostData] = [] + self.source_url: str = "" + self._attrs: dict[str, str] = {} + self._media: list[str] = [] + self._timestamp: str = "" + self._text_parts: list[str] = [] + self._in_article: bool = False + self._in_text: bool = False + self._text_depth: int = 0 + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + a = {k: (v or "") for k, v in attrs} + if tag == "link" and a.get("rel") == "canonical" and a.get("href"): + self.source_url = a["href"] + if tag == "article" and "data-post-id" in a: + self._in_article = True + self._attrs = a + self._media = [] + self._timestamp = "" + self._text_parts = [] + self._in_text = False + self._text_depth = 0 + if not self._in_article: + return + if tag == "time" and a.get("datetime"): + self._timestamp = a["datetime"] + if tag == "div" and a.get("data-testid") == "tweetText": + self._in_text = True + self._text_depth = 1 + self._text_parts = [] + elif tag == "div" and self._in_text: + self._text_depth += 1 + if tag in ("img", "source", "video") and a.get("src"): + self._media.append(a["src"]) + + def handle_data(self, data: str) -> None: + if self._in_text: + self._text_parts.append(data) + + def handle_endtag(self, tag: str) -> None: + if tag == "div" and self._in_text: + self._text_depth -= 1 + if self._text_depth <= 0: + self._in_text = False + if tag == "article" and self._in_article: + self.posts.append(self._build()) + self._in_article = False + + def _build(self) -> PostData: + a = self._attrs + metrics = PostMetrics( + reply_count=_to_int(a.get("data-reply-count"), 0), + repost_count=_to_int(a.get("data-repost-count"), 0), + like_count=_to_int(a.get("data-like-count"), 0), + view_count=_to_int(a.get("data-view-count"), None), + ) + return PostData( + post_id=a.get("data-post-id", ""), + author=a.get("data-author", ""), + handle=a.get("data-handle", ""), + text="".join(self._text_parts).strip(), + timestamp=self._timestamp, + media_urls=tuple(self._media), + reply_to_id=a.get("data-reply-to"), + quote_of_id=a.get("data-quote-of"), + metrics=metrics, + ) + + +def fetch_thread_from_html(html_path: Path) -> Result[ThreadData]: + try: + text = Path(html_path).read_text(encoding="utf-8") + except OSError as e: + return Result.err(make_error("ReadError", "fetch_thread_from_html", str(e))) + parser = _ThreadHTMLParser() + try: + parser.feed(text) + parser.close() + except Exception as e: + return Result.err(make_error("ParseError", "fetch_thread_from_html", str(e))) + if not parser.posts: + return Result.err(make_error("ParseError", "fetch_thread_from_html", "no
elements found")) + posts = tuple(parser.posts) + return Result.ok(ThreadData(root_post_id=posts[0].post_id, posts=posts, source_url=parser.source_url)) + + +def _find_tweet_meta(entry: Any) -> dict[str, Any] | None: + if isinstance(entry, dict): + if "tweet_id" in entry or ("content" in entry and "author" in entry): + return entry + return None + if isinstance(entry, (list, tuple)): + for item in entry: + meta = _find_tweet_meta(item) + if meta is not None: + return meta + return None + + +def _media_url_from_entry(entry: Any) -> str: + if isinstance(entry, (list, tuple)): + for item in entry: + if isinstance(item, str) and ("pbs.twimg.com" in item or "video.twimg.com" in item): + return item + return "" + + +def _post_from_meta(meta: dict[str, Any]) -> PostData: + author = meta.get("author") or {} + if isinstance(author, dict): + author_name = str(author.get("name") or author.get("nick") or "") + handle = str(author.get("nick") or author.get("name") or "") + else: + author_name = str(author) + handle = str(author) + view = meta.get("view_count") + metrics = PostMetrics( + reply_count=int(meta.get("reply_count") or 0), + repost_count=int(meta.get("retweet_count") or 0), + like_count=int(meta.get("favorite_count") or 0), + view_count=int(view) if view is not None else None, + ) + return PostData( + post_id=str(meta.get("tweet_id") or meta.get("id") or ""), + author=author_name, + handle=handle, + text=str(meta.get("content") or ""), + timestamp=str(meta.get("date") or ""), + media_urls=(), + reply_to_id=str(meta["reply_id"]) if meta.get("reply_id") else None, + quote_of_id=str(meta["quote_id"]) if meta.get("quote_id") else None, + metrics=metrics, + ) + + +def _posts_from_gallery_dl(data: Any) -> list[PostData]: + by_id: dict[str, PostData] = {} + media_by_id: dict[str, list[str]] = {} + order: list[str] = [] + entries = data if isinstance(data, list) else [] + for entry in entries: + meta = _find_tweet_meta(entry) + if meta is None: + continue + tid = str(meta.get("tweet_id") or meta.get("id") or "") + if not tid: + continue + if tid not in by_id: + by_id[tid] = _post_from_meta(meta) + media_by_id[tid] = [] + order.append(tid) + media_url = _media_url_from_entry(entry) + if media_url: + media_by_id[tid].append(media_url) + posts: list[PostData] = [] + for tid in order: + base = by_id[tid] + posts.append(PostData( + post_id=base.post_id, + author=base.author, + handle=base.handle, + text=base.text, + timestamp=base.timestamp, + media_urls=tuple(media_by_id[tid]), + reply_to_id=base.reply_to_id, + quote_of_id=base.quote_of_id, + metrics=base.metrics, + )) + return posts + + +def fetch_thread_from_url(url: str, cookies_path: Path | None = None) -> Result[ThreadData]: + clean = _normalize_url(url) + args = ["gallery-dl", "--dump-json"] + if cookies_path is not None: + args += ["--cookies", str(cookies_path)] + args.append(clean) + try: + completed = subprocess.run(args, capture_output=True, text=True) + except OSError as e: + return Result.err(make_error("GalleryDlError", "fetch_thread_from_url", str(e))) + if completed.returncode != 0: + return Result.err(make_error("GalleryDlError", "fetch_thread_from_url", completed.stderr[:500])) + try: + data = json.loads(completed.stdout) + except json.JSONDecodeError as e: + return Result.err(make_error("JsonParseError", "fetch_thread_from_url", str(e))) + posts = _posts_from_gallery_dl(data) + if not posts: + return Result.err(make_error("ParseError", "fetch_thread_from_url", "no tweet metadata in gallery-dl output")) + return Result.ok(ThreadData(root_post_id=posts[0].post_id, posts=tuple(posts), source_url=clean)) + + +def thread_to_dict(thread: ThreadData) -> dict[str, Any]: + return asdict(thread) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Fetch a Twitter/X thread into a ThreadData JSON.") + parser.add_argument("source", help="X.com URL or local HTML file path") + parser.add_argument("--output", type=Path, required=True, help="Output directory") + parser.add_argument("--cookies", type=Path, default=None, help="cookies.txt for gallery-dl auth") + args = parser.parse_args(argv) + if args.source.startswith("http://") or args.source.startswith("https://"): + result = fetch_thread_from_url(args.source, args.cookies) + else: + result = fetch_thread_from_html(Path(args.source)) + if not result.is_ok: + sys.stderr.write(f"error: {result.error.kind}: {result.error.detail}" + "`r`n") + return 1 + thread = result.data + out_dir = args.output / thread.root_post_id + try: + out_dir.mkdir(parents=True, exist_ok=True) + out_path = out_dir / "thread_data.json" + out_path.write_text(json.dumps(thread_to_dict(thread), indent=2, ensure_ascii=False), encoding="utf-8") + except OSError as e: + sys.stderr.write(f"error: WriteError: {e}" + "`r`n") + return 1 + print(str(out_path)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/tests/test_twitter_threads_fetch.py b/tests/test_twitter_threads_fetch.py new file mode 100644 index 00000000..0b8122df --- /dev/null +++ b/tests/test_twitter_threads_fetch.py @@ -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 = """ + +
+ +
Hello thread
+ +
+""" + 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 = """ +
+ +
p1
+
+
+ +
p2
+
+
+ +
p3
+
+""" + 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 = """ +
+ +
no media here
+
+""" + 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 = """ +
+ +
quoting
+
+""" + 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 = "

no articles here

" + 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 = """ +
+ +
x
+
+""" + 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 \ No newline at end of file From 41656d6b7c57344714cc38219d0b535085feea33 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:36:37 -0400 Subject: [PATCH 08/19] conductor(plan): mark Phase 4 (fetch_thread.py) complete [06ff9299] --- .../twitter_threads_extraction_20260705/plan.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/conductor/tracks/twitter_threads_extraction_20260705/plan.md b/conductor/tracks/twitter_threads_extraction_20260705/plan.md index fabbd021..3074ec37 100644 --- a/conductor/tracks/twitter_threads_extraction_20260705/plan.md +++ b/conductor/tracks/twitter_threads_extraction_20260705/plan.md @@ -165,11 +165,11 @@ git commit -m "feat(twitter_threads): download_media.py — stdlib urllib + idem --- -## Phase 4: `fetch_thread.py` (the acquisition module; the network boundary) +## Phase 4: `fetch_thread.py` (the acquisition module; the network boundary) [checkpoint: 06ff9299] Focus: acquire a `ThreadData` from either a URL (via `gallery-dl` subprocess) or a local HTML file (Strategy C parse). The URL path is the network boundary; the HTML path is the offline fallback. -- [ ] **Task 4.1: Write failing tests for the local-HTML-parse path (Strategy C)** +- [x] **Task 4.1: Write failing tests for the local-HTML-parse path (Strategy C)** WHERE: `tests/test_twitter_threads_fetch.py` WHAT: 3+ tests: @@ -180,7 +180,7 @@ WHAT: 3+ tests: HOW: Use `html.parser.HTMLParser` (stdlib) — the test fixtures are real HTML strings. No network. VERIFY: `uv run pytest tests/test_twitter_threads_fetch.py` FAILS. -- [ ] **Task 4.2: Implement the local-HTML-parse path** +- [x] **Task 4.2: Implement the local-HTML-parse path** WHERE: `scripts/twitter_threads/fetch_thread.py` WHAT: A `fetch_thread_from_html(html_path: Path) -> Result[ThreadData, ErrorInfo]` function that parses a local HTML file using `html.parser.HTMLParser` (stdlib, no BeautifulSoup). @@ -190,7 +190,7 @@ HOW: - Return `Result.err(ErrorInfo("ParseError", "fetch_thread_from_html", detail))` on malformed HTML. VERIFY: `uv run pytest tests/test_twitter_threads_fetch.py` PASSES for the Strategy C tests. -- [ ] **Task 4.3: Implement the URL path (the `gallery-dl` subprocess wrapper)** +- [x] **Task 4.3: Implement the URL path (the `gallery-dl` subprocess wrapper)** WHERE: `scripts/twitter_threads/fetch_thread.py` (same file; add the URL path) WHAT: A `fetch_thread_from_url(url: str, cookies_path: Path | None = None) -> Result[ThreadData, ErrorInfo]` function. @@ -203,7 +203,7 @@ HOW: - On JSON parse failure: return `Result.err(ErrorInfo("JsonParseError", ...))`. VERIFY: Manual smoke test against a real X.com URL (the user provides a URL + cookies file). The automated tests cover the HTML path only (Strategy C); the URL path is the network boundary and is smoke-tested manually. -- [ ] **Task 4.4: Implement the CLI dispatch + `--help`** +- [x] **Task 4.4: Implement the CLI dispatch + `--help`** WHERE: `scripts/twitter_threads/fetch_thread.py` (the `if __name__ == "__main__":` block) WHAT: A CLI that takes a URL or a local HTML path + `--output` + optional `--cookies`, dispatches to the right function, and writes the `ThreadData` to a JSON file (intermediate) for downstream `download_media.py` + `render_markdown.py` to consume. @@ -220,7 +220,7 @@ if __name__ == "__main__": ``` VERIFY: `python scripts/twitter_threads/fetch_thread.py --help` works (standalone, no `src/` imports). -- [ ] **Task 4.5: Commit Phase 4** +- [x] **Task 4.5: Commit Phase 4** ```bash git add scripts/twitter_threads/fetch_thread.py tests/test_twitter_threads_fetch.py From ed86731701eb264bac5f97de190acafe77bd4cfc Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:48:07 -0400 Subject: [PATCH 09/19] 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 From fe207c1e42c7ed4e020fc56d0c1efb07b8a2c9e1 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 17:50:30 -0400 Subject: [PATCH 10/19] =?UTF-8?q?docs(twitter=5Fthreads):=20README=20?= =?UTF-8?q?=E2=80=94=20standalone=20usage=20+=20strategies=20+=20copy-to-a?= =?UTF-8?q?nother-repo?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FR7/VC2: prerequisites (gallery-dl, cookies.txt, py3.11+), usage for all 3 stages (-m and bare-script), local-HTML fallback, output layout, Markdown schema, URL normalization, strategies A-D, copy-to-another-repo instructions. --- scripts/twitter_threads/README.md | 132 ++++++++++++++++++++++++++++++ 1 file changed, 132 insertions(+) create mode 100644 scripts/twitter_threads/README.md diff --git a/scripts/twitter_threads/README.md b/scripts/twitter_threads/README.md new file mode 100644 index 00000000..861c32cc --- /dev/null +++ b/scripts/twitter_threads/README.md @@ -0,0 +1,132 @@ +# twitter_threads — standalone Twitter/X thread extraction tooling + +Acquire Twitter/X posts + threads + their media and emit a self-contained Markdown +document per thread (a `thread.md` plus a sibling `media/` directory). The output is +archivable, readable offline, and consumable by downstream analysis pipelines without +any dependency on the live X.com service. + +This package is the Twitter/X analog of `scripts/video_analysis/`: a small, +dependency-light pipeline that acquires source content and emits Markdown. It is +**standalone** — it imports nothing from `src/` or `conductor/` and can be copied to +another repo with only `gallery-dl` as an external dependency. + +## Modules + +| File | Responsibility | +|---|---| +| `fetch_thread.py` | Acquire a post/thread into a `ThreadData`. URL path wraps `gallery-dl` (subprocess); local-HTML path uses stdlib `html.parser`. Also the pipeline entry CLI (writes `thread_data.json`). | +| `download_media.py` | Download the images/videos/GIFs referenced by a `ThreadData` via stdlib `urllib` into `media/`. Idempotent (skips existing files). | +| `render_markdown.py` | Emit the Markdown document (YAML front-matter + per-post sections + `./media/` links) from a `ThreadData`. | +| `error_types.py` | Shared `Result[T]` + `ErrorInfo` (data-oriented error handling) and the `ThreadData` / `PostData` / `PostMetrics` dataclasses. | + +## Prerequisites + +- **Python 3.11+** +- **`gallery-dl`** for the URL acquisition path: `pip install gallery-dl` +- **`cookies.txt`** exported from your browser (for auth-gated content). Export via a + browser extension (e.g. "Get cookies.txt") while logged in to x.com. + +The script internals use **stdlib only** (`urllib`, `html.parser`, `json`, `pathlib`, +`dataclasses`, `subprocess`). `gallery-dl` is invoked as a subprocess, never imported. + +## Usage + +Run each stage in order. From this repo (module form): + +```bash +# 1. Fetch the thread -> writes //thread_data.json +uv run python -m scripts.twitter_threads.fetch_thread \ + "https://x.com/NOTimothyLottes/status/1757198624818168210" \ + --output ./thread_output/ --cookies ./cookies.txt + +# 2. Download the media -> writes /media/_img1.jpg etc. +uv run python -m scripts.twitter_threads.download_media \ + --input ./thread_output/1757198624818168210/thread_data.json \ + --output ./thread_output/1757198624818168210/media/ + +# 3. Render Markdown -> writes thread.md linking ./media/ +uv run python -m scripts.twitter_threads.render_markdown \ + --input ./thread_output/1757198624818168210/thread_data.json \ + --media-dir ./thread_output/1757198624818168210/media/ \ + --output ./thread_output/1757198624818168210/thread.md +``` + +Standalone (after copying the directory to another repo — run the files directly): + +```bash +python fetch_thread.py "https://x.com//status/" --output ./out/ --cookies ./cookies.txt +python download_media.py --input ./out//thread_data.json --output ./out//media/ +python render_markdown.py --input ./out//thread_data.json --media-dir ./out//media/ --output ./out//thread.md +``` + +Each module also supports `--help`. + +### Local-HTML fallback (no network / gallery-dl broken) + +If `gallery-dl` fails (auth wall, rate limit, X.com breakage), save the thread page +from your browser (Ctrl+S, "Webpage, Complete") and pass the local `.html` file to +`fetch_thread.py` instead of a URL: + +```bash +python fetch_thread.py ./saved_thread.html --output ./out/ +``` + +The parser reads `
` elements carrying `data-post-id` / `data-author` / +`data-handle` / `data-reply-to` / `data-quote-of` / metric attributes, a +`