ed
2c680e23e4
download_media now shells out to gallery-dl (handles X.com auth/403/video->mp4) instead of urllib, returns the downloaded files, + media_files_for_post glob helper + --cookies CLI. render_markdown embeds images with  and videos with an HTML <video controls> tag (were plain links). extract_corpus uses the package download_media (one download path). Tests rewritten for the gallery-dl API. Corpus: 4 threads, 33 media (32 img embeds + 1 video), 0 missing.
64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
"""Download a Twitter/X thread's media via gallery-dl (handles auth, 403, and video/GIF -> mp4)."""
|
|
from __future__ import annotations
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
try:
|
|
from scripts.twitter_threads.error_types import ErrorInfo, make_error, Result, ThreadData, thread_from_dict
|
|
except ImportError:
|
|
from error_types import ErrorInfo, make_error, Result, ThreadData, thread_from_dict
|
|
|
|
|
|
|
|
def download_media(thread: ThreadData, output_dir: Path, cookies_path: Path | None = None) -> 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)))
|
|
args = ["gallery-dl", "-o", "text-tweets=true", "-o", "conversations=true", "-D", str(output_dir)]
|
|
if cookies_path is not None:
|
|
args += ["--cookies", str(cookies_path)]
|
|
args.append(thread.source_url)
|
|
try:
|
|
completed = subprocess.run(args, capture_output=True, text=True)
|
|
except OSError as e:
|
|
return Result.err(make_error("GalleryDlError", "download_media", str(e)))
|
|
if completed.returncode != 0:
|
|
return Result.err(make_error("GalleryDlError", "download_media", completed.stderr[:500]))
|
|
post_ids = {p.post_id for p in thread.posts}
|
|
files = [f for f in output_dir.rglob("*") if f.is_file() and f.name.split("_")[0] in post_ids]
|
|
return Result.ok(sorted(files, key=lambda f: f.name))
|
|
|
|
def media_files_for_post(output_dir: Path, post_id: str) -> list[str]:
|
|
return sorted(f.name for f in output_dir.glob(f"{post_id}_*") if f.is_file())
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Download media for a Twitter/X thread_data.json via gallery-dl.")
|
|
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")
|
|
parser.add_argument("--cookies", type=Path, default=None, help="cookies.txt for gallery-dl auth")
|
|
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, args.cookies)
|
|
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())
|