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

conductor(track): init twitter_threads_extraction_20260705 — standalone Twitter/X thread extraction tooling

Browse Source 
Scripts + workflow for extracting Twitter/X posts and threads into
Markdown with associated media. Mirrors the scripts/video_analysis/
pattern. Standalone requirement: zero imports from src/, conductor/,
or scripts.video_analysis — copy-pasteable to another repo with only
gallery-dl as the external dep.

5 modules: __init__.py, error_types.py (Result[T, ErrorInfo] +
ThreadData/PostData typed dataclasses), fetch_thread.py (gallery-dl
subprocess for URLs + html.parser fallback for local HTML),
download_media.py (stdlib urllib, idempotent), render_markdown.py
(YAML front-matter + per-post sections + ./media/ links).

Reference project: C:\projects\forth\bootslop — the corpus feeds
bootslop's scripts and reference-generation pipeline. Acceptance
corpus: 8 threads (@NOTimothyLottes x6 + @VPCOMPRESSB x2) extracted
to tests/artifacts/twitter_threads_corpus/. The ?s=20 quote-share
suffix on the @VPCOMPRESSB URLs must be stripped by fetch_thread.py
before acquisition (added to FR2 as URL normalization).

5 phases / 23 tasks. 8 verification criteria (VC1-VC8). TDD red-first
on the pure-function modules (render_markdown, types, media naming).
This commit is contained in:
ed
2026-07-05 16:50:05 -04:00
parent 4c9fc99cd4
commit 0908f8fa28
6 changed files with 681 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conductor/tracks/twitter_threads_extraction_20260705/plan.md
+292
View File
@@ -0,0 +1,292 @@
# Plan: Twitter/X Thread Extraction Tooling
Track: `twitter_threads_extraction_20260705`
Branch: master (scripts + tests only; no `src/` changes, no GUI changes)
Spec: `conductor/tracks/twitter_threads_extraction_20260705/spec.md`
Standalone tooling track. The deliverable is a `scripts/twitter_threads/` package + tests + README. No application code changes.
---
## Phase 1: Scaffold + Error Types + Data Classes
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`**
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)**
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`.
HOW:
```python
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)
```
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`)**
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:
```python
@dataclass(frozen=True, slots=True)
class PostData:
post_id: str
author: str
handle: str
text: str
timestamp: str # ISO 8601
media_urls: tuple[str, ...]
reply_to_id: str | None
quote_of_id: str | None
metrics: PostMetrics
@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 ThreadData:
root_post_id: str
posts: tuple[PostData, ...]
source_url: str
```
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**
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**
```bash
git add scripts/twitter_threads/__init__.py scripts/twitter_threads/error_types.py tests/test_twitter_threads_types.py
git commit -m "feat(twitter_threads): scaffold package + error types + typed dataclasses"
```
---
## Phase 2: `render_markdown.py` (the pure-function module; TDD-first)
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`**
WHERE: `tests/test_twitter_threads_render.py`
WHAT: 5+ tests:
1. `test_render_single_post` — a `ThreadData` with 1 post; assert the YAML front-matter has the right fields; assert the post body is rendered; assert media links are `./media/<filename>`.
2. `test_render_thread` — a `ThreadData` with 3 posts (a thread); assert 3 `## Post N` sections in chronological order; assert each has the timestamp + reply-to marker.
3. `test_render_quote_tweet` — a `ThreadData` with a post that has `quote_of_id` set; assert the quote-tweet renders as a nested blockquote with a link.
4. `test_render_media_links` — a post with 2 image URLs + 1 video URL; assert 3 media links, named `<post_id>_img1.jpg`, `<post_id>_img2.jpg`, `<post_id>_vid1.mp4`.
5. `test_render_metrics_in_frontmatter` — assert `reply_count`, `repost_count`, `like_count`, `view_count` appear in the YAML front-matter; assert `view_count: null` when `view_count` is `None`.
6. `test_render_title_from_first_post` — the thread title is the first post's first line (truncated to 80 chars).
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`**
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.
HOW:
- Build the YAML front-matter from `thread.posts[0]` (the root post's metadata).
- Build the body: iterate `thread.posts` in order; for each, emit `## Post N (<timestamp>)` + optional `— reply to Post M` marker.
- For each `media_url` in `post.media_urls`: emit `[Media K](./media/<post_id>_<kind><K>.<ext>)`.
- For quote-tweets: emit a `> [Quoting @<handle>](<quoted_url>)` blockquote.
- Write to `output` via `Path.write_text(content, encoding="utf-8")`.
- 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**
```bash
git add scripts/twitter_threads/render_markdown.py tests/test_twitter_threads_render.py
git commit -m "feat(twitter_threads): render_markdown.py — YAML front-matter + per-post sections + media links"
```
---
## Phase 3: `download_media.py` (the media downloader)
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`**
WHERE: `tests/test_twitter_threads_media.py`
WHAT: 4+ tests:
1. `test_download_naming` — a post with 2 image URLs (`...?format=jpg&name=large`, `...?format=png&name=large`); assert the downloaded files are named `<post_id>_img1.jpg` and `<post_id>_img2.png`.
2. `test_download_video_naming` — a post with a video URL (`.../vid/.../1234567890.mp4`); assert the file is named `<post_id>_vid1.mp4`.
3. `test_download_idempotent` — pre-create the target file with the expected byte size; assert the download is skipped (no HTTP call made).
4. `test_download_http_error` — mock `urlopen` to raise `URLError`; assert `Result.err` with `ErrorInfo.kind == "HttpError"`.
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_<test_name>/` 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`**
WHERE: `scripts/twitter_threads/download_media.py`
WHAT: The `download_media(thread: ThreadData, output_dir: Path) -> Result[list[Path], ErrorInfo]` function.
HOW:
- For each post in `thread.posts`, for each `media_url` in `post.media_urls`:
- Derive the kind (`img` / `vid` / `gif`) from the URL or Content-Type.
- Derive the extension (`.jpg`, `.png`, `.mp4`).
- Compute the target filename: `<post_id>_<kind><index>.<ext>`.
- If the target file exists and has byte size > 0: skip (idempotent).
- Else: `urllib.request.urlopen(media_url)` → read bytes → write to target.
- Return `Result.ok(list_of_paths)` on success; `Result.err(ErrorInfo)` on the first HTTP failure.
- 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**
```bash
git add scripts/twitter_threads/download_media.py tests/test_twitter_threads_media.py
git commit -m "feat(twitter_threads): download_media.py — stdlib urllib + idempotent + typed naming"
```
---
## Phase 4: `fetch_thread.py` (the acquisition module; the network boundary)
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)**
WHERE: `tests/test_twitter_threads_fetch.py`
WHAT: 3+ tests:
1. `test_parse_html_single_post` — a small fixture HTML file (in `tests/artifacts/`) containing one tweet's text + author + timestamp + 1 image URL; assert the parsed `ThreadData` has 1 post with the right fields.
2. `test_parse_html_thread` — a fixture HTML file with a 3-post thread; assert 3 `PostData` instances in chronological order.
3. `test_parse_html_no_media` — a post with no media; assert `media_urls == ()`.
4. `test_parse_html_malformed` — a malformed HTML file; assert `Result.err` with `ErrorInfo.kind == "ParseError"`.
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**
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).
HOW:
- Subclass `HTMLParser`; walk the DOM; extract post text (the tweet-text container), author, handle, timestamp, media URLs (the `img[src]` and `video[src]` tags in the tweet container).
- Construct `PostData` instances; chain them into a `ThreadData`.
- 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)**
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.
HOW:
- Build the `gallery-dl` args: `["gallery-dl", "--dump-json", "--write-metadata", url]`. If `cookies_path` is provided, add `--cookies <cookies_path>`.
- `subprocess.run(args, capture_output=True, text=True)`.
- Parse the JSON stdout: `gallery-dl` emits one JSON object per post. Parse each into a `PostData`.
- Chain into `ThreadData`.
- On `gallery-dl` non-zero exit: return `Result.err(ErrorInfo("GalleryDlError", "fetch_thread_from_url", stderr[:500]))`.
- 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`**
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.
HOW:
```python
if __name__ == "__main__":
import argparse
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()
...
```
VERIFY: `python scripts/twitter_threads/fetch_thread.py --help` works (standalone, no `src/` imports).
- [ ] **Task 4.5: Commit Phase 4**
```bash
git add scripts/twitter_threads/fetch_thread.py tests/test_twitter_threads_fetch.py
git commit -m "feat(twitter_threads): fetch_thread.py — gallery-dl subprocess (URL) + html.parser (local HTML fallback)"
```
---
## Phase 5: README + End-to-End CLI + Verification
Focus: the standalone-usage README, an end-to-end CLI test, and the final verification.
- [ ] **Task 5.1: Write `README.md`**
WHERE: `scripts/twitter_threads/README.md`
WHAT: Per spec FR7 — prerequisites (`gallery-dl`, `cookies.txt`, Python 3.11+), usage (both `uv run` and standalone `python`), output layout, "copy to another repo" instructions, and the Strategy B/D documentation as alternatives.
HOW: Markdown; no code.
- [ ] **Task 5.2: Verify the standalone requirement (VC6 + VC7)**
WHAT:
- `python scripts/twitter_threads/fetch_thread.py --help` works from the repo root with no `uv run` (standalone invocation).
- `grep -r "from src\." scripts/twitter_threads/` returns nothing.
- `grep -r "import src\." scripts/twitter_threads/` returns nothing.
- `grep -r "from conductor\." scripts/twitter_threads/` returns nothing.
- `grep -r "from scripts\.video_analysis" scripts/twitter_threads/` returns nothing.
VERIFY: All greps return empty; the `--help` works.
- [ ] **Task 5.3: End-to-end smoke test against the 8-thread corpus (@NOTimothyLottes + @VPCOMPRESSB)**
WHAT: Run the full pipeline against all 8 reference URLs from the spec's "Reference Project + Test Corpus" section. These are the acceptance corpus for the `C:\projects\forth\bootslop` reference-generation pipeline.
URLs:
1. `https://x.com/NOTimothyLottes/status/1757198624818168210`
2. `https://x.com/NOTimothyLottes/status/1653570742762479620`
3. `https://x.com/NOTimothyLottes/status/1917646466417381426`
4. `https://x.com/NOTimothyLottes/status/1917645859791200562`
5. `https://x.com/NOTimothyLottes/status/1917644904055910502`
6. `https://x.com/NOTimothyLottes/status/1917642786804785230`
7. `https://x.com/VPCOMPRESSB/status/1991383117571957052` (note: `?s=20` query suffix must be stripped by `fetch_thread.py` before acquisition)
8. `https://x.com/VPCOMPRESSB/status/1987744335333622188` (note: `?s=20` query suffix must be stripped)
HOW (per URL):
```bash
uv run python -m scripts.twitter_threads.fetch_thread "<url>" --output ./tests/artifacts/twitter_threads_corpus/ --cookies ./cookies.txt
uv run python -m scripts.twitter_threads.download_media --input ./tests/artifacts/twitter_threads_corpus/<id>/thread_data.json --output ./tests/artifacts/twitter_threads_corpus/<id>/media/
uv run python -m scripts.twitter_threads.render_markdown --input ./tests/artifacts/twitter_threads_corpus/<id>/thread_data.json --media-dir ./tests/artifacts/twitter_threads_corpus/<id>/media/ --output ./tests/artifacts/twitter_threads_corpus/<id>/thread.md
```
VERIFY: 8 `thread.md` files exist with YAML front-matter + post sections + media links; 8 `media/` directories have the downloaded assets. This is the acceptance corpus for the track — if all 8 extract cleanly, the track ships and the corpus is handed off to `bootslop`. This is a manual verification (network boundary; the user provides the cookies.txt).
- [ ] **Task 5.4: Run the full test suite for the new tests**
WHAT: `uv run pytest tests/test_twitter_threads_types.py tests/test_twitter_threads_render.py tests/test_twitter_threads_media.py tests/test_twitter_threads_fetch.py -v`
VERIFY: All tests pass. This is the batched verification (the only verification that matters per the Isolated-Pass Verification Fallacy rule).
- [ ] **Task 5.5: Commit Phase 5 + the README**
```bash
git add scripts/twitter_threads/README.md
git commit -m "docs(twitter_threads): README — standalone usage + copy-to-another-repo instructions"
```
- [ ] **Task 5.6: Conductor — User Manual Verification (Protocol in workflow.md)**
Present the verification results to the user. PAUSE for user confirmation before marking the track complete.
- [ ] **Task 5.7: Mark the track complete + write the TRACK_COMPLETION report**
WHERE: `docs/reports/TRACK_COMPLETION_twitter_threads_extraction_20260705.md`
WHAT: A short report (per the `tier2_autonomous_sandbox_20260616` precedent): what was done, files created, tests passing, the standalone-requirement verification, and any deferred items.
HOW: Commit + update `conductor/tracks/twitter_threads_extraction_20260705/state.toml` to `status = "completed"`.