integrity_scanner_fuer_statische_Webseiten/scanner/crawler.py

"""Pure-Python crawler — no wget dependency."""
import logging
import time
from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse

import requests
from bs4 import BeautifulSoup

logger = logging.getLogger(__name__)

# Query-param keys that indicate calendar / session / tracking / cache noise.
# Matched as whole keys (not prefixes) so legitimate params like ?view= or
# ?value= are never silently dropped. utm_* is a known tracking prefix family.
_NOISE_PARAM_KEYS = {
    "year", "month", "day", "date", "week",
    "session", "token", "sid", "nonce", "fbclid",
    "v", "ver", "version", "cache", "cb", "_", "_t",
}


def _is_noise_param(key: str) -> bool:
    """True if a query-param key is calendar/session/tracking/cache noise."""
    k = key.lower()
    return k in _NOISE_PARAM_KEYS or k.startswith("utm_")


def normalize_url(url: str) -> str | None:
    """
    Normalise a URL: lowercase scheme+host, drop fragment, sort and filter
    query params that look like cache-busters or tracking noise.
    Returns None for non-http(s) URLs.
    """
    try:
        p = urlparse(url.strip())
        if p.scheme not in ("http", "https"):
            return None
        netloc = p.netloc.lower()
        path = p.path or "/"
        # Remove fragment
        p = p._replace(netloc=netloc, path=path, fragment="")
        # Keep only non-noise query params, sorted for stable comparison
        if p.query:
            clean_params = sorted(
                (k, v)
                for k, vs in parse_qs(p.query, keep_blank_values=True).items()
                for v in vs
                if not _is_noise_param(k)
            )
            p = p._replace(query=urlencode(clean_params))
        return urlunparse(p)
    except Exception:
        return None


def should_crawl(
    url: str,
    base_netloc: str,
    skip_extensions: list[str],
    exclude_paths: tuple | list = (),
) -> bool:
    """Return True if url should be followed during a crawl."""
    p = urlparse(url)
    if p.netloc != base_netloc:
        return False
    path_lower = p.path.lower()
    for ext in skip_extensions:
        if path_lower.endswith(ext):
            return False
    if any(p.path.startswith(ep) for ep in exclude_paths):
        return False
    # Skip URLs whose query carries calendar/session/tracking noise
    if p.query and any(_is_noise_param(k) for k in parse_qs(p.query)):
        return False
    return True


class Crawler:
    def __init__(self, cfg: dict):
        self.target = cfg["target"].rstrip("/")
        self.base_netloc = urlparse(self.target).netloc
        self.max_pages: int = cfg["crawl"]["max_pages"]
        self.delay: float = cfg["crawl"]["delay_seconds"]
        self.skip_ext: list[str] = cfg["crawl"]["skip_extensions"]
        self.exclude_paths: list[str] = cfg["crawl"].get("exclude_paths", [])
        self.timeout: int = cfg["request_timeout"]
        self.headers = {"User-Agent": cfg["user_agent"]}
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def crawl(self) -> dict:
        """
        Crawl the target site breadth-first.

        Returns a dict with:
          - pages: list of page-dicts (url, final_url, status, html, headers, ...)
          - errors: list of {url, error}
        """
        seen: set[str] = set()
        queue: list[str] = [normalize_url(self.target + "/") or self.target + "/"]
        pages: list[dict] = []
        errors: list[dict] = []

        try:
            self._crawl_loop(queue, seen, pages, errors)
        finally:
            self.session.close()

        logger.info(
            "Crawl finished: %d pages, %d errors, %d queued but skipped",
            len(pages),
            len(errors),
            len(queue),
        )
        return {
            "target": self.target,
            "pages": pages,
            "errors": errors,
        }

    def _crawl_loop(
        self,
        queue: list[str],
        seen: set[str],
        pages: list[dict],
        errors: list[dict],
    ) -> None:
        while queue and len(pages) < self.max_pages:
            url = queue.pop(0)
            if url in seen:
                continue
            seen.add(url)

            if self.delay > 0 and pages:
                time.sleep(self.delay)

            logger.debug("Crawling %s", url)
            try:
                resp = self.session.get(
                    url,
                    timeout=self.timeout,
                    allow_redirects=True,
                )
                final_url = normalize_url(resp.url) or resp.url
                content_type = resp.headers.get("content-type", "")
                is_html = "text/html" in content_type and resp.status_code == 200

                page: dict = {
                    "url": url,
                    "final_url": final_url,
                    "status": resp.status_code,
                    "content_type": content_type,
                    "response_headers": dict(resp.headers),
                    "html": resp.text if is_html else None,
                    "redirected": final_url != url,
                }
                pages.append(page)

                if is_html:
                    for link_url in self._extract_follow_links(resp.text, final_url):
                        if link_url not in seen and link_url not in queue:
                            queue.append(link_url)

            except requests.exceptions.RequestException as exc:
                errors.append({"url": url, "error": str(exc)})
                logger.warning("Error crawling %s: %s", url, exc)

    def _extract_follow_links(self, html: str, base_url: str) -> list[str]:
        """Extract internal <a href> links worth following."""
        result = []
        try:
            soup = BeautifulSoup(html, "lxml")
            for tag in soup.find_all("a", href=True):
                href = tag["href"].strip()
                abs_url = normalize_url(urljoin(base_url, href))
                if abs_url and should_crawl(abs_url, self.base_netloc, self.skip_ext, self.exclude_paths):
                    result.append(abs_url)
        except Exception as exc:
            logger.debug("Link extraction failed on %s: %s", base_url, exc)
        return result