pdfcrawler.py
https://medium.com/@jensbeckerdev/learn-cybersecurity-like-a-pro-in-2026-2d42d82e98f0Anonymous
python
04/15/2026 1:51 PM
20.8 KB
8
No Index
import io
import json
import random
import re
import time
from pathlib import Path
from urllib.parse import parse_qs, quote_plus, urlparse
import requests
from bs4 import BeautifulSoup
from pypdf import PdfReader
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from pyfiglet import Figlet
from colorama import init as colorama_init
from termcolor import colored
pdf_search = [
"python",
"bash scripting",
"powershell scripting",
"go programming",
"rust programming",
"c programming",
"c++ programming",
"malware development basics",
"windows server administration",
"linux server administration",
"linux command line",
"windows command line",
"cyber security",
"ethical hacking",
"python malware",
"red teaming",
"python for beginners",
"start programming",
"windows exploitation",
"linux exploitation",
"oscp preparation",
"windows malware development",
"rootkit development",
"windows kernel exploitation",
"linux kernel exploitation",
"windows driver development",
"linux driver development",
"windows reverse engineering",
"linux reverse engineering",
]
# Optional backward-compatible variable name as independent copy
# (same topics, but no shared list reference).
pdf_research = list(pdf_search)
BASE_EXPORT_DIR = Path("exports")
MAX_RESULTS_PER_QUERY = 60
MAX_DOWNLOADS_PER_QUERY = 12
MIN_PDF_SIZE_BYTES = 200_000
MAX_PDF_SIZE_BYTES = 80_000_000
MIN_PDF_PAGES = 20
MIN_DELAY_SECONDS = 0.8
MAX_DELAY_SECONDS = 1.8
USER_AGENT = (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)
POSITIVE_URL_HINTS = {"ebook", "book", "handbook", "guide", "manual", "tutorial", "reference"}
NEGATIVE_URL_HINTS = {"slides", "cheatsheet", "exam", "dump", "answers", "worksheet", "brochure", "flyer"}
def banner():
colorama_init()
f = Figlet(font="slant")
print(colored(f.renderText("PDF Crawler"), "cyan"))
def create_http_session() -> requests.Session:
session = requests.Session()
retry = Retry(
total=3,
connect=3,
read=3,
backoff_factor=0.6,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "HEAD"],
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("https://", adapter)
session.mount("http://", adapter)
session.headers.update({"User-Agent": USER_AGENT})
return session
def slugify(value: str) -> str:
value = value.strip().lower()
value = re.sub(r"[^a-z0-9]+", "-", value)
return value.strip("-") or "query"
def sanitize_filename(name: str) -> str:
cleaned = re.sub(r"[\\/:*?\"<>|]+", "_", name)
cleaned = re.sub(r"\s+", " ", cleaned).strip()
return cleaned[:140] if cleaned else "document"
def extract_google_target(href: str) -> str | None:
if href.startswith("/url?"):
parsed = urlparse(href)
target = parse_qs(parsed.query).get("q", [None])[0]
return target
if href.startswith("http://") or href.startswith("https://"):
return href
return None
def extract_bing_target(href: str) -> str | None:
if href.startswith("http://") or href.startswith("https://"):
return href
return None
def is_probable_pdf_url(url: str) -> bool:
parsed = urlparse(url)
path_lower = parsed.path.lower()
query_lower = parsed.query.lower()
return path_lower.endswith(".pdf") or ".pdf" in path_lower or "pdf" in query_lower
def build_search_query(topic: str) -> str:
return f'"{topic}" (ebook OR handbook OR guide OR manual OR tutorial) filetype:pdf'
def build_relaxed_query(topic: str) -> str:
return f'"{topic}" (reference OR notes OR documentation) filetype:pdf'
def request_with_pacing(session: requests.Session, url: str, timeout: int = 20) -> requests.Response:
time.sleep(random.uniform(MIN_DELAY_SECONDS, MAX_DELAY_SECONDS))
response = session.get(url, timeout=timeout)
response.raise_for_status()
return response
def google_search(topic: str, session: requests.Session, max_results: int = MAX_RESULTS_PER_QUERY) -> list[str]:
search_query = build_search_query(topic)
url = f"https://www.google.com/search?q={quote_plus(search_query)}&num={max_results}&hl=en"
response = request_with_pacing(session, url, timeout=20)
if "https://www.google.com/sorry/" in response.url or "/sorry/index" in response.url:
raise requests.HTTPError("Google rate limit/captcha detected")
soup = BeautifulSoup(response.text, "html.parser")
links: list[str] = []
seen: set[str] = set()
for a in soup.find_all("a", href=True):
target = extract_google_target(a["href"])
if not target:
continue
if not is_probable_pdf_url(target):
continue
normalized = target.split("#", 1)[0]
if normalized not in seen:
seen.add(normalized)
links.append(normalized)
return links
def bing_search(topic: str, session: requests.Session, max_results: int = MAX_RESULTS_PER_QUERY) -> list[str]:
query = build_search_query(topic)
url = f"https://www.bing.com/search?q={quote_plus(query)}&count={max_results}"
response = request_with_pacing(session, url, timeout=20)
soup = BeautifulSoup(response.text, "html.parser")
links: list[str] = []
seen: set[str] = set()
for a in soup.select("li.b_algo h2 a[href], a[href]"):
target = extract_bing_target(a.get("href", ""))
if not target or not is_probable_pdf_url(target):
continue
normalized = target.split("#", 1)[0]
if normalized not in seen:
seen.add(normalized)
links.append(normalized)
return links
def duckduckgo_search(topic: str, session: requests.Session, max_results: int = MAX_RESULTS_PER_QUERY) -> list[str]:
query = build_relaxed_query(topic)
url = f"https://duckduckgo.com/html/?q={quote_plus(query)}"
response = request_with_pacing(session, url, timeout=20)
soup = BeautifulSoup(response.text, "html.parser")
links: list[str] = []
seen: set[str] = set()
for a in soup.select("a.result__a[href], a[href]"):
href = a.get("href", "")
if not href:
continue
parsed = urlparse(href)
if "duckduckgo.com" in parsed.netloc and parsed.path.startswith("/l/"):
real_target = parse_qs(parsed.query).get("uddg", [None])[0]
target = real_target or href
else:
target = href
if not target or not is_probable_pdf_url(target):
continue
normalized = target.split("#", 1)[0]
if normalized not in seen:
seen.add(normalized)
links.append(normalized)
if len(links) >= max_results:
break
return links
def archive_org_search(topic: str, session: requests.Session, max_results: int = 15) -> list[str]:
# Internet Archive advanced search endpoint
query = f"({topic}) AND mediatype:texts"
search_url = (
"https://archive.org/advancedsearch.php?"
f"q={quote_plus(query)}&fl[]=identifier&sort[]=downloads+desc&rows={max_results}&page=1&output=json"
)
response = request_with_pacing(session, search_url, timeout=25)
payload = response.json()
identifiers = [
doc.get("identifier")
for doc in payload.get("response", {}).get("docs", [])
if doc.get("identifier")
]
links: list[str] = []
for identifier in identifiers:
# Most common downloadable filename pattern on archive.org
links.append(f"https://archive.org/download/{identifier}/{identifier}.pdf")
return links
def collect_candidate_links(topic: str, session: requests.Session, max_results: int = MAX_RESULTS_PER_QUERY) -> list[str]:
engines = [
("google", google_search),
("bing", bing_search),
("duckduckgo", duckduckgo_search),
("archive", archive_org_search),
]
combined: list[str] = []
seen: set[str] = set()
for engine_name, engine_func in engines:
try:
urls = engine_func(topic, session, max_results=max_results)
if urls:
print(f" [+] {engine_name}: {len(urls)} Kandidaten")
else:
print(f" [.] {engine_name}: 0 Kandidaten")
except Exception as exc:
print(f" [-] {engine_name} failed: {exc}")
continue
for url in urls:
normalized = url.split("#", 1)[0]
if normalized not in seen:
seen.add(normalized)
combined.append(normalized)
combined.sort(key=lambda u: base_quality_score(u, topic), reverse=True)
return combined[: max_results * 3]
def base_quality_score(url: str, topic: str) -> int:
url_lower = url.lower()
topic_tokens = [t for t in re.split(r"\W+", topic.lower()) if len(t) > 2]
score = 0
if any(token in url_lower for token in topic_tokens):
score += 2
if any(keyword in url_lower for keyword in POSITIVE_URL_HINTS):
score += 1
if any(keyword in url_lower for keyword in NEGATIVE_URL_HINTS):
score -= 2
return score
def analyze_pdf(content: bytes) -> tuple[int, str]:
try:
reader = PdfReader(io.BytesIO(content), strict=False)
pages = len(reader.pages)
title = ""
if reader.metadata and getattr(reader.metadata, "title", None):
title = str(reader.metadata.title).strip()
return pages, title
except Exception:
return 0, ""
def evaluate_pdf_quality(url: str, content: bytes, topic: str, min_pages: int = MIN_PDF_PAGES, min_size: int = MIN_PDF_SIZE_BYTES, min_score: int = 1) -> tuple[bool, dict]:
info = {
"url": url,
"size_bytes": len(content),
"score": base_quality_score(url, topic),
"reason": "",
"pages": 0,
"title": "",
}
if not content.startswith(b"%PDF"):
info["reason"] = "Datei beginnt nicht mit PDF-Header"
return False, info
if info["size_bytes"] < min_size:
info["reason"] = "PDF zu klein (wahrscheinlich kein Buch)"
return False, info
if info["size_bytes"] > MAX_PDF_SIZE_BYTES:
info["reason"] = "PDF zu groß"
return False, info
pages, title = analyze_pdf(content)
info["pages"] = pages
info["title"] = title
if pages < min_pages:
info["reason"] = "Zu wenige Seiten für E-Book-Qualität"
return False, info
title_lower = title.lower()
topic_tokens = [t for t in re.split(r"\W+", topic.lower()) if len(t) > 2]
if title_lower and any(token in title_lower for token in topic_tokens):
info["score"] += 2
if any(keyword in title_lower for keyword in NEGATIVE_URL_HINTS):
info["score"] -= 2
if info["score"] < min_score:
info["reason"] = "Niedriger Relevanzscore"
return False, info
info["reason"] = "ok"
return True, info
def download_pdfs(
links: list[str],
topic: str,
session: requests.Session,
max_downloads: int = MAX_DOWNLOADS_PER_QUERY,
min_pages: int = MIN_PDF_PAGES,
min_size: int = MIN_PDF_SIZE_BYTES,
min_score: int = 1,
) -> int:
topic_slug = slugify(topic)
topic_dir = BASE_EXPORT_DIR / topic_slug
topic_dir.mkdir(parents=True, exist_ok=True)
manifest_path = topic_dir / "manifest.jsonl"
downloaded = 0
seen_urls: set[str] = set()
with manifest_path.open("a", encoding="utf-8") as manifest:
for url in links:
if downloaded >= max_downloads:
break
if url in seen_urls:
continue
seen_urls.add(url)
row = {"query": topic, "url": url, "status": "skipped", "message": ""}
try:
response = request_with_pacing(session, url, timeout=30)
content_type = (response.headers.get("Content-Type") or "").lower()
if "pdf" not in content_type and ".pdf" not in url.lower():
row["message"] = "Kein PDF-Content-Type"
manifest.write(json.dumps(row, ensure_ascii=False) + "\n")
continue
content = response.content
is_high_quality, details = evaluate_pdf_quality(
url,
content,
topic,
min_pages=min_pages,
min_size=min_size,
min_score=min_score,
)
row.update(details)
if not is_high_quality:
row["message"] = details.get("reason", "Qualitätsfilter nicht bestanden")
manifest.write(json.dumps(row, ensure_ascii=False) + "\n")
continue
parsed = urlparse(url)
url_filename = Path(parsed.path).name or "document.pdf"
if not url_filename.lower().endswith(".pdf"):
url_filename += ".pdf"
title_candidate = details.get("title") or Path(url_filename).stem
safe_name = sanitize_filename(title_candidate)
out_path = topic_dir / f"{safe_name}.pdf"
if out_path.exists():
row["status"] = "exists"
row["file"] = str(out_path)
row["message"] = "Bereits vorhanden"
manifest.write(json.dumps(row, ensure_ascii=False) + "\n")
continue
out_path.write_bytes(content)
downloaded += 1
row["status"] = "downloaded"
row["file"] = str(out_path)
row["message"] = "OK"
manifest.write(json.dumps(row, ensure_ascii=False) + "\n")
print(f"[+] Downloaded ({downloaded}/{max_downloads}): {out_path}")
except Exception as exc:
row["status"] = "error"
row["message"] = str(exc)
manifest.write(json.dumps(row, ensure_ascii=False) + "\n")
print(f"[-] Failed: {url} -> {exc}")
return downloaded
def process_topic(topic: str, session: requests.Session) -> int:
links = collect_candidate_links(topic, session)
print(f"[+] Found {len(links)} candidate PDF links for: {topic}")
if not links:
return 0
strict_downloaded = download_pdfs(
links,
topic,
session,
max_downloads=MAX_DOWNLOADS_PER_QUERY,
min_pages=MIN_PDF_PAGES,
min_size=MIN_PDF_SIZE_BYTES,
min_score=1,
)
if strict_downloaded > 0:
return strict_downloaded
print(" [.] Keine Treffer im strengen Modus, starte relaxed fallback...")
relaxed_downloaded = download_pdfs(
links,
topic,
session,
max_downloads=max(4, MAX_DOWNLOADS_PER_QUERY // 2),
min_pages=8,
min_size=80_000,
min_score=0,
)
return relaxed_downloaded
if __name__ == "__main__":
banner()
print("[+] Starting PDF Crawler with quality filters...")
BASE_EXPORT_DIR.mkdir(parents=True, exist_ok=True)
http = create_http_session()
summary: dict[str, int] = {}
for topic in pdf_search:
print(f"\n[...] Searching for: {topic}")
try:
downloaded_count = process_topic(topic, http)
summary[topic] = downloaded_count
if downloaded_count == 0:
print(f"[!] Keine PDF gespeichert für Topic: {topic}")
else:
print(f"[+] Saved {downloaded_count} PDF(s) for: {topic}")
except Exception as exc:
print(f"[-] Search failed for '{topic}': {exc}")
print("\n[+] Run summary")
for topic, count in summary.items():
print(f" - {topic}: {count} PDF(s)")
Editor is loading...