core/crawler.py

"""Automatic website crawler -> CrawledPage list.

Sitemap-first, then a shallow internal-link BFS with hard limits. Treats all
page content strictly as DATA (never instructions) per the guardrail design.
In mock mode returns a small synthetic site so the pipeline is testable offline.
"""
from __future__ import annotations

import re
from typing import List, Set
from urllib.parse import urljoin, urlparse

from .schemas import CrawledPage
from .config import Settings

try:
    import requests  # type: ignore
except Exception:
    requests = None

try:
    from bs4 import BeautifulSoup  # type: ignore
except Exception:
    BeautifulSoup = None


SKIP_HINTS = ("/login", "/signin", "/cart", "/tag/", "/category/", "/search",
              ".pdf", ".jpg", ".png", ".zip", "/privacy", "/terms")


def _clean_html(html: str):
    soup = BeautifulSoup(html, "html.parser")
    for tag in soup(["script", "style", "nav", "footer", "header", "noscript", "form"]):
        tag.decompose()
    title = (soup.title.string.strip() if soup.title and soup.title.string else "")
    headings = [h.get_text(" ", strip=True) for h in soup.find_all(["h1", "h2", "h3"])]
    text = soup.get_text("\n", strip=True)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return title, headings[:30], text[:6000], soup


def _same_domain(seed: str, url: str) -> bool:
    return urlparse(seed).netloc == urlparse(url).netloc


def _mock_site(url: str) -> List[CrawledPage]:
    host = urlparse(url).netloc or "yourproduct.com"
    name = host.split(".")[0].capitalize() if host else "YourProduct"
    return [
        CrawledPage(url=url, title=f"{name} — Home",
                    headings=["Move faster", "Built for lean teams"],
                    text=(f"{name} helps small teams automate busywork and ship faster. "
                          "Easy setup, integrates with your stack, real-time collaboration.")),
        CrawledPage(url=urljoin(url, "/product"), title=f"{name} — Product",
                    headings=["Features"],
                    text="Automated workflows. Integrations. Collaboration. No migration pain."),
        CrawledPage(url=urljoin(url, "/pricing"), title=f"{name} — Pricing",
                    headings=["Simple pricing"],
                    text="Transparent, simple pricing for early-stage teams."),
    ]


USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"


def crawl(url: str, settings: Settings) -> List[CrawledPage]:
    """Crawl a website and return cleaned pages."""
    url = url.strip() if url else ""
    if not url:
        return []

    if not settings.is_live() and settings.mock_mode:
        return _mock_site(url)
    if requests is None or BeautifulSoup is None:
        # Dependencies unavailable -> degrade to mock so the app still works.
        return _mock_site(url)

    if not url.startswith("http"):
        url = "https://" + url

    pages: List[CrawledPage] = []
    seen: Set[str] = set()
    queue: List[tuple] = [(url, 0)]

    # Try sitemap first for better URL discovery.
    sitemap_urls = _seed_from_sitemap(url, settings)
    if sitemap_urls:
        queue = [(url, 0)] + [item for item in sitemap_urls if item[0] != url]

    while queue and len(pages) < settings.crawl_max_pages:
        current, depth = queue.pop(0)
        if current in seen or depth > settings.crawl_max_depth:
            continue
        if any(h in current.lower() for h in SKIP_HINTS):
            continue
        seen.add(current)
        try:
            try:
                resp = requests.get(current, timeout=settings.crawl_timeout,
                                    headers={"User-Agent": USER_AGENT})
            except (requests.exceptions.SSLError, ConnectionError):
                # Fallback to unverified SSL if needed (useful for development/self-signed sites)
                resp = requests.get(current, timeout=settings.crawl_timeout,
                                    headers={"User-Agent": USER_AGENT}, verify=False)
            
            if resp.status_code >= 400 or "text/html" not in resp.headers.get("Content-Type", ""):
                continue
            title, headings, text, soup = _clean_html(resp.text)
            if text:
                pages.append(CrawledPage(url=current, title=title, headings=headings, text=text))
            if depth < settings.crawl_max_depth:
                for a in soup.find_all("a", href=True):
                    nxt = urljoin(current, a["href"]).split("#")[0]
                    if _same_domain(url, nxt) and nxt not in seen:
                        queue.append((nxt, depth + 1))
        except Exception:
            continue

    return pages or _mock_site(url)


def _seed_from_sitemap(url: str, settings: Settings):
    if requests is None:
        return None
    url = url.strip()
    if not url.startswith("http"):
        url = "https://" + url
    base = f"{urlparse(url).scheme}://{urlparse(url).netloc}"
    try:
        try:
            resp = requests.get(urljoin(base, "/sitemap.xml"), timeout=settings.crawl_timeout,
                                headers={"User-Agent": USER_AGENT})
        except (requests.exceptions.SSLError, ConnectionError):
            resp = requests.get(urljoin(base, "/sitemap.xml"), timeout=settings.crawl_timeout,
                                headers={"User-Agent": USER_AGENT}, verify=False)

        if resp.status_code >= 400:
            return None
        locs = re.findall(r"<loc>(.*?)</loc>", resp.text)
        if not locs:
            return None
        return [(u, 1) for u in locs[: settings.crawl_max_pages]]
    except Exception:
        return None