"""Download free-access PDFs (textbooks, course notes) into data/books/. Sources are defined in researchpath/corpus.py :: CANONICAL_TEXT_SOURCES. Only downloads PDFs (source.filename ends with .pdf). Skips any file that already exists — safe to re-run. Usage: uv run python scripts/fetch_pdfs.py """ from __future__ import annotations import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import requests from rich.console import Console from rich.table import Table from researchpath.corpus import CANONICAL_TEXT_SOURCES console = Console() ROOT = Path(__file__).resolve().parents[1] BOOKS_DIR = ROOT / "data" / "books" BOOKS_DIR.mkdir(parents=True, exist_ok=True) HEADERS = {"User-Agent": "ResearchPath/1.0 (educational corpus builder)"} TIMEOUT = 120 # seconds def _download(url: str, dest: Path) -> tuple[bool, str]: """Return (success, message).""" if dest.exists(): return True, f"already exists ({dest.stat().st_size // 1024} KB)" try: resp = requests.get(url, headers=HEADERS, timeout=TIMEOUT, stream=True) resp.raise_for_status() total = 0 with open(dest, "wb") as f: for chunk in resp.iter_content(chunk_size=65536): f.write(chunk) total += len(chunk) return True, f"downloaded {total // 1024} KB" except requests.RequestException as e: return False, str(e) def main() -> int: pdf_sources = [s for s in CANONICAL_TEXT_SOURCES if s.filename.endswith(".pdf")] table = Table(title="PDF download results") table.add_column("Source ID", style="bold") table.add_column("Status") table.add_column("Detail") ok = 0 for source in pdf_sources: dest = BOOKS_DIR / source.filename success, msg = _download(source.url, dest) status = "[green]OK[/green]" if success else "[red]FAIL[/red]" table.add_row(source.source_id, status, msg) if success: ok += 1 time.sleep(1) # polite crawl delay console.print(table) console.print(f"\n[bold]{'[green]' if ok == len(pdf_sources) else '[yellow]'}{ok}/{len(pdf_sources)} PDFs ready in data/books/[/bold]") return 0 if ok > 0 else 1 if __name__ == "__main__": sys.exit(main())