integrity_scanner_fuer_statische_Webseiten/scanner/crawler.py

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

import requests
from bs4 import BeautifulSoup

logger = logging.getLogger(__name__)

_BROWSER_UA = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)

# 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 = (),
    include_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
    if include_paths and not any(p.path.startswith(ip) for ip in include_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.include_paths: list[str] = cfg["crawl"].get("include_paths", [])
        self.timeout: int = cfg["request_timeout"]
        self.headers = {"User-Agent": cfg["user_agent"]}
        self.session = requests.Session()
        self.session.headers.update(self.headers)
        self.use_sitemap: bool = cfg["crawl"].get("sitemap", True)
        self.on_progress: object = None  # callable(fetched: int, queued: int)

    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()
        if self.include_paths:
            queue: list[str] = [
                normalize_url(self.target + p) or (self.target + p)
                for p in self.include_paths
                if should_crawl(
                    normalize_url(self.target + p) or (self.target + p),
                    self.base_netloc, self.skip_ext,
                    self.exclude_paths, self.include_paths,
                )
            ]
        else:
            seed = normalize_url(self.target + "/") or self.target + "/"
            queue: list[str] = [seed] if should_crawl(
                seed, self.base_netloc, self.skip_ext,
                self.exclude_paths, self.include_paths,
            ) else []
        pages: list[dict] = []
        errors: list[dict] = []

        if self.use_sitemap:
            n = self._seed_from_sitemap(queue, seen)
            if n:
                logger.info("Sitemap: %d additional URL(s) added to queue", n)

        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,
                )
                # 403 often means bot-UA blocking; retry with browser UA
                if resp.status_code == 403:
                    try:
                        r2 = self.session.get(
                            url, timeout=self.timeout, allow_redirects=True,
                            headers={"User-Agent": _BROWSER_UA},
                        )
                        if r2.status_code != 403:
                            resp = r2
                    except requests.exceptions.RequestException:
                        pass
                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)

                if self.on_progress:
                    self.on_progress(len(pages), len(queue))

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

    def _find_sitemap_urls(self) -> list[str]:
        """Return sitemap URLs listed in robots.txt, or empty list."""
        urls = []
        try:
            resp = self.session.get(
                f"{self.target}/robots.txt", timeout=self.timeout
            )
            if resp.status_code == 200:
                for line in resp.text.splitlines():
                    if line.lower().startswith("sitemap:"):
                        url = line.split(":", 1)[1].strip()
                        if url:
                            urls.append(url)
        except requests.exceptions.RequestException:
            pass
        return urls

    def _seed_from_sitemap(self, queue: list[str], seen: set[str]) -> int:
        """
        Fetch sitemap(s) and add unseen URLs to queue.
        Handles sitemap index files (one level of nesting).
        Returns the number of URLs added.
        """
        _SM = "http://www.sitemaps.org/schemas/sitemap/0.9"

        sitemap_urls = self._find_sitemap_urls()
        if not sitemap_urls:
            sitemap_urls = [f"{self.target}/sitemap.xml"]

        visited: set[str] = set()
        pending = list(sitemap_urls)
        added = 0

        while pending:
            smap_url = pending.pop(0)
            if smap_url in visited:
                continue
            visited.add(smap_url)

            try:
                resp = self.session.get(smap_url, timeout=self.timeout)
                if resp.status_code != 200:
                    continue
                root = ET.fromstring(resp.content)
            except Exception:
                continue

            tag = root.tag.split("}")[-1] if "}" in root.tag else root.tag

            if tag == "sitemapindex":
                # Sitemap index — collect child sitemap URLs
                for loc in root.iter(f"{{{_SM}}}loc"):
                    child_url = loc.text.strip() if loc.text else ""
                    if child_url and child_url not in visited:
                        pending.append(child_url)
            else:
                # Regular sitemap — collect page URLs
                for loc in root.iter(f"{{{_SM}}}loc"):
                    raw = loc.text.strip() if loc.text else ""
                    url = normalize_url(raw)
                    if url and should_crawl(
                        url, self.base_netloc, self.skip_ext, self.exclude_paths,
                        self.include_paths,
                    ) and url not in seen and url not in queue:
                        queue.append(url)
                        added += 1

        return added

    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, self.include_paths):
                    result.append(abs_url)
        except Exception as exc:
            logger.debug("Link extraction failed on %s: %s", base_url, exc)
        return result