diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..e6fc6c0 --- /dev/null +++ b/.env.example @@ -0,0 +1,27 @@ +# ============================================================ +# JonasSearch Crawler Gamma 0.1 +# ============================================================ + +# PostgreSQL +JONASSEARCH_DB_HOST=127.0.0.1 +JONASSEARCH_DB_PORT=25432 +JONASSEARCH_DB_NAME=jonassearch +JONASSEARCH_DB_USER=jonassearch +JONASSEARCH_DB_PASSWORD=MeinStarkesPasswort123! + +# Crawler +JONASSEARCH_START_URL=https://reddit.com +JONASSEARCH_WORKERS=4 + +# Mindestabstand zwischen Requests zur gleichen Domain +JONASSEARCH_DELAY=3.0 + +# robots.txt Cache +JONASSEARCH_ROBOTS_CACHE_SECONDS=86400 + +# HTTP +JONASSEARCH_TIMEOUT=20 +JONASSEARCH_MAX_CONTENT_LENGTH=5242880 + +# User-Agent +JONASSEARCH_USER_AGENT=JonasSearchBot/0.1 (+https://jonassearch.local/bot) diff --git a/crawler.py b/crawler.py new file mode 100644 index 0000000..02b3563 --- /dev/null +++ b/crawler.py @@ -0,0 +1,859 @@ +#!/usr/bin/env python3 +""" +JonasSearch Crawler Gamma 0.1 + +PostgreSQL-based parallel crawler. + +Features: +- PostgreSQL queue with row-level locking (FOR UPDATE SKIP LOCKED) +- Multiple crawler workers can safely share one queue +- Stores title, headings, cleaned text and links +- Search vector is maintained by the PostgreSQL trigger +- URL deduplication via UNIQUE(url) +- Priority queue +- Per-host crawl delay +- robots.txt support with in-memory cache +- Graceful shutdown +- Basic crawler statistics + +Expected PostgreSQL schema: + pages + crawl_queue + crawler_stats + +Connection defaults can be overridden with environment variables: + JONASSEARCH_DB_HOST + JONASSEARCH_DB_PORT + JONASSEARCH_DB_NAME + JONASSEARCH_DB_USER + JONASSEARCH_DB_PASSWORD + +Example: + python3 crawler.py +""" + +import os +import re +import signal +import time +import logging +import multiprocessing as mp +from collections import defaultdict +from html import unescape +from urllib.parse import urljoin, urlparse, urldefrag +from urllib.robotparser import RobotFileParser +from dotenv import load_dotenv + +import psycopg2 +from psycopg2.extras import execute_values +from urllib.request import Request, urlopen +from urllib.error import HTTPError, URLError + +try: + from bs4 import BeautifulSoup +except ImportError: + BeautifulSoup = None + +load_dotenv() + + +# ============================================================ +# CONFIG +# ============================================================ + +DB_HOST = os.getenv("JONASSEARCH_DB_HOST", "127.0.0.1") +DB_PORT = int(os.getenv("JONASSEARCH_DB_PORT", "25432")) +DB_NAME = os.getenv("JONASSEARCH_DB_NAME", "jonassearch") +DB_USER = os.getenv("JONASSEARCH_DB_USER", "jonassearch") +DB_PASSWORD = os.getenv("JONASSEARCH_DB_PASSWORD", "") + +START_URL = os.getenv("JONASSEARCH_START_URL", "https://github.com/") + +WORKERS = int(os.getenv("JONASSEARCH_WORKERS", "3")) + +# Minimum delay between requests to the same host, per worker. +# robots.txt can specify a larger crawl delay. +DEFAULT_DELAY = float(os.getenv("JONASSEARCH_DELAY", "3.0")) + +ROBOTS_CACHE_SECONDS = int( + os.getenv("JONASSEARCH_ROBOTS_CACHE_SECONDS", str(24 * 60 * 60)) +) + +REQUEST_TIMEOUT = float(os.getenv("JONASSEARCH_TIMEOUT", "20")) + +MAX_CONTENT_LENGTH = int( + os.getenv("JONASSEARCH_MAX_CONTENT_LENGTH", str(5 * 1024 * 1024)) +) + +USER_AGENT = os.getenv( + "JONASSEARCH_USER_AGENT", + "JonasSearchBot/0.1 (+https://jonassearch.local/bot)" +) + +# Only crawl normal HTTP(S) pages. +ALLOWED_SCHEMES = {"http", "https"} + +# Ignore obvious non-HTML resources. +BLOCKED_EXTENSIONS = { + ".7z", ".avi", ".bin", ".bmp", ".bz2", ".class", ".csv", ".doc", + ".docx", ".gif", ".gz", ".ico", ".iso", ".jpeg", ".jpg", ".m4a", + ".mkv", ".mov", ".mp3", ".mp4", ".mpeg", ".pdf", ".png", ".ppt", + ".pptx", ".rar", ".svg", ".tar", ".tgz", ".tif", ".tiff", ".webm", + ".webp", ".xls", ".xlsx", ".xml", ".zip" +} + +# Do not follow these URL schemes. +BLOCKED_PREFIXES = ( + "mailto:", + "javascript:", + "tel:", + "data:", + "ftp:", + "file:", +) + +# Keep generated pages reasonably useful for search. +MAX_TITLE_LENGTH = 1000 +MAX_HEADINGS = 200 +MAX_HEADING_LENGTH = 1000 +MAX_TEXT_LENGTH = 2_000_000 +MAX_LINKS = 10_000 + + +# ============================================================ +# LOGGING / SHUTDOWN +# ============================================================ + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(processName)s] %(levelname)s %(message)s", + datefmt="%H:%M:%S", +) + +STOP = mp.Event() + + +def handle_signal(signum, _frame): + logging.info("Shutdown signal received.") + STOP.set() + + +signal.signal(signal.SIGINT, handle_signal) +signal.signal(signal.SIGTERM, handle_signal) + + +# ============================================================ +# DATABASE +# ============================================================ + +def db_connect(): + return psycopg2.connect( + host=DB_HOST, + port=DB_PORT, + dbname=DB_NAME, + user=DB_USER, + password=DB_PASSWORD, + connect_timeout=10, + application_name="jonassearch-crawler", + ) + + +def ensure_stats_table(conn): + with conn.cursor() as cur: + cur.execute(""" + CREATE TABLE IF NOT EXISTS crawler_stats ( + id BIGSERIAL PRIMARY KEY, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(), + crawled INTEGER NOT NULL DEFAULT 0, + queued INTEGER NOT NULL DEFAULT 0, + failed INTEGER NOT NULL DEFAULT 0 + ) + """) + conn.commit() + + +def write_stats(conn, crawled, queued, failed): + with conn.cursor() as cur: + cur.execute(""" + INSERT INTO crawler_stats (crawled, queued, failed) + VALUES (%s, %s, %s) + """, (crawled, queued, failed)) + conn.commit() + + +def get_next_queue_item(conn): + """ + Atomically claim one pending URL. + + SKIP LOCKED is the important part: several crawler processes can + consume the same PostgreSQL queue without waiting for each other. + """ + with conn.cursor() as cur: + cur.execute(""" + WITH next_item AS ( + SELECT id + FROM crawl_queue + WHERE status = 'pending' + ORDER BY priority DESC, discovered_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + UPDATE crawl_queue q + SET status = 'processing', + attempts = attempts + 1, + last_attempt_at = now() + FROM next_item + WHERE q.id = next_item.id + RETURNING q.id, q.url, q.source_url + """) + row = cur.fetchone() + conn.commit() + return row + + +def finish_queue_item(conn, queue_id, success=True): + with conn.cursor() as cur: + cur.execute(""" + UPDATE crawl_queue + SET status = %s + WHERE id = %s + """, ("done" if success else "failed", queue_id)) + conn.commit() + + +def requeue_stale_items(conn, older_than_seconds=3600): + """ + Recover URLs left in 'processing' after a crash. + """ + with conn.cursor() as cur: + cur.execute(""" + UPDATE crawl_queue + SET status = 'pending' + WHERE status = 'processing' + AND last_attempt_at < now() - (%s * interval '1 second') + """, (older_than_seconds,)) + changed = cur.rowcount + conn.commit() + if changed: + logging.info("Requeued %d stale processing URLs.", changed) + + +def queue_urls(conn, links, source_url): + if not links: + return 0 + + # Link URLs also come from the web and can carry NUL bytes. + safe_links = clean_text_list(links) + safe_source = clean_text(source_url) + + rows = [(url, safe_source, 0, "pending") for url in safe_links] + + with conn.cursor() as cur: + execute_values( + cur, + """ + INSERT INTO crawl_queue (url, source_url, priority, status) + VALUES %s + ON CONFLICT (url) DO NOTHING + """, + rows, + template="(%s, %s, %s, %s)", + page_size=500, + ) + inserted = cur.rowcount + + conn.commit() + return inserted + + +# ============================================================ +# TEXT SANITIZATION +# ============================================================ + +def clean_text(value): + """ + Remove NUL characters from a string. + + Web content can contain a literal "\x00" anywhere in the extracted + text. PostgreSQL rejects NUL bytes inside ordinary text columns, so a + single dirty string would abort the INSERT and crash the worker. + Sanitizing centrally before every DB write keeps the crawl running. + """ + if isinstance(value, str): + return value.replace("\x00", "") + return value + + +def clean_text_list(values): + """Sanitize every string of a list (headings, links, ...).""" + return [clean_text(v) for v in values] + + +# ============================================================ +# URL NORMALIZATION +# ============================================================ + +def normalize_url(url, base_url=None): + if not url: + return None + + url = unescape(url).strip() + + if base_url: + url = urljoin(base_url, url) + + if any(url.lower().startswith(prefix) for prefix in BLOCKED_PREFIXES): + return None + + url, _fragment = urldefrag(url) + + parsed = urlparse(url) + + if parsed.scheme.lower() not in ALLOWED_SCHEMES: + return None + + if not parsed.netloc: + return None + + # Lowercase scheme and hostname. + scheme = parsed.scheme.lower() + hostname = (parsed.hostname or "").lower() + + if not hostname: + return None + + # Drop default ports. + port = parsed.port + if (scheme == "http" and port == 80) or (scheme == "https" and port == 443): + port = None + + if port: + netloc = f"{hostname}:{port}" + else: + netloc = hostname + + # Preserve path/query, but remove empty query marker. + path = parsed.path or "/" + query = parsed.query + + normalized = f"{scheme}://{netloc}{path}" + if query: + normalized += "?" + query + + # Avoid obvious files. + path_lower = path.lower() + for ext in BLOCKED_EXTENSIONS: + if path_lower.endswith(ext): + return None + + return normalized + + +# ============================================================ +# ROBOTS +# ============================================================ + +class RobotsCache: + def __init__(self): + self.cache = {} + self.last_checked = {} + + def allowed(self, url): + parsed = urlparse(url) + host = parsed.netloc.lower() + + now = time.monotonic() + + if ( + host not in self.cache + or now - self.last_checked.get(host, 0) > ROBOTS_CACHE_SECONDS + ): + self._load(host, parsed.scheme) + + rp = self.cache[host] + try: + return rp.can_fetch(USER_AGENT, url) + except Exception: + return True + + def delay(self, url): + parsed = urlparse(url) + host = parsed.netloc.lower() + + rp = self.cache.get(host) + if not rp: + return DEFAULT_DELAY + + # RobotFileParser does not expose Crawl-delay consistently across + # all Python versions, so use it when available. + try: + delay = rp.crawl_delay(USER_AGENT) + if delay is None: + delay = rp.crawl_delay("*") + if delay is not None: + return max(DEFAULT_DELAY, float(delay)) + except Exception: + pass + + return DEFAULT_DELAY + + def _load(self, host, scheme): + robots_url = f"{scheme}://{host}/robots.txt" + + rp = RobotFileParser() + rp.set_url(robots_url) + + try: + req = Request( + robots_url, + headers={"User-Agent": USER_AGENT} + ) + with urlopen(req, timeout=REQUEST_TIMEOUT) as response: + data = response.read(512 * 1024).decode( + "utf-8", errors="replace" + ) + rp.parse(data.splitlines()) + except Exception: + # If robots.txt cannot be retrieved, default to allowing the URL. + # This is deliberately conservative with respect to availability. + rp.parse([]) + + self.cache[host] = rp + self.last_checked[host] = time.monotonic() + + +# ============================================================ +# HTTP +# ============================================================ + +class RateLimiter: + def __init__(self): + self.last_request = defaultdict(float) + + def wait(self, url, delay): + host = urlparse(url).netloc.lower() + now = time.monotonic() + wait_for = delay - (now - self.last_request[host]) + + if wait_for > 0: + logging.info("[RATE] %s: %.1fs", host, wait_for) + time.sleep(wait_for) + + self.last_request[host] = time.monotonic() + + +def fetch(url, limiter, robots): + parsed = urlparse(url) + + if not robots.allowed(url): + return None, None, "robots" + + delay = robots.delay(url) + limiter.wait(url, delay) + + req = Request( + url, + headers={ + "User-Agent": USER_AGENT, + "Accept": ( + "text/html,application/xhtml+xml," + "application/xml;q=0.9,*/*;q=0.1" + ), + "Accept-Language": "en,de;q=0.8", + }, + ) + + try: + with urlopen(req, timeout=REQUEST_TIMEOUT) as response: + status = getattr(response, "status", 200) + content_type = response.headers.get("Content-Type", "") + content_length = response.headers.get("Content-Length") + + if content_length: + try: + if int(content_length) > MAX_CONTENT_LENGTH: + return status, None, "too_large" + except ValueError: + pass + + # We only parse HTML/XHTML. + ctype = content_type.lower() + if ctype and not ( + "text/html" in ctype + or "application/xhtml+xml" in ctype + ): + return status, None, "not_html" + + data = response.read(MAX_CONTENT_LENGTH + 1) + + if len(data) > MAX_CONTENT_LENGTH: + return status, None, "too_large" + + charset = response.headers.get_content_charset() or "utf-8" + text = data.decode(charset, errors="replace") + + return status, text, None + + except HTTPError as e: + return e.code, None, f"http_{e.code}" + except (URLError, TimeoutError, OSError) as e: + return None, None, f"network:{e}" + except Exception as e: + return None, None, f"error:{e}" + + +# ============================================================ +# HTML PARSING +# ============================================================ + +def clean_text_bs4(soup): + for tag in soup([ + "script", "style", "noscript", "template", + "svg", "canvas", "iframe", "object", + "embed", "form" + ]): + tag.decompose() + + # Remove navigation/boilerplate-ish elements where possible. + for tag in soup.find_all([ + "nav", "footer", "header", "aside" + ]): + tag.decompose() + + text = soup.get_text(" ", strip=True) + + # Normalize whitespace. + text = re.sub(r"\s+", " ", text) + return text[:MAX_TEXT_LENGTH] + + +def parse_html(html, base_url): + if BeautifulSoup is None: + raise RuntimeError( + "BeautifulSoup fehlt. Installiere python3-bs4." + ) + + soup = BeautifulSoup(html, "html.parser") + + title = "" + if soup.title: + title = soup.title.get_text(" ", strip=True) + + title = re.sub(r"\s+", " ", title)[:MAX_TITLE_LENGTH] + + headings = [] + for tag in soup.find_all(re.compile(r"^h[1-6]$")): + value = re.sub(r"\s+", " ", tag.get_text(" ", strip=True)) + if value: + headings.append(value[:MAX_HEADING_LENGTH]) + if len(headings) >= MAX_HEADINGS: + break + + links = [] + seen = set() + + for tag in soup.find_all("a", href=True): + normalized = normalize_url(tag.get("href"), base_url) + if not normalized: + continue + + if normalized in seen: + continue + + seen.add(normalized) + links.append(normalized) + + if len(links) >= MAX_LINKS: + break + + content = clean_text_bs4(soup) + + domain = urlparse(base_url).hostname or "" + domain = domain.lower() + + # TLD is intentionally simple here; the DB can later be extended with + # Public Suffix List handling if needed. + tld = "" + if "." in domain: + tld = "." + domain.rsplit(".", 1)[-1] + + return { + "title": title, + "headings": headings, + "content": content, + "links": links, + "domain": domain, + "tld": tld, + } + + +# ============================================================ +# PAGE STORAGE +# ============================================================ + +def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms): + # Sanitize everything that came from the web before it reaches + # PostgreSQL. NUL bytes are legal in HTML but illegal in text columns. + url = clean_text(url) + source_url = clean_text(source_url) + content_type = clean_text(content_type) + + safe_parsed = { + "title": clean_text(parsed["title"]), + "headings": clean_text_list(parsed["headings"]), + "content": clean_text(parsed["content"]), + "links": clean_text_list(parsed["links"]), + "domain": clean_text(parsed["domain"]), + "tld": clean_text(parsed["tld"]), + } + + with conn.cursor() as cur: + cur.execute(""" + INSERT INTO pages ( + url, + title, + headings, + content, + links, + domain, + tld, + source_url, + http_status, + content_type, + discovered_at, + crawled_at, + crawl_time_ms + ) + VALUES ( + %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, + COALESCE( + (SELECT discovered_at + FROM crawl_queue + WHERE url = %s), + now() + ), + now(), + %s + ) + ON CONFLICT (url) DO UPDATE SET + title = EXCLUDED.title, + headings = EXCLUDED.headings, + content = EXCLUDED.content, + links = EXCLUDED.links, + domain = EXCLUDED.domain, + tld = EXCLUDED.tld, + source_url = EXCLUDED.source_url, + http_status = EXCLUDED.http_status, + content_type = EXCLUDED.content_type, + crawled_at = now(), + crawl_time_ms = EXCLUDED.crawl_time_ms + """, ( + url, + safe_parsed["title"], + safe_parsed["headings"], + safe_parsed["content"], + safe_parsed["links"], + safe_parsed["domain"], + safe_parsed["tld"], + source_url, + status, + content_type, + url, + crawl_ms, + )) + + conn.commit() + + +# ============================================================ +# WORKER +# ============================================================ + +def worker(worker_id): + process_name = f"crawler-{worker_id}" + mp.current_process().name = process_name + + conn = None + robots = RobotsCache() + limiter = RateLimiter() + + crawled = 0 + failed = 0 + last_stats = time.monotonic() + + try: + conn = db_connect() + + logging.info( + "Started. DB=%s:%s/%s", + DB_HOST, DB_PORT, DB_NAME + ) + + while not STOP.is_set(): + item = get_next_queue_item(conn) + + if not item: + logging.info("Queue empty. Worker waiting...") + time.sleep(5) + continue + + queue_id, url, source_url = item + + logging.info("=" * 70) + logging.info("[CRAWL] %s", url) + logging.info("[FROM ] %s", source_url or "-") + + started = time.monotonic() + + status, html, error = fetch(url, limiter, robots) + + if error: + logging.info("[HTTP ] %s -> %s", url, error) + finish_queue_item(conn, queue_id, success=False) + failed += 1 + continue + + logging.info("[HTTP ] %s -> HTTP %s", url, status) + + # A redirect normally arrives as the final URL in urllib. + # The requested URL remains the canonical queue identity. + if html is None: + finish_queue_item(conn, queue_id, success=False) + failed += 1 + continue + + try: + parsed = parse_html(html, url) + except Exception as exc: + logging.exception("[PARSE] %s: %s", url, exc) + finish_queue_item(conn, queue_id, success=False) + failed += 1 + continue + + crawl_ms = int((time.monotonic() - started) * 1000) + + logging.info("[TITLE] %s", parsed["title"][:160]) + logging.info("[DOMAIN] %s", parsed["domain"]) + logging.info("[HEADS ] %d", len(parsed["headings"])) + logging.info("[LINKS ] %d gefunden", len(parsed["links"])) + + save_page( + conn=conn, + url=url, + source_url=source_url, + status=status, + content_type="text/html", + parsed=parsed, + crawl_ms=crawl_ms, + ) + + inserted = queue_urls( + conn, + parsed["links"], + url + ) + + logging.info( + "[SORT ] %s -> %d Links, %d neue URLs", + url, + len(parsed["links"]), + inserted, + ) + + finish_queue_item(conn, queue_id, success=True) + + crawled += 1 + + # Periodic stats; not on every page to avoid unnecessary DB load. + if time.monotonic() - last_stats >= 30: + try: + with conn.cursor() as cur: + cur.execute( + "SELECT COUNT(*) FROM crawl_queue WHERE status='pending'" + ) + queued = cur.fetchone()[0] + + write_stats(conn, crawled, queued, failed) + last_stats = time.monotonic() + except Exception: + conn.rollback() + + except Exception: + logging.exception("Worker crashed.") + raise + finally: + if conn is not None: + conn.close() + + logging.info( + "Stopped. crawled=%d failed=%d", + crawled, failed + ) + + +# ============================================================ +# STARTUP +# ============================================================ + +def main(): + print("=" * 70) + print(" JonasSearch Crawler Gamma 0.1") + print("=" * 70) + print(f"Start: {START_URL}") + print(f"Workers: {WORKERS}") + print(f"Delay: {DEFAULT_DELAY:.1f} Sekunden") + print(f"Robots-Cache: {ROBOTS_CACHE_SECONDS // 3600} Stunden") + print(f"DB: {DB_HOST}:{DB_PORT}/{DB_NAME}") + print() + + conn = db_connect() + + ensure_stats_table(conn) + requeue_stale_items(conn) + + # Seed URL. ON CONFLICT makes restarting harmless. + queue_urls(conn, [normalize_url(START_URL)], None) + + conn.close() + + workers = [] + + for i in range(WORKERS): + p = mp.Process(target=worker, args=(i,), name=f"crawler-{i}") + p.start() + workers.append(p) + + try: + while any(p.is_alive() for p in workers): + time.sleep(1) + + # If a worker dies unexpectedly, stop the remaining workers. + for p in workers: + if p.exitcode not in (None, 0): + logging.error( + "Worker %s exited with code %s.", + p.name, + p.exitcode, + ) + STOP.set() + + except KeyboardInterrupt: + STOP.set() + + finally: + STOP.set() + + for p in workers: + p.join(timeout=10) + + for p in workers: + if p.is_alive(): + p.terminate() + p.join() + + print("Crawler stopped.") + + +if __name__ == "__main__": + mp.set_start_method("spawn", force=True) + main()