865 lines
24 KiB
Python
865 lines
24 KiB
Python
#!/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 pathlib import Path
|
||
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
|
||
|
||
# .env zuerst aus dem Arbeitsverzeichnis laden (Standard), zusätzlich aus
|
||
# dem Projektordner – so findet der Prozess die Konfiguration auch dann,
|
||
# wenn er von woanders gestartet wird (z. B. via systemd).
|
||
load_dotenv()
|
||
load_dotenv(Path(__file__).resolve().parent / ".env")
|
||
|
||
|
||
# ============================================================
|
||
# 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()
|