scraper/pdf_scraper.py

import urllib3
import requests
from pathlib import Path
from urllib.parse import urljoin, urlparse, unquote
from collections import deque

from bs4 import BeautifulSoup

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}

_SKIP_EXTENSIONS = {
    ".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".ico",
    ".mp4", ".mp3", ".wav", ".zip", ".css", ".js",
}


def _normalise_url(url: str) -> str:
    parsed = urlparse(url)
    return parsed._replace(scheme="https", fragment="").geturl().rstrip("/")


def _is_pdf_url(url: str) -> bool:
    return Path(urlparse(url).path.lower()).suffix == ".pdf"


def _is_html_url(url: str) -> bool:
    ext = Path(urlparse(url).path.lower()).suffix
    return ext not in _SKIP_EXTENSIONS and ext != ".pdf"


def _safe_filename(url: str) -> str:
    name = unquote(Path(urlparse(url).path).name)
    if not name.lower().endswith(".pdf"):
        name += ".pdf"
    # replace characters that are invalid in Windows filenames
    for ch in r'\/:*?"<>|':
        name = name.replace(ch, "_")
    return name


def _download_pdf(url: str, save_dir: Path) -> str | None:
    filename = _safe_filename(url)
    filepath = save_dir / filename

    if filepath.exists():
        print(f"  [EXISTS]      {filename}")
        return str(filepath)

    try:
        response = requests.get(url, headers=_HEADERS, verify=False, timeout=30, stream=True)
        response.raise_for_status()

        content_type = response.headers.get("content-type", "")
        if "pdf" not in content_type and not url.lower().endswith(".pdf"):
            return None

        with open(filepath, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)

        size_kb = filepath.stat().st_size // 1024
        print(f"  [DOWNLOADED]  {filename} ({size_kb} KB)")
        return str(filepath)

    except Exception as e:
        print(f"  [FAILED]      {url} -> {e}")
        return None


def find_pdfs(base_url: str, max_pages: int = 1000) -> list:
    """
    Crawl base_url and all its internal sub-links.
    Only discovers PDF URLs — does not download anything.
    Returns a list of PDF URLs found.
    """
    domain = urlparse(base_url).netloc
    visited_pages = set()
    found_pdfs = set()
    queue = deque([_normalise_url(base_url)])

    print(f"Scanning for PDFs: {base_url}")
    print(f"Max pages to scan: {max_pages}")
    print()

    while queue:
        url = queue.popleft()

        if _is_pdf_url(url):
            if url not in found_pdfs:
                found_pdfs.add(url)
                print(f"  [PDF FOUND] {url}")
            continue

        if not _is_html_url(url) or url in visited_pages:
            continue

        if len(visited_pages) >= max_pages:
            break

        visited_pages.add(url)

        try:
            response = requests.get(url, headers=_HEADERS, verify=False, timeout=15)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, "html.parser")

            for tag in soup.find_all("a", href=True):
                absolute = urljoin(url, tag["href"])
                parsed = urlparse(absolute)

                if parsed.scheme not in ("http", "https"):
                    continue

                clean = _normalise_url(absolute)

                if _is_pdf_url(clean):
                    if clean not in found_pdfs:
                        queue.append(clean)
                elif parsed.netloc == domain and clean not in visited_pages:
                    queue.append(clean)

            print(f"[{len(visited_pages):>4}] Scanned: {url}")

        except Exception as e:
            print(f"[SKIP] {url} -> {e}")
            continue

    print(f"\nScan complete.")
    print(f"  Pages scanned : {len(visited_pages)}")
    print(f"  PDFs found    : {len(found_pdfs)}")
    return list(found_pdfs)


def download_pdfs(pdf_urls: list, save_dir: str = "data/raw") -> list:
    """
    Downloads a list of PDF URLs into save_dir.
    Returns list of downloaded file paths.
    """
    save_path = Path(save_dir)
    save_path.mkdir(parents=True, exist_ok=True)
    downloaded = []

    print(f"\nDownloading {len(pdf_urls)} PDFs to '{save_dir}'...")
    for url in pdf_urls:
        path = _download_pdf(url, save_path)
        if path:
            downloaded.append(path)

    print(f"\nDone. {len(downloaded)} PDFs downloaded.")
    return downloaded


def scrape_pdfs(base_url: str, save_dir: str = "data/raw", max_pages: int = 1000) -> list:
    """
    Crawl base_url and all its internal sub-links.
    Find and download every PDF discovered — including those hosted on external CDNs.
    Returns a list of file paths for all downloaded PDFs.
    """
    save_path = Path(save_dir)
    save_path.mkdir(parents=True, exist_ok=True)

    domain = urlparse(base_url).netloc
    visited_pages = set()
    found_pdfs = set()
    downloaded = []
    queue = deque([_normalise_url(base_url)])

    print(f"Starting PDF scrape: {base_url}")
    print(f"Saving to: {save_dir}")
    print(f"Max pages: {max_pages}")
    print()

    while queue:
        url = queue.popleft()

        # handle PDF links directly
        if _is_pdf_url(url):
            if url not in found_pdfs:
                found_pdfs.add(url)
                path = _download_pdf(url, save_path)
                if path:
                    downloaded.append(path)
            continue

        # skip non-HTML and already visited
        if not _is_html_url(url) or url in visited_pages:
            continue

        if len(visited_pages) >= max_pages:
            break

        visited_pages.add(url)

        try:
            response = requests.get(url, headers=_HEADERS, verify=False, timeout=15)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, "html.parser")

            for tag in soup.find_all("a", href=True):
                absolute = urljoin(url, tag["href"])
                parsed = urlparse(absolute)

                if parsed.scheme not in ("http", "https"):
                    continue

                clean = _normalise_url(absolute)

                if _is_pdf_url(clean):
                    # PDFs can be on any domain (CDN, S3, etc.) — always follow
                    if clean not in found_pdfs:
                        queue.append(clean)
                elif parsed.netloc == domain and clean not in visited_pages:
                    # only follow HTML links that stay on the same domain
                    queue.append(clean)

            print(f"[{len(visited_pages):>4}] Scanned: {url}")

        except Exception as e:
            print(f"[SKIP] {url} -> {e}")
            continue

    print(f"\nDone.")
    print(f"  Pages scanned : {len(visited_pages)}")
    print(f"  PDFs found    : {len(found_pdfs)}")
    print(f"  PDFs downloaded: {len(downloaded)}")
    return downloaded