File size: 2,312 Bytes
3f31583
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
"""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())