backend/core/crawl4ai_client.py

"""
Shared Crawl4AI scraper — darmowa alternatywa dla Firecrawl.
Domyślnie używa lekkiej strategii HTTP (bez Playwright/Chromium).
Ustaw CRAWL4AI_USE_BROWSER=1 dla stron wymagających renderowania JS.
"""

import logging
import os
import re
import time
from typing import List, Optional
from urllib.parse import urljoin, urlparse

from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

logger = logging.getLogger(__name__)

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

# Domeny z twardą blokadą (Imperva/403) — pomijamy Crawl4AI, używamy Jina/BS4 upstream
_HARD_BLOCKED_DOMAINS = frozenset(
    {
        "parp.gov.pl",
        "www.parp.gov.pl",
        "bgk.pl",
        "www.bgk.pl",
    }
)
_SSL_RELAXED_DOMAINS = frozenset(
    {
        "zus.pl",
        "www.zus.pl",
        "prewencja.zus.pl",
    }
)


def _domain_of(url: str) -> str:
    try:
        return urlparse(url).netloc.lower()
    except Exception:
        return ""


def is_hard_blocked_url(url: str) -> bool:
    """Public helper — PARP/BGK skip Crawl4AI (Imperva)."""
    return _is_hard_blocked(url)


def _is_hard_blocked(url: str) -> bool:
    domain = _domain_of(url)
    return any(domain == d or domain.endswith("." + d) for d in _HARD_BLOCKED_DOMAINS)


def _is_ssl_relaxed(url: str) -> bool:
    domain = _domain_of(url)
    return any(domain == d or domain.endswith("." + d) for d in _SSL_RELAXED_DOMAINS)


def _use_browser() -> bool:
    return os.environ.get("CRAWL4AI_USE_BROWSER", "").lower() in ("1", "true", "yes")


async def _scrape_ssl_relaxed(url: str) -> str:
    """Requests+BS4 z wyłączoną weryfikacją TLS (zus.pl — cert chain issues on HF)."""
    import asyncio

    import requests
    import urllib3
    from bs4 import BeautifulSoup

    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    headers = {
        "User-Agent": _DEFAULT_UA,
        "Accept-Language": "pl-PL,pl;q=0.9,en-US;q=0.8,en;q=0.7",
    }

    def _fetch() -> str:
        resp = requests.get(url, headers=headers, timeout=30, verify=False, allow_redirects=True)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, "html.parser")
        for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
            tag.decompose()
        return soup.get_text(separator="\n", strip=True)

    try:
        text = await asyncio.to_thread(_fetch)
        if text:
            logger.info("[Crawl4AI] SSL-relaxed fetch OK dla %s (%d znaków)", url, len(text))
        return text
    except Exception as exc:
        logger.warning("[Crawl4AI] SSL-relaxed fetch failed dla %s: %s", url, exc)
        return ""


def _run_config():
    from crawl4ai import CacheMode, CrawlerRunConfig

    return CrawlerRunConfig(cache_mode=CacheMode.BYPASS, word_count_threshold=10)


def _build_crawler():
    from crawl4ai import AsyncWebCrawler, BrowserConfig, HTTPCrawlerConfig
    from crawl4ai.async_crawler_strategy import AsyncHTTPCrawlerStrategy

    if _use_browser():
        browser_cfg = BrowserConfig(headless=True, verbose=False, text_mode=True)
        return AsyncWebCrawler(browser_config=browser_cfg, verbose=False)

    http_cfg = HTTPCrawlerConfig(
        headers={
            "User-Agent": _DEFAULT_UA,
            "Accept-Language": "pl-PL,pl;q=0.9,en-US;q=0.8,en;q=0.7",
        }
    )
    strategy = AsyncHTTPCrawlerStrategy(browser_config=http_cfg)
    return AsyncWebCrawler(crawler_strategy=strategy, verbose=False)


def _result_item(result):
    try:
        return result[0]
    except (IndexError, TypeError, KeyError):
        return result


def _markdown_from_result(result) -> str:
    if result is None or not getattr(result, "success", False):
        return ""
    item = _result_item(result)
    if not getattr(item, "success", True):
        return ""
    md = getattr(item, "markdown", None)
    if md is None:
        return ""
    if hasattr(md, "raw_markdown"):
        return (md.raw_markdown or "").strip()
    return str(md).strip()


def _links_from_result(result, base_url: str) -> List[str]:
    item = _result_item(result)
    links_dict = getattr(item, "links", None) or {}
    found = []
    for group in ("internal", "external"):
        for link in links_dict.get(group, []):
            href = link.get("href") if isinstance(link, dict) else str(link)
            if not href:
                continue
            if href.startswith("http"):
                found.append(href)
            else:
                found.append(urljoin(base_url, href))
    return found


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(Exception),
)
async def scrape_url_to_markdown(url: str) -> str:
    """Pobiera stronę i zwraca czysty Markdown."""
    if _is_hard_blocked(url):
        logger.debug("[Crawl4AI] Domena zablokowana (Imperva) — pomijam: %s", url)
        return ""
    if _is_ssl_relaxed(url):
        return await _scrape_ssl_relaxed(url)
    config = _run_config()
    async with _build_crawler() as crawler:
        result = await crawler.arun(url=url, config=config)
        md = _markdown_from_result(result)
        if not md:
            item = _result_item(result)
            err = getattr(item, "error_message", "") or ""
            logger.warning(f"[Crawl4AI] Pusty wynik dla {url}" + (f": {err}" if err else ""))
        return md


async def discover_links(
    url: str,
    search_terms: Optional[List[str]] = None,
    limit: int = 10,
) -> List[str]:
    """Odkrywa podstrony na tej samej domenie (zamiennik Firecrawl map_url)."""
    config = _run_config()
    async with _build_crawler() as crawler:
        result = await crawler.arun(url=url, config=config)
        all_links = _links_from_result(result, url)

    base_domain = urlparse(url).netloc
    terms = search_terms or ["regulamin", "doc", "pdf", "wytyczne"]
    pattern = re.compile("|".join(re.escape(t) for t in terms), re.I)

    filtered = []
    for link in all_links:
        if urlparse(link).netloc != base_domain:
            continue
        if pattern.search(link):
            filtered.append(link)

    if not filtered:
        filtered = [link for link in all_links if urlparse(link).netloc == base_domain]

    seen = set()
    unique = []
    for link in filtered:
        if link not in seen:
            seen.add(link)
            unique.append(link)
    return unique[:limit]


async def check_crawl4ai_health() -> dict:
    """Health check dla panelu admina."""
    start = time.perf_counter()
    try:
        md = await scrape_url_to_markdown("https://example.com")
        latency = int((time.perf_counter() - start) * 1000)
        if md:
            mode = "browser" if _use_browser() else "http"
            return {
                "status": "ok",
                "message": f"Crawl4AI działa (tryb: {mode})",
                "latency_ms": latency,
            }
        return {"status": "error", "message": "Crawl4AI zwrócił pusty wynik", "latency_ms": latency}
    except Exception as e:
        latency = int((time.perf_counter() - start) * 1000)
        return {"status": "error", "message": str(e), "latency_ms": latency}