feat(twitter_threads): fetch_thread.py — html.parser (local HTML) + gallery-dl subprocess (URL) + URL query-strip + CLI
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).
This commit is contained in:
@@ -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 <article> 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())
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user