"""Custom COMPASS website crawler
Much more simplistic than the Crawl4AI crawler, but designed to access
some links that Crawl4AI cannot (such as those behind a button
interface).
"""
import math
import asyncio
import logging
import operator
from functools import cached_property
from collections import Counter
from contextlib import AsyncExitStack
from urllib.parse import urljoin, urlsplit
from crawl4ai.models import Link as c4AILink
from bs4 import BeautifulSoup
from rebrowser_playwright.async_api import async_playwright
from rebrowser_playwright.async_api import Error as RBPlaywrightError
from playwright._impl._errors import Error as PlaywrightError # ruff:ignore[import-private-name]
from elm.web.utilities import pw_page, PWKwargs, get_redirected_url
from elm.web.document import HTMLDocument
from elm.web.file_loader import AsyncWebFileLoader
from elm.web.website_crawl import ELMLinkScorer, _SCORE_KEY # ruff:ignore[import-private-name]
from compass.utilities.url import URLPartFilter, normalize_domain, sanitize_url
from compass.services.threaded import TempFileCache
from compass.web.file_loader import COMPASSWebFileLoader
from compass.utilities.parsing import is_pdf_doc
logger = logging.getLogger(__name__)
_DEPTH_KEY = "web_crawl_depth"
_CLICKABLE_SELECTORS = [
"button", # "a", "p"
]
_BLACKLIST_SUBSTRINGS = [
"facebook",
"twitter",
"linkedin",
"instagram",
"youtube",
"instagram",
"pinterest", # cspell: disable-line
"tiktok", # cspell: disable-line
"x.com",
"snapchat",
"reddit",
"mailto:",
"tel:",
"javascript:",
"login",
"signup",
"sign up",
"signin",
"sign in",
"register",
"subscribe",
"donate",
"shop",
"cart",
"careers",
"event",
"events",
"calendar",
]
DOC_THRESHOLD = 5
"""Default max documents to collect before terminating COMPASS crawl"""
[docs]
class COMPASSLinkScorer(ELMLinkScorer):
"""Custom URL scorer for COMPASS website crawling"""
def _assign_value(self, text):
"""Score based on the presence of keywords in link text"""
score = 0
text = text.casefold().replace("plant", "")
for kw, kw_score in self.keyword_points.items():
if kw in text:
score += kw_score
return score
class _Link(c4AILink):
"""Crawl4AI Link subclass with a few utilities"""
def __hash__(self):
return hash(self.href.casefold())
def __repr__(self):
return (
f"Link(title={self.title!r}, href={self.href!r}, "
f"base_domain={self.base_domain!r})"
)
def __str__(self):
return f"{self.title} ({self.href})"
def __eq__(self, other):
if isinstance(other, str):
return self.href.casefold() == other.casefold()
if not isinstance(other, c4AILink):
return NotImplemented
return self.href.casefold() == other.href.casefold()
@cached_property
def consistent_domain(self):
"""bool: ``True`` if the link is from the base domain"""
norm_b = normalize_domain(self.base_domain)
norm_href = normalize_domain(self.href)
is_consistent = norm_href == norm_b or (
bool(norm_b) and norm_href.endswith(f".{norm_b}")
)
logger.trace(
"Consistent domain between %s and %s: %r",
norm_href,
norm_b,
is_consistent,
)
return is_consistent
@property
def resembles_pdf(self):
"""bool: ``True`` if the link has "pdf" in title or href"""
return "pdf" in self.title.casefold() or "pdf" in self.href.casefold()
class _PageVisit:
"""Track traversal metadata and document identity for a visited URL
Navigation pages may be revisited at a shallower depth when crawling
with a depth limit. Documents have no descendants to explore, so
they never need a second visit. Document identity is retained even
when validation rejects the document, preventing repeated work.
The source identifies the fetched document, which may differ from
the visited URL, and is stored case-folded for matching. Source
aliases do not create additional visits.
"""
def __init__(self, depth, score, is_document=False, source=None):
"""
Parameters
----------
depth : int
Link depth of this visit, with the landing page at zero.
score : float
Relevance score assigned to the link for this visit.
is_document : bool, optional
Whether this visit corresponds to a document rather than a
navigation page. By default, ``False``.
source : str, optional
The source URL of the document, which may differ from the
visited URL. By default, ``None``.
"""
self.depth = depth
self.score = score
self.is_document = is_document
self.source = source
def can_revisit(self, depth):
"""Check whether a navigation-page visit is shallower
Parameters
----------
depth : int
Link depth of the proposed visit.
Returns
-------
bool
Whether this is a shallower visit to a navigation page.
"""
return not self.is_document and depth < self.depth
[docs]
class COMPASSCrawler:
"""A simple website crawler to search for ordinance documents"""
# ruff:ignore[too-many-arguments,too-many-positional-arguments]
def __init__(
self,
validator,
url_scorer,
file_loader_kwargs=None,
already_visited=None,
browser_semaphore=None,
num_link_scores_to_check_per_page=4,
max_pages=100,
max_same_score_links_per_page=20,
url_ignore_substrings=None,
url_keep_substrings=None,
max_depth=None,
):
"""
Parameters
----------
validator : callable
An async callable that takes a document instance (containing
the text from a PDF or a webpage) and returns a boolean
indicating whether the text passes the validation check.
This is used to determine whether or not to keep (i.e.
return) the document.
url_scorer : callable
An async callable that takes a list of dictionaries
containing URL information and assigns each dictionary a
`score` key representing the score for that URL. The input
URL dictionaries will each have at least one key: "href".
This key will contain the URL of the link. The dictionary
may also have other attributes such as "title", which
contains the link title text.
file_loader_kwargs : dict, optional
Additional keyword-value argument pairs to pass to the
:class:`~elm.web.file_loader.AsyncWebFileLoader` class. If
this dictionary contains the ``pw_launch_kwargs`` key, it's
value (assumes to be another dictionary) will be used to
initialize the playwright instances used for the crawl.
By default, ``None``.
already_visited : set, optional
A set of URLs (either strings or ``Link`` objects)
that have already been visited. This is used to avoid
re-visiting links that have already been checked.
By default, ``None``.
browser_semaphore : :class:`asyncio.Semaphore`, optional
Semaphore instance that can be used to limit the number of
playwright browsers open concurrently. If ``None``, no
limits are applied. By default, ``None``.
num_link_scores_to_check_per_page : int, default=4
Number of top unique-scoring links per page to use for
recursive crawling. This helps the crawl stay focused on the
most likely links to contain documents of interest.
max_pages : int, default=100
Maximum number of pages to crawl before terminating,
regardless of whether the document was found or not.
By default, ``100``.
max_same_score_links_per_page : int, default=20
Maximum number of links with the same score to check per
page. This helps prevent the crawler from spending too much
time on links that have identical scores.
By default, ``20``.
url_ignore_substrings : iterable of str, optional
URL parts that exclude matching crawl candidates. By
default, ``None``.
url_keep_substrings : iterable of str, optional
URL parts that override all crawl blacklist matches. By
default, ``None``.
max_depth : int, optional
Maximum link depth to check, including the starting page at
depth zero. By default, ``None`` (unlimited).
"""
self.validator = validator
self.url_scorer = url_scorer
self.num_scores_to_check_per_page = num_link_scores_to_check_per_page
self.checked_previously = already_visited or set()
self.max_pages = max_pages
self.max_same_score_links_per_page = max_same_score_links_per_page
self.max_depth = max_depth
self.url_filter = URLPartFilter(
[*_BLACKLIST_SUBSTRINGS, *(url_ignore_substrings or [])],
url_keep_substrings,
)
file_loader_kwargs = file_loader_kwargs or {}
flk = {"verify_ssl": False}
flk.update(file_loader_kwargs or {})
# Fast file loader that always uses poppler
self.fast_afl = AsyncWebFileLoader(**flk)
# best parsing file loader selected by user
self.final_afl = COMPASSWebFileLoader(**flk)
self.pw_launch_kwargs = PWKwargs.launch_kwargs()
self.pw_launch_kwargs.update(
file_loader_kwargs.get("pw_launch_kwargs") or {}
)
self.browser_semaphore = (
AsyncExitStack()
if browser_semaphore is None
else browser_semaphore
)
self._out_docs = []
self._already_visited = {}
self._failed_external_domains = set()
self._should_stop = None
[docs]
async def run(
self,
base_url,
crawl_timeout_s,
termination_callback=None,
on_new_page_visit_hook=None,
):
"""Run the COMPASS website crawler
Parameters
----------
base_url : str
URL of the website to start crawling from.
crawl_timeout_s : float, optional
A float representing the maximum time in seconds to allow
the crawl to run.
termination_callback : callable, optional
An async callable that takes a list of documents and returns
a boolean indicating whether to stop crawling. If ``None``,
the crawl will simply terminates when :obj:`DOC_THRESHOLD`
number of documents have been found. By default, ``None``.
on_new_page_visit_hook : callable, optional
An async callable that is called every time a new page is
found during the crawl. The callable should accept a single
argument, which is the page ``Link`` instance.
If ``None``, no additional processing is done on new pages.
By default, ``None``.
Returns
-------
list
List of document instances that passed the validation
check. Each document contains the text from a PDF and has an
attribute `source` that contains the URL of the document.
This list may be empty if no documents are found.
"""
self._should_stop = termination_callback or _default_found_enough_docs
try:
async with asyncio.timeout(crawl_timeout_s):
await self._run(
base_url, on_new_page_visit_hook=on_new_page_visit_hook
)
except TimeoutError:
logger.exception(
"COMPASS Website crawl deadline (%s) exceeded for %r; "
"returning %d collected crawl docs",
f"{int(crawl_timeout_s):,d}s",
base_url,
len(self._out_docs),
)
except Exception:
logger.exception(
"Error collecting documents using COMPASS web crawl for %r; "
"returning %d collected crawl docs",
base_url,
len(self._out_docs),
)
self._should_stop = None
self._log_crawl_stats()
self._out_docs.sort(key=lambda x: -1 * x.attrs[_SCORE_KEY])
return self._out_docs
# ruff:ignore[complex-structure]
# complexipy: ignore
async def _run(
self,
base_url,
link=None,
depth=0,
score=0,
on_new_page_visit_hook=None,
):
"""Recursive web crawl function"""
if link is None:
base_url, link = await self._reset_crawl(base_url)
logger.debug(
"Starting COMPASS crawl for base URL: %s\nLink: %r",
base_url,
link,
)
if self.url_filter.blacklist_match(link.href) is not None:
logger.debug("Skipping blacklisted URL: %s", link.href)
return
if self._document_source_seen(link.href):
logger.debug(
"Skipping previously collected document URL: %s", link.href
)
return
if not self._should_continue_page_crawl(link, depth):
return
if on_new_page_visit_hook:
await on_new_page_visit_hook(link)
visit = self._already_visited[link] = _PageVisit(depth, score)
logger.trace("self._already_visited=%r", self._already_visited)
if await self._website_link_is_doc(link, depth, score):
visit.is_document = True
return
if self.max_depth is not None and depth >= self.max_depth:
return
page_links = await self._get_links_from_page(link)
for next_link in self._top_scored_links(page_links, link):
prev_len = len(self._out_docs)
next_href = await get_redirected_url(
next_link["href"], verify=False
)
await self._run(
base_url,
link=_Link(
title=next_link["title"],
href=next_href,
base_domain=base_url,
),
depth=depth + 1,
score=next_link["score"],
on_new_page_visit_hook=on_new_page_visit_hook,
)
doc_was_just_found = ( # fmt: off
len(self._out_docs) == (prev_len + 1)
and (
self._out_docs[-1].attrs.get(_DEPTH_KEY, -1) == (depth + 1)
)
)
if doc_was_just_found:
if await self.validator(self._out_docs[-1]):
logger.debug(" - Document passed validation check!")
await self._load_last_doc_with_final_afl(next_link["href"])
else:
self._out_docs = self._out_docs[:-1]
if await self._should_terminate_crawl():
break
return
async def _load_last_doc_with_final_afl(self, link):
"""Load the last document with the final file loader"""
old_doc = self._out_docs[-1]
link = old_doc.attrs.get("source", link)
try:
doc = await self.final_afl.fetch(link)
doc.attrs[_DEPTH_KEY] = old_doc.attrs[_DEPTH_KEY]
doc.attrs[_SCORE_KEY] = old_doc.attrs[_SCORE_KEY]
if not doc.empty:
self._out_docs[-1] = doc
except KeyboardInterrupt:
raise
except Exception as e:
msg = (
"Encountered error of type %r while trying to fetch "
"content from %s"
)
err_type = type(e)
logger.exception(msg, err_type, link)
async def _reset_crawl(self, base_url):
"""Reset crawl state and initialize crawling link"""
self._out_docs = []
self._already_visited = {}
self._failed_external_domains = set()
base_url = sanitize_url(base_url)
base_url = await get_redirected_url(base_url, verify=False)
return base_url, _Link(
title="Landing Page",
href=sanitize_url(urljoin(base_url, base_url.split(" ")[0])),
base_domain=base_url,
)
async def _website_link_is_doc(self, link, depth, score):
"""Check if website link contains doc"""
if link in self.checked_previously and link.consistent_domain:
# Don't re-check pages on main website
return False
if await self._website_link_is_pdf(link, depth, score):
return True
# at this point the page is NOT a PDF. However, it could still
# just be a normal webpage on the main domain that we haven't
# visited before. In that case, just return False
if link.consistent_domain:
return False
# now we are on an external page that we either have not visited
# before or that we have seen but determined is NOT a PDF file
return await self._website_link_as_html_doc(link, depth, score)
async def _website_link_is_pdf(self, link, depth, score):
"""Fetch page content; keep only PDFs"""
if not link.consistent_domain and not link.resembles_pdf:
logger.debug(
"Skipping external non-PDF candidate: %s",
link.href,
)
return False
parsed = urlsplit(link.href)
if parsed.scheme not in {"http", "https"}:
logger.debug("Skipping non-http URL: %s", link.href)
return False
if (
not link.consistent_domain
and parsed.netloc.casefold() in self._failed_external_domains
):
logger.debug(
"Skipping external domain after previous fetch failure: %s",
parsed.netloc,
)
return False
logger.debug("Loading Link: %s", link)
return await self._loaded_link_is_pdf(link, depth, score, parsed)
async def _loaded_link_is_pdf(self, link, depth, score, parsed):
"""Check if the loaded link is a PDF document"""
try:
doc = await self.fast_afl.fetch(link.href)
except KeyboardInterrupt:
raise
except Exception as e:
msg = (
"Encountered error of type %r while trying to fetch "
"content from %s"
)
err_type = type(e)
logger.exception(msg, err_type, link)
if not link.consistent_domain and parsed.netloc:
self._failed_external_domains.add(parsed.netloc.casefold())
return False
if is_pdf_doc(doc):
logger.debug(" - Found PDF!")
doc.attrs[_DEPTH_KEY] = depth
doc.attrs[_SCORE_KEY] = score
source = doc.attrs.get("source", link.href).casefold()
if self._record_document(link, depth, score, source):
self._out_docs.append(doc)
return True
return False
async def _website_link_as_html_doc(self, link, depth, score):
"""Fetch page content as HTML doc"""
logger.debug("Loading Link as HTML: %s", link)
html_text = await self._get_text_no_err(link.href)
attrs = {_DEPTH_KEY: depth, _SCORE_KEY: score, "source": link.href}
doc = HTMLDocument([html_text], attrs=attrs)
cache_fn = await TempFileCache.call(doc, doc.text)
if cache_fn is not None:
doc.attrs["cache_fn"] = cache_fn
if self._record_document(link, depth, score, link.href):
self._out_docs.append(doc)
return True
def _should_continue_page_crawl(self, link, depth):
"""Determine whether the crawler should continue crawling"""
previous_visit = self._already_visited.get(link)
if previous_visit is None:
return True
if self.max_depth is None:
# We visited this page with no depth restriction, so we've
# already seen all other possible links - no need to recheck
return False
# We visited this page but potentially didn't check links on it
# due to depth restriction, so see if we are now visiting it
# with a shallower depth, which allows further exploration
return previous_visit.can_revisit(depth)
def _document_source_seen(self, source):
"""Check known document URLs and reported sources"""
source_key = str(source).casefold()
return any(
visit.is_document
and (
link.href.casefold() == source_key
or visit.source == source_key
)
for link, visit in self._already_visited.items()
)
def _record_document(self, link, depth, score, source):
"""Record a document and return whether its source is new"""
is_new = not self._document_source_seen(source)
visit = self._already_visited.setdefault(
link, _PageVisit(depth, score)
)
visit.is_document = True
visit.source = str(source).casefold()
return is_new
async def _get_links_from_page(self, link):
"""Get all links from a page sorted by relevance score"""
if not link.consistent_domain:
logger.debug("Detected new domain, stopping link discovery")
return []
html_text = await self._get_text_no_err(link.href)
page_links = []
if html_text:
page_links = _extract_links_from_html(
html_text, base_url=link.href, url_filter=self.url_filter
)
page_links = await self.url_scorer(
[dict(link) for link in page_links]
)
page_links = sorted(
page_links, key=operator.itemgetter("score"), reverse=True
)
_debug_info_on_links(page_links)
return page_links
def _top_scored_links(self, page_links, link):
"""Yield the configured number of links from each top score"""
num_link_scores_checked = 0
num_same_score_links_checked = 0
has_warned = False
curr_url_score = None
for page_link in page_links:
page_link_score = page_link["score"]
if page_link_score == 0:
continue
if curr_url_score != page_link_score:
if self._max_score_cats_checked(num_link_scores_checked, link):
return
curr_url_score = page_link_score
num_link_scores_checked += 1
num_same_score_links_checked = 0
has_warned = False
if self._max_same_score_links_checked(
num_same_score_links_checked
):
has_warned = self._warn_about_max_links(
has_warned, curr_url_score, link
)
continue
num_same_score_links_checked += 1
yield page_link
def _max_score_cats_checked(self, num_link_scores_checked, link):
"""Determine if the max score categories limit is reached"""
if num_link_scores_checked >= self.num_scores_to_check_per_page:
logger.debug(
"Reached max score categories (%d) to check for page: %s",
self.num_scores_to_check_per_page,
link.href,
)
return True
return False
def _max_same_score_links_checked(self, num_same_score_links_checked):
"""Determine if the same-score-link limit has been reached"""
return (
num_same_score_links_checked >= self.max_same_score_links_per_page
)
def _warn_about_max_links(self, has_warned, curr_url_score, link):
"""Log a warning for max links for the current score"""
if not has_warned:
logger.warning(
"Reached max links (%d) to check for score %d for page: %s",
self.max_same_score_links_per_page,
curr_url_score,
link.href,
)
return True
async def _get_text_no_err(self, url):
"""Get all text from a page; return empty string if pw error"""
try:
text = await self._get_text(url)
except (PlaywrightError, RBPlaywrightError):
text = ""
return text
async def _get_text(self, url):
"""Get all html text from a page"""
all_text = []
timeout_ms = 180_000 # milliseconds
pw_page_kwargs = {
"intercept_routes": True,
"ignore_https_errors": True,
"timeout": timeout_ms,
}
async with async_playwright() as p, self.browser_semaphore:
browser = await p.chromium.launch(**self.pw_launch_kwargs)
async with pw_page(browser, **pw_page_kwargs) as page:
await page.goto(url)
logger.debug(
"Waiting up to %d min for '%s' to load...",
math.ceil(timeout_ms / 60_000),
url,
)
await page.wait_for_load_state(
"networkidle", timeout=timeout_ms
)
all_text.append(await page.content())
all_text += await _get_text_from_all_locators(page)
return "\n".join(all_text)
async def _should_terminate_crawl(self):
"""Check if crawl should terminate"""
if await self._should_stop(self._out_docs):
logger.debug("Exiting crawl early due to user condition")
return True
if len(self._already_visited) >= self.max_pages:
logger.debug(" - Too many links visited, stopping recursion")
return True
logger.trace(
"Only checked %d pages, continuing crawl...",
len(self._already_visited),
)
return False
def _log_crawl_stats(self):
"""Log statistics about crawled pages and depths"""
logger.info("Crawled %d pages", len(self._already_visited))
logger.info("Found %d potential document(s)", len(self._out_docs))
logger.debug("Average score: %.2f", self._compute_avg_link_score())
logger.debug("Pages crawled by depth:")
for depth, count in sorted(self._crawl_depth_counts().items()):
logger.debug(" Depth %d: %d pages", depth, count)
def _compute_avg_link_score(self):
"""Compute the average score of the crawled results"""
if not self._already_visited:
return 0
return sum(
visit.score for visit in self._already_visited.values()
) / len(self._already_visited)
def _crawl_depth_counts(self):
"""Compute number of pages per depth"""
depth_counts = Counter()
depth_counts.update(
visit.depth for visit in self._already_visited.values()
)
return depth_counts
async def _default_found_enough_docs(out_docs): # ruff:ignore[unused-async]
"""Check if a predetermined # of documents has been found
The number to check is set by the module-level constant
:obj:`DOC_THRESHOLD`.
"""
return len(out_docs) >= DOC_THRESHOLD
def _debug_info_on_links(links):
"""Send debug info on links to logger"""
num_links = len(links)
if num_links <= 0:
logger.debug("Found no links!")
return
logger.debug("Found %d links:", len(links))
for link in links[:3]:
logger.debug(
" - %d: %s (%s)", link["score"], link["title"], link["href"]
)
if num_links > 3: # ruff:ignore[magic-value-comparison]
logger.debug(" ...")
def _extract_links_from_html(text, base_url, url_filter=None):
"""Parse HTML and extract all links"""
soup = BeautifulSoup(text, "html.parser")
links = [
(a.get_text().strip(), a["href"])
for a in soup.find_all("a", href=True)
]
return set(_sanitized_links(links, base_url, url_filter=url_filter))
def _sanitized_links(links, base_url, url_filter=None):
"""Sanitized links from the given list of (title, path) tuples"""
url_filter = url_filter or URLPartFilter(_BLACKLIST_SUBSTRINGS)
for title, path in links:
if not title or not path:
continue
href = sanitize_url(urljoin(base_url, path))
if urlsplit(href).scheme not in {"http", "https"}:
continue
if _is_blacklisted(title, path, href, url_filter):
continue
yield _Link(title=title, href=href, base_domain=base_url)
def _is_blacklisted(title, path, url, url_filter):
"""Check whether a link title or URL is blacklisted"""
if url_filter.is_whitelisted(url):
return False
title_and_path = f"{title}\n{path}".casefold()
if any(substr in title_and_path for substr in _BLACKLIST_SUBSTRINGS):
return True
return url_filter.blacklist_match(url) is not None
async def _get_text_from_all_locators(page):
"""Go through locators on page and get text behind them"""
all_text = []
for selector in _CLICKABLE_SELECTORS:
logger.trace("Checking selector %r", selector)
locators = page.locator(selector)
locator_count = await locators.count()
logger.trace(" - Found %d instances", locator_count)
for index in range(locator_count):
try:
text = await _get_locator_text(locators, index, page)
except (PlaywrightError, RBPlaywrightError):
continue
if text:
all_text.append(text)
return all_text
async def _get_locator_text(locators, index, page):
"""Get text after clicking on one of the page locators"""
locator = locators.nth(index)
if not await locator.is_visible():
return None
if not await locator.is_enabled():
return None
await locator.click(timeout=10_000)
return await page.content()
