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

add usage of priority system

Browse Source
This commit is contained in:
jhkaktus
2026-09-06 22:05:47 +02:00
parent 0adf9eb42c
commit 2e1701c442
1 changed files with 618 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crawler.py
+618 -18
View File
@@ -10,7 +10,10 @@ Features:
- Stores title, headings, cleaned text and links
- Search vector is maintained by the PostgreSQL trigger
- URL deduplication via UNIQUE(url)
- Priority queue
- Deadlock-safe parallel queue inserts (deterministic ordering,
bounded retries with rollback + backoff)
- Smart priority queue: prefers never-crawled URLs, new domains and
new hosts, with a spread bonus for little-explored territory
- Per-host crawl delay
- robots.txt support with in-memory cache
- Graceful shutdown
@@ -34,6 +37,7 @@ Example:
import os
import re
import math
import signal
import time
import logging
@@ -41,7 +45,7 @@ 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.parse import parse_qsl, urljoin, urlparse, urldefrag
from urllib.robotparser import RobotFileParser
from dotenv import load_dotenv
@@ -131,6 +135,93 @@ MAX_LINKS = 10_000
# the INSERT keeps the crawl running without touching the DB schema.
MAX_QUEUE_URL_BYTES = 2000
# Rows per INSERT statement when filling the queue. Short statements
# keep the lock window on the UNIQUE index small.
QUEUE_INSERT_PAGE_SIZE = 500
# Deadlock handling for queue inserts: several independent crawler
# instances write overlapping URL batches into the same crawl_queue.
# Depending on insert order, PostgreSQL may then abort one transaction
# with DeadlockDetected (SQLSTATE 40P01). Instead of letting the worker
# crash, the insert is retried a few times with a short, growing
# backoff. The deterministic ordering in queue_urls() makes such
# deadlocks rare in the first place - without requiring any change to
# the database schema or to the existing index.
QUEUE_INSERT_MAX_ATTEMPTS = 3
QUEUE_INSERT_RETRY_BASE_DELAY = 0.5
# ============================================================
# URL PRIORITY CONFIG
# ============================================================
# The queue is consumed with ORDER BY priority DESC, but so far every
# URL was inserted with priority 0, so the sort never differentiated.
# The constants below feed calculate_url_priority() (see the "URL
# PRIORITY" section): the score prefers the unknown - never-crawled
# URLs, new domains, new hosts - and gives hosts with little presence
# in the crawl so far a spread bonus, WITHOUT ever blocking a domain.
# A heavily crawled host merely loses priority, so crawl capacity
# flows to new territory first.
# Base score every URL starts with (the old "everything equal" level).
PRIORITY_BASE = 100
# URL is not yet present in crawl_queue/pages (strongest single signal).
PRIORITY_BONUS_NEW_URL = 200
# Registrable domain has no pages and no queue entries yet.
PRIORITY_BONUS_NEW_DOMAIN = 120
# Host is unknown although the domain may already be known (subdomain
# or www variant of an existing site).
PRIORITY_BONUS_NEW_HOST = 60
# Spread bonus: log2-falloff boost for hosts with a small presence in
# the crawl, so many different domains/hosts are reached before the
# crawl digs deeper into a single ecosystem. 0 for hosts that do not
# exist in the DB yet (those get the explicit new-host bonuses above).
PRIORITY_SPREAD_BONUS_MAX = 80
PRIORITY_SPREAD_BONUS_DIVISOR = 20.0
# Source preference, integrated into the new system instead of
# replacing it: links out of little-explored territory are promising,
# links from a heavily crawled source are pulled slightly back.
PRIORITY_SOURCE_NOVELTY_BONUS = 20
PRIORITY_SOURCE_PENALTY_DIVISOR = 10.0
# Same-host link floods (big navigation lists, archives) are flattened,
# not blocked: a host with this many known URLs gets the penalty.
PRIORITY_CLUSTER_HOST_THRESHOLD = 25
PRIORITY_CLUSTER_PENALTY = 60
# URL variants of already known pages (trailing slash, tracking,
# session or pagination parameters) lose against genuinely new pages.
PRIORITY_VARIANT_PENALTY = 30
# How many known URLs are sampled for the variant check, so batches
# with many known URLs cannot turn into a runtime blow-up.
VARIANT_CHECK_SAMPLE_LIMIT = 200
# Query-string keys that mark trivial variants of the same page.
VARIANT_QUERY_KEYS = {
"session", "sessionid", "sid", "phpsessid", "jsessionid",
"aspsessionid", "utm_source", "utm_medium", "utm_campaign",
"utm_term", "utm_content", "fbclid", "gclid", "ref", "referer",
"page", "p", "start", "offset", "from",
}
# How long a crawler process reuses its aggregate host/domain counts
# before reloading them from the database. Each process keeps its own
# view; slightly stale counts only shift priorities a little, never
# correctness, so no cross-process locking is needed.
HOST_STATS_CACHE_SECONDS = 30
# Row cap for the statistics queries in _load_host_stats(). These are
# arbitrary samples over tables without host columns (the schema must
# not be extended), not complete counts - they only steer priorities,
# never block URLs. Missed hosts are treated as unexplored, which errs
# in favor of spread.
HOST_STATS_SAMPLE_LIMIT = 20000
# ============================================================
# LOGGING / SHUTDOWN
@@ -250,7 +341,56 @@ def requeue_stale_items(conn, older_than_seconds=3600):
logging.info("Requeued %d stale processing URLs.", changed)
def _insert_queue_rows(conn, rows):
"""
Insert pre-ordered queue rows and return the number of newly
inserted rows.
All pages run inside ONE transaction (commit only at the end), so
the batch stays atomic as before. rowcount is accumulated per page
because execute_values() does not aggregate rowcount across pages.
"""
inserted = 0
with conn.cursor() as cur:
for start in range(0, len(rows), QUEUE_INSERT_PAGE_SIZE):
page = rows[start:start + QUEUE_INSERT_PAGE_SIZE]
execute_values(
cur,
"""
INSERT INTO crawl_queue (url, source_url, priority, status)
VALUES %s
ON CONFLICT (url) DO NOTHING
""",
page,
template="(%s, %s, %s, %s)",
page_size=QUEUE_INSERT_PAGE_SIZE,
)
inserted += max(cur.rowcount, 0)
conn.commit()
return inserted
def queue_urls(conn, links, source_url):
"""
Queue links for crawling, deadlock-safe for parallel crawlers that
share one crawl_queue and its UNIQUE index crawl_queue_url_key.
Strategy (no DB schema or index changes required):
- Clean and deduplicate the links within the batch.
- Compute priorities for all URLs (novelty + spread + source
preference - repetition penalties) with batched DB lookups only.
- Sort by priority, then deterministically by URL, so that
concurrent crawlers insert the same URLs in the same order into
the UNIQUE index. Lock acquisition on the index then happens in
a consistent order, which prevents most lock inversions
(deadlocks) at the source.
- ON CONFLICT (url) DO NOTHING keeps re-inserts harmless.
- If PostgreSQL still detects a deadlock, roll back the aborted
transaction and retry the whole batch a few times with a short
growing backoff instead of crashing the worker. Any other
database error is handled the same way.
"""
if not links:
return 0
@@ -276,24 +416,484 @@ def queue_urls(conn, links, source_url):
if not filtered_links:
return 0
rows = [(url, safe_source, 0, "pending") for url in filtered_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,
# Priorities are computed BEFORE the deterministic URL sort: the
# priority sort decides the crawl order, the final URL ordering
# keeps index lock acquisition consistent between crawlers.
try:
rows, stats = build_queue_rows(conn, filtered_links, safe_source)
except Exception:
# Priority computation must never take the worker down: fall
# back to the previous behavior (everything priority 0) and
# still queue the links.
logging.exception(
"[QUEUE] Priority computation failed; queuing without priorities."
)
inserted = cur.rowcount
try:
conn.rollback()
except Exception:
pass
unique_links = sorted(dict.fromkeys(filtered_links))
rows = [(url, safe_source, 0, "pending") for url in unique_links]
stats = {}
conn.commit()
return inserted
if stats:
log_queue_stats(safe_source, stats)
for attempt in range(1, QUEUE_INSERT_MAX_ATTEMPTS + 1):
try:
inserted = _insert_queue_rows(conn, rows)
if attempt > 1:
logging.info(
"[QUEUE] Insert succeeded on retry %d/%d.",
attempt,
QUEUE_INSERT_MAX_ATTEMPTS,
)
return inserted
except psycopg2.errors.DeadlockDetected as exc:
# The failed transaction has poisoned the connection's
# current transaction; roll it back so the connection stays
# usable for the retry and for everything after it.
try:
conn.rollback()
except Exception:
logging.exception("[QUEUE] Rollback after deadlock failed.")
if attempt < QUEUE_INSERT_MAX_ATTEMPTS:
delay = QUEUE_INSERT_RETRY_BASE_DELAY * (2 ** (attempt - 1))
logging.warning(
"[QUEUE] Deadlock while inserting %d URLs "
"(attempt %d/%d), retrying in %.1fs: %s",
len(rows),
attempt,
QUEUE_INSERT_MAX_ATTEMPTS,
delay,
exc,
)
time.sleep(delay)
except psycopg2.Error as exc:
# Any other database error (connection loss, constraint
# trouble, ...) must not crash the worker either. Same
# controlled retry with backoff as for deadlocks.
try:
conn.rollback()
except Exception:
logging.exception("[QUEUE] Rollback after DB error failed.")
if attempt < QUEUE_INSERT_MAX_ATTEMPTS:
delay = QUEUE_INSERT_RETRY_BASE_DELAY * (2 ** (attempt - 1))
logging.warning(
"[QUEUE] DB error while inserting %d URLs "
"(attempt %d/%d), retrying in %.1fs: %s",
len(rows),
attempt,
QUEUE_INSERT_MAX_ATTEMPTS,
delay,
exc,
)
time.sleep(delay)
# All retries exhausted. Log this clearly and keep the worker
# alive; the skipped URLs can be rediscovered from other pages.
logging.error(
"[QUEUE] Insert of %d URLs from %s failed after %d attempts. "
"Sample skipped URLs: %s",
len(rows),
safe_source or "-",
QUEUE_INSERT_MAX_ATTEMPTS,
", ".join(sorted(url for url, _s, _p, _st in rows)[:5])
+ ("..." if len(rows) > 5 else ""),
)
return 0
# ============================================================
# URL PRIORITY
# ============================================================
# Hosts known to serve near-infinite URL spaces (URL floods). Not a
# block list: links to these hosts are still queued, they just do not
# get preferential treatment for sheer quantity.
DIVERSITY_SOFT_HOSTS = {"github.com"}
# Cache for aggregate host/domain counts, per process. Multiple
# crawler processes/instances each maintain their own view; counts are
# advisory only (they shape priorities, never block), so a slightly
# stale view between reloads is harmless by design.
_HOST_STATS = {
"loaded_at": 0.0,
"host_counts": {}, # registrable domain -> crawled pages (pages)
"host_page_counts": {}, # hostname -> crawled pages (pages)
"queue_counts": {}, # hostname -> pending queue entries (crawl_queue)
}
def _registerable_domain(hostname):
"""
Registrable domain of a hostname, approximated with the last two
labels (example.com, sub.example.org -> example.com). Public Suffix
List handling would be more precise but is intentionally out of
scope here, matching the existing simple TLD logic in parse_html.
"""
if not hostname:
return ""
labels = [label for label in hostname.split(".") if label]
if len(labels) >= 2:
return ".".join(labels[-2:])
return "".join(labels)
def _split_url(url):
"""
Split a normalized URL into (hostname, registrable_domain) using
the same URL processing as the rest of the crawler (urlparse).
Falls back to the empty hostname rather than raising so that one
malformed URL can never break a batch.
"""
try:
hostname = (urlparse(url).hostname or "").lower().strip(".")
except ValueError:
return "", ""
return hostname, _registerable_domain(hostname)
def _strip_tracking_query(url):
"""
Remove tracking/session-style query parameters; returns None if the
URL has no such parameters at all.
"""
try:
parsed = urlparse(url)
except ValueError:
return None
if not parsed.query:
return None
query_pairs = [
(key, value) for key, value in parse_qsl(parsed.query, keep_blank_values=True)
if key.lower() not in VARIANT_QUERY_KEYS
]
if len(query_pairs) == len(parse_qsl(parsed.query)):
return None
query = "&".join(
f"{key}={value}" for key, value in query_pairs
)
netloc = parsed.netloc
stripped = f"{parsed.scheme}://{netloc}{parsed.path or '/'}"
if query:
stripped += "?" + query
return stripped
def _is_probable_variant(url, known_urls):
"""
Heuristic for "URL looks like a trivial variant of an already known
page": same address after dropping tracking/session/pagination
parameters, or a trailing-slash twin of a known URL. Purely local
computation - known_urls comes from a single batched DB lookup.
"""
if url in known_urls:
return True
stripped = _strip_tracking_query(url)
if stripped is not None and stripped in known_urls:
return True
if url.endswith("/") and url[:-1] in known_urls:
return True
if not url.endswith("/") and (url + "/") in known_urls:
return True
return False
def _load_host_stats(conn):
"""
Refresh the aggregate host/domain view if it is stale.
Two read-only grouped queries in ONE transaction, so this cannot
interact with concurrent queue inserts:
- pages.domain stores the crawl-time hostname; page counts are
folded to their registrable domain (and kept per host) in Python.
- crawl_queue hosts are extracted from the URL, so the queue counts
stay in sync with what is actually pending.
"""
now = time.monotonic()
if now - _HOST_STATS["loaded_at"] < HOST_STATS_CACHE_SECONDS:
return
try:
with conn.cursor() as cur:
cur.execute(
"SELECT domain FROM pages "
"ORDER BY domain LIMIT %s",
(HOST_STATS_SAMPLE_LIMIT,),
)
domain_counts = defaultdict(int)
host_page_counts = defaultdict(int)
for (host,) in cur.fetchall():
hostname = (host or "").lower().strip(".")
if not hostname:
continue
host_page_counts[hostname] += 1
domain_counts[_registerable_domain(hostname)] += 1
cur.execute(
"SELECT url FROM crawl_queue "
"WHERE status = 'pending' "
"ORDER BY url LIMIT %s",
(HOST_STATS_SAMPLE_LIMIT,),
)
queue_counts = defaultdict(int)
for (queued_url,) in cur.fetchall():
hostname, _domain = _split_url(queued_url)
if hostname:
queue_counts[hostname] += 1
except Exception:
# Statistics are an optimization. On failure keep the last view
# (or an empty one); the crawl continues with slightly worse
# priorities instead of losing URLs.
logging.exception("[QUEUE] Host statistics could not be loaded.")
try:
conn.rollback()
except Exception:
pass
_HOST_STATS["loaded_at"] = now
return
_HOST_STATS["host_counts"] = domain_counts
_HOST_STATS["host_page_counts"] = host_page_counts
_HOST_STATS["queue_counts"] = queue_counts
_HOST_STATS["loaded_at"] = now
def calculate_url_priority(
url,
*,
url_known,
domain_page_counts,
host_page_counts,
host_queued,
source_domain_pages,
batch_host_known,
):
"""
Central priority computation for one URL.
Concept (see the URL PRIORITY CONFIG constants):
priority = base
+ novelty bonuses for unknown URL / domain / host
+ spread bonus for little-represented hosts
+ source preference from the previously used rules
- repetition/clustering penalties
- variant penalty for near-duplicates of known pages
All inputs are plain integers looked up beforehand; this function
does no I/O of its own so it stays cheap and easy to extend.
"""
hostname, domain = _split_url(url)
priority = PRIORITY_BASE
# --- Novelty bonuses --------------------------------------
if not url_known:
priority += PRIORITY_BONUS_NEW_URL
domain_pages = domain_page_counts.get(domain, 0)
host_pages = host_page_counts.get(hostname, 0)
host_known = host_pages + host_queued
if domain_pages == 0:
# Never-crawled domain: the strongest exploration signal.
priority += PRIORITY_BONUS_NEW_DOMAIN
priority += PRIORITY_BONUS_NEW_HOST
elif host_pages == 0:
# New host within an already known domain (subdomain or www
# variant of an existing site).
priority += PRIORITY_BONUS_NEW_HOST
# --- Spread bonus -----------------------------------------
# Log2 falloff: a host with little presence in the crawl gets a
# strong boost, a huge host none at all. Unseen hosts do not get a
# spread bonus (they already receive the explicit new bonuses).
if host_known > 0:
spread = PRIORITY_SPREAD_BONUS_MAX - (
math.log2(1.0 + host_known / PRIORITY_SPREAD_BONUS_DIVISOR)
* PRIORITY_SPREAD_BONUS_MAX
/ math.log2(1.0 + PRIORITY_SPREAD_BONUS_DIVISOR)
)
priority += max(0.0, min(PRIORITY_SPREAD_BONUS_MAX, spread))
# --- Source preference ------------------------------------
# Integrated from the previous behavior: links out of little-
# explored territory are promising, links from an exhausted source
# are pulled back slightly.
if source_domain_pages == 0:
priority += PRIORITY_SOURCE_NOVELTY_BONUS
else:
priority -= (
math.log2(1.0 + source_domain_pages)
* PRIORITY_SOURCE_PENALTY_DIVISOR
)
# --- Repetition / clustering penalties ---------------------
if hostname in DIVERSITY_SOFT_HOSTS:
priority -= PRIORITY_CLUSTER_PENALTY
elif batch_host_known >= PRIORITY_CLUSTER_HOST_THRESHOLD:
# A flood of links to the same host within one page: flatten it
# instead of letting one host dominate the next queue window.
priority -= PRIORITY_CLUSTER_PENALTY
if url_known:
priority -= PRIORITY_VARIANT_PENALTY
# Keep the value in a sane integer range for the existing column.
return int(max(0, min(priority, 2 ** 31 - 1)))
def _find_known_urls(conn, urls):
"""
Look up which of the batch URLs already exist, in ONE query per
table using ANY(%s) instead of per-URL queries. Returns the set of
URLs that are already known (queued or crawled).
"""
known = set()
sample = urls[:VARIANT_CHECK_SAMPLE_LIMIT]
if not sample:
return known
try:
with conn.cursor() as cur:
cur.execute(
"SELECT url FROM crawl_queue WHERE url = ANY(%s)",
(sample,),
)
known.update(row[0] for row in cur.fetchall())
cur.execute(
"SELECT url FROM pages WHERE url = ANY(%s)",
(sample,),
)
known.update(row[0] for row in cur.fetchall())
except Exception:
logging.exception(
"[QUEUE] Known-URL lookup failed; assuming all URLs are new."
)
try:
conn.rollback()
except Exception:
pass
return known
return known
def build_queue_rows(conn, links, source_url):
"""
Turn sanitized, deduplicated links into queue rows ordered by the
computed priority (highest first).
Returns (rows, stats). Stats is a small dict for batch logging:
new_domains / new_hosts / known_hosts / spread_applied / min/max.
"""
unique_links = sorted(dict.fromkeys(links))
if not unique_links:
return [], {}
_load_host_stats(conn)
domain_page_counts = _HOST_STATS["host_counts"]
host_page_counts = _HOST_STATS["host_page_counts"]
queue_counts = _HOST_STATS["queue_counts"]
# Batched lookups only, never per-URL queries: one known-URL
# sample for variant detection plus per-batch host tallies.
known_urls = _find_known_urls(conn, unique_links)
host_known_in_batch = defaultdict(int)
for url in known_urls:
hostname, _domain = _split_url(url)
if hostname:
host_known_in_batch[hostname] += 1
_source_host, source_domain = _split_url(source_url or "")
source_domain_pages = domain_page_counts.get(source_domain, 0)
rows = []
new_domains = 0
new_hosts = 0
known_hosts = 0
spread_applied = 0
for url in unique_links:
hostname, domain = _split_url(url)
url_known = url in known_urls or _is_probable_variant(url, known_urls)
priority = calculate_url_priority(
url,
url_known=url_known,
domain_page_counts=domain_page_counts,
host_page_counts=host_page_counts,
host_queued=queue_counts.get(hostname, 0),
source_domain_pages=source_domain_pages,
batch_host_known=host_known_in_batch.get(hostname, 0),
)
# Batch statistics for the [QUEUE] log line.
domain_known = bool(domain_page_counts.get(domain, 0))
host_known = bool(
host_page_counts.get(hostname, 0)
or queue_counts.get(hostname, 0)
)
if domain and not domain_known:
new_domains += 1
if hostname:
if not host_known:
new_hosts += 1
elif not url_known:
known_hosts += 1
if priority > PRIORITY_BASE:
spread_applied += 1
rows.append((url, source_url, priority, "pending"))
# Highest priority first, URL as deterministic tiebreaker. The
# ordering also keeps concurrent batches inserting in a stable
# order, which is what keeps UNIQUE-index deadlocks rare.
rows.sort(key=lambda row: (-row[2], row[0]))
stats = {
"count": len(unique_links),
"new_domains": new_domains,
"new_hosts": new_hosts,
"known_hosts": known_hosts,
"spread_applied": spread_applied,
"priority_min": min((row[2] for row in rows), default=0),
"priority_max": max((row[2] for row in rows), default=0),
}
return rows, stats
def log_queue_stats(source_url, stats):
"""
One compact log line per batch so the journal later shows whether
the crawl actually spreads across more domains/hosts.
"""
if not stats:
return
logging.info(
"[QUEUE] %d URLs | new domains: %d | new hosts: %d | "
"known hosts: %d | priority range: %d-%d | "
"spread bonus applied: %d | source: %s",
stats.get("count", 0),
stats.get("new_domains", 0),
stats.get("new_hosts", 0),
stats.get("known_hosts", 0),
stats.get("priority_min", 0),
stats.get("priority_max", 0),
stats.get("spread_applied", 0),
source_url or "-",
)
# ============================================================