Serverjonas/search
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Crawler and env files

Browse Source
This commit is contained in:
tech2m
2026-09-02 18:32:20 +02:00
parent ca0f617f62
commit f1c7ecc14d
2 changed files with 886 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+27
View File
@@ -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)
crawler.py
+859
View File
@@ -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()