scripts/fetch_prelinger_pd.py

# /// script
# requires-python = ">=3.11"
# dependencies = ["requests"]
# ///
"""Mirror the explicitly-PD subset of the Prelinger Archives into a bucket.

Filter: collection:prelinger, mediatype:movies, licenseurl contains 'publicdomain'
(per-item machine-checkable rights basis — see the vault note for the licensing
rationale). For each item: best mp4 derivative (512kb preferred) -> /out/films/,
plus a metadata sidecar -> /out/meta/{id}.json written only AFTER the film copy
completes, so sidecar-exists == film-is-whole (FUSE has no rename; copy-then-marker
replaces tmp-then-rename).

Resumable: re-run skips items whose sidecar exists. Politeness: 4 concurrent
downloads, streaming, no retry hammering.
"""

import json
import shutil
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import requests

OUT = Path("/out")
FILMS, META = OUT / "films", OUT / "meta"
QUERY = "collection:prelinger AND mediatype:movies AND licenseurl:*publicdomain*"
WORKERS = 4
UA = {"User-Agent": "prelinger-pd-mirror/0.1 (davanstrien; HF dataset prep)"}


def scrape_items() -> list[dict]:
    items, cursor = [], None
    while True:
        params = {"q": QUERY, "count": 10000,
                  "fields": "identifier,licenseurl,title,date,runtime,item_size"}
        if cursor:
            params["cursor"] = cursor
        r = requests.get("https://archive.org/services/search/v1/scrape",
                         params=params, headers=UA, timeout=120).json()
        items.extend(r.get("items", []))
        cursor = r.get("cursor")
        if not cursor:
            return items


def pick_mp4(identifier: str) -> tuple[str, int] | None:
    try:
        meta = requests.get(f"https://archive.org/metadata/{identifier}",
                            headers=UA, timeout=60).json()
    except (requests.RequestException, json.JSONDecodeError):
        return None
    mp4s = [(f["name"], int(f.get("size", 0) or 0))
            for f in meta.get("files", []) if f["name"].endswith(".mp4")]
    if not mp4s:
        return None
    mp4s.sort(key=lambda x: ("512kb" not in x[0], x[0]))
    return mp4s[0]


def fetch(item: dict) -> str:
    ident = item["identifier"]
    sidecar = META / f"{ident}.json"
    if sidecar.exists():
        return "skip"
    picked = pick_mp4(ident)
    if picked is None:
        return "no-mp4"
    name, size = picked
    local = Path("/tmp") / f"{ident}.mp4"
    try:
        with requests.get(f"https://archive.org/download/{ident}/{name}",
                          headers=UA, stream=True, timeout=1200) as r:
            r.raise_for_status()
            with open(local, "wb") as f:
                for chunk in r.iter_content(1 << 20):
                    f.write(chunk)
        dest = FILMS / f"{ident}.mp4"
        shutil.copyfile(local, dest)
        sidecar.write_text(json.dumps({
            "identifier": ident,
            "file": name,
            "bytes": local.stat().st_size,
            "licenseurl": item.get("licenseurl"),
            "title": item.get("title"),
            "date": item.get("date"),
            "runtime": item.get("runtime"),
            "source": f"https://archive.org/details/{ident}",
            "fetched": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }))
        return "ok"
    except requests.RequestException as e:
        print(f"[err] {ident}: {type(e).__name__}", flush=True)
        return "err"
    finally:
        local.unlink(missing_ok=True)


def main():
    FILMS.mkdir(parents=True, exist_ok=True)
    META.mkdir(parents=True, exist_ok=True)
    items = scrape_items()
    print(f"[plan] {len(items)} PD items to mirror", flush=True)
    counts = {"ok": 0, "skip": 0, "err": 0, "no-mp4": 0}
    t0 = time.time()
    with ThreadPoolExecutor(WORKERS) as ex:
        for n, res in enumerate(ex.map(fetch, items), 1):
            counts[res] += 1
            if n % 25 == 0 or n == len(items):
                gb = 0.0
                for f in FILMS.glob("*.mp4"):
                    try:
                        gb += f.stat().st_size / 1e9
                    except OSError:  # racing a concurrent copy on the FUSE mount
                        pass
                print(f"[progress] {n}/{len(items)} {counts} "
                      f"{gb:.1f} GB {time.time()-t0:.0f}s", flush=True)
    print(f"[MIRROR DONE] {counts}", flush=True)


if __name__ == "__main__":
    main()