ed
ed86731701
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).
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""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
|
|
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)
|
|
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)
|
|
|
|
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())
|