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