Web scraping для RAG: Firecrawl, BeautifulSoup, структурированная загрузка

import os
from firecrawl import FirecrawlApp   # pip install firecrawl-py
from core.document import Document


class FirecrawlScraper:
    """
    Обёртка над Firecrawl API.
    Возвращает Document с clean Markdown из любого URL.
    """

    def __init__(self, api_key: str | None = None):
        self._app = FirecrawlApp(
            api_key=api_key or os.environ["FIRECRAWL_API_KEY"]
        )

    def scrape(self, url: str) -> Document | None:
        """Загружает одну страницу, возвращает Document или None."""
        result = self._app.scrape_url(
            url,
            params={
                "formats": ["markdown"],        # только Markdown, не HTML
                "onlyMainContent": True,        # убрать навигацию и footer
                "excludeTags": ["nav", "header", "footer", "aside"],
                "waitFor": 1000,                # ждать 1с после JS-рендеринга
            }
        )

        if not result or not result.get("markdown"):
            return None

        metadata = result.get("metadata", {})
        return Document(
            page_content=result["markdown"],
            metadata={
                "source":      url,
                "title":       metadata.get("title", ""),
                "description": metadata.get("description", ""),
                "og_image":    metadata.get("ogImage", ""),
                "language":    metadata.get("language", ""),
                "scraper":     "firecrawl",
            }
        )

import time
from firecrawl import FirecrawlApp
from core.document import Document


def crawl_site(
    url: str,
    max_pages: int = 200,
    include_paths: list[str] | None = None,  # ["^/docs/", "^/api/"]
    exclude_paths: list[str] | None = None,  # ["^/blog/", "^/changelog/"]
    api_key: str | None = None,
) -> list[Document]:
    """
    Обходит весь сайт начиная с url.
    include_paths / exclude_paths — regex-паттерны для фильтрации URL.
    """
    app = FirecrawlApp(api_key=api_key or os.environ["FIRECRAWL_API_KEY"])

    crawl_params = {
        "limit": max_pages,
        "scrapeOptions": {
            "formats": ["markdown"],
            "onlyMainContent": True,
        },
    }
    if include_paths:
        crawl_params["includePaths"] = include_paths
    if exclude_paths:
        crawl_params["excludePaths"] = exclude_paths

    # Запускаем crawl — он асинхронный
    job = app.async_crawl_url(url, params=crawl_params)
    job_id = job["id"]
    print(f"Crawl started: job {job_id}")

    # Polling статуса
    while True:
        status = app.check_crawl_status(job_id)
        state = status.get("status")
        completed = status.get("completed", 0)
        total = status.get("total", "?")
        print(f"  {state}: {completed}/{total} pages")

        if state == "completed":
            break
        if state == "failed":
            raise RuntimeError(f"Crawl failed: {status.get('error')}")
        time.sleep(3)

    # Собираем результаты
    documents = []
    for page in status.get("data", []):
        md = page.get("markdown", "")
        if not md or not md.strip():
            continue
        meta = page.get("metadata", {})
        documents.append(Document(
            page_content=md,
            metadata={
                "source":      page.get("url", ""),
                "title":       meta.get("title", ""),
                "description": meta.get("description", ""),
                "scraper":     "firecrawl",
            }
        ))

    print(f"Crawl complete: {len(documents)} pages loaded")
    return documents


# Пример: загружаем документацию, только раздел /docs/
if __name__ == "__main__":
    docs = crawl_site(
        url="https://docs.example.com",
        max_pages=150,
        include_paths=["^/docs/"],
        exclude_paths=["^/docs/changelog", "^/docs/release-notes"],
    )
    for d in docs[:3]:
        print(d)

import re
import httpx
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
from core.document import Document

# Заголовки, имитирующие обычный браузер
DEFAULT_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (X11; Linux x86_64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept-Language": "en-US,en;q=0.9,ru;q=0.8",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}

NOISE_TAGS = {
    "script", "style", "nav", "header", "footer",
    "aside", "form", "button", "noscript", "iframe",
    "svg", "figure", "figcaption",
}
NOISE_CLASS_RE = re.compile(
    r"(nav|menu|footer|header|sidebar|cookie|banner|breadcrumb"
    r"|social|share|comment|related|ad-|advertisement|popup)",
    re.I,
)


def scrape_page(url: str, timeout: int = 15) -> Document | None:
    """Загружает статическую HTML-страницу и извлекает основной контент."""
    with httpx.Client(headers=DEFAULT_HEADERS, timeout=timeout,
                      follow_redirects=True) as client:
        response = client.get(url)

    if response.status_code != 200:
        print(f"  HTTP {response.status_code}: {url}")
        return None

    soup = BeautifulSoup(response.text, "lxml")

    # Мета-данные
    title = soup.title.get_text(strip=True) if soup.title else ""
    description = ""
    if (m := soup.find("meta", attrs={"name": "description"})):
        description = m.get("content", "")

    # Удаляем шум
    for tag in soup.find_all(NOISE_TAGS):
        tag.decompose()
    for tag in soup.find_all(True):
        classes = " ".join(tag.get("class", []))
        if NOISE_CLASS_RE.search(classes) or NOISE_CLASS_RE.search(tag.get("id", "")):
            tag.decompose()

    # Ищем основной контент
    content_area = (
        soup.find("article")
        or soup.find("main")
        or soup.find(attrs={"role": "main"})
        or soup.find(id=re.compile(r"content|article|main", re.I))
        or soup.body
    )
    if not content_area:
        return None

    text = _html_to_markdown(content_area)
    text = re.sub(r"\n{3,}", "\n\n", text).strip()

    if len(text) < 100:   # слишком мало текста — пропускаем
        return None

    return Document(
        page_content=text,
        metadata={
            "source":      url,
            "title":       title,
            "description": description,
            "scraper":     "beautifulsoup",
        }
    )


def _html_to_markdown(element) -> str:
    """Рекурсивно конвертирует HTML-элемент в Markdown."""
    parts = []
    for child in element.children:
        if not hasattr(child, "name"):
            t = str(child).strip()
            if t:
                parts.append(t)
            continue
        name = child.name
        inner = _html_to_markdown(child).strip()
        if name == "h1":
            parts.append(f"\n# {inner}\n")
        elif name == "h2":
            parts.append(f"\n## {inner}\n")
        elif name in ("h3", "h4", "h5"):
            parts.append(f"\n### {inner}\n")
        elif name == "p":
            if inner:
                parts.append(f"\n{inner}\n")
        elif name in ("ul", "ol"):
            parts.append(inner)
        elif name == "li":
            parts.append(f"- {inner}")
        elif name in ("strong", "b"):
            parts.append(f"**{inner}**")
        elif name in ("em", "i"):
            parts.append(f"*{inner}*")
        elif name == "code":
            parts.append(f"`{inner}`")
        elif name == "pre":
            code = child.get_text()
            lang = ""
            if (c := child.find("code")):
                classes = c.get("class", [])
                for cls in classes:
                    if cls.startswith("language-"):
                        lang = cls[9:]
            parts.append(f"\n```{lang}\n{code.strip()}\n```\n")
        elif name == "a":
            href = child.get("href", "")
            if inner and href:
                parts.append(f"[{inner}]({href})")
            else:
                parts.append(inner)
        elif name == "table":
            parts.append(_table_to_markdown(child))
        elif name in ("div", "section", "article", "span"):
            if inner:
                parts.append(inner)
    return "\n".join(p for p in parts if p.strip())

def _table_to_markdown(table_tag) -> str:
    """Конвертирует HTML-таблицу в Markdown."""
    rows = []
    for tr in table_tag.find_all("tr"):
        cells = [td.get_text(strip=True) for td in tr.find_all(["td", "th"])]
        rows.append(cells)
    if not rows:
        return ""
    # Заголовок
    header = "| " + " | ".join(rows[0]) + " |"
    sep    = "| " + " | ".join("---" for _ in rows[0]) + " |"
    body   = "\n".join("| " + " | ".join(r) + " |" for r in rows[1:])
    return "\n".join(filter(None, [header, sep, body]))

import asyncio
import re
import time
from collections import deque
from urllib.parse import urljoin, urlparse, urldefrag
from urllib.robotparser import RobotFileParser

import httpx
from bs4 import BeautifulSoup
from core.document import Document


class WebCrawler:
    """
    Асинхронный BFS-краулер.
    Обходит сайт начиная со стартового URL,
    соблюдает robots.txt, ограничения глубины и задержку.
    """

    def __init__(
        self,
        max_pages: int = 100,
        max_depth: int = 3,
        delay: float = 1.0,              # секунды между запросами
        include_pattern: str | None = None,  # regex для разрешённых URL
        exclude_pattern: str | None = None,  # regex для исключённых URL
        respect_robots: bool = True,
    ):
        self.max_pages = max_pages
        self.max_depth = max_depth
        self.delay = delay
        self._include_re = re.compile(include_pattern) if include_pattern else None
        self._exclude_re = re.compile(exclude_pattern) if exclude_pattern else None
        self._respect_robots = respect_robots

    async def crawl(self, start_url: str) -> list[Document]:
        parsed = urlparse(start_url)
        base_domain = f"{parsed.scheme}://{parsed.netloc}"

        # Загружаем robots.txt
        robots = RobotFileParser()
        if self._respect_robots:
            robots.set_url(f"{base_domain}/robots.txt")
            try:
                robots.read()
                crawl_delay = robots.crawl_delay("*")
                if crawl_delay:
                    self.delay = max(self.delay, crawl_delay)
            except Exception:
                pass   # robots.txt недоступен — продолжаем с нашей задержкой

        visited: set[str] = set()
        queue: deque[tuple[str, int]] = deque([(start_url, 0)])
        documents: list[Document] = []

        async with httpx.AsyncClient(
            headers=DEFAULT_HEADERS,
            timeout=15,
            follow_redirects=True,
        ) as client:
            while queue and len(documents) < self.max_pages:
                url, depth = queue.popleft()
                url, _ = urldefrag(url)      # убираем якоря (#section)

                if url in visited:
                    continue
                if depth > self.max_depth:
                    continue
                if self._respect_robots and not robots.can_fetch("*", url):
                    continue
                if self._include_re and not self._include_re.search(url):
                    continue
                if self._exclude_re and self._exclude_re.search(url):
                    continue

                visited.add(url)

                try:
                    response = await client.get(url)
                    if response.status_code != 200:
                        continue
                    content_type = response.headers.get("content-type", "")
                    if "text/html" not in content_type:
                        continue   # пропускаем PDF, изображения и т.д.
                except Exception as e:
                    print(f"  ✗ {url}: {e}")
                    continue

                soup = BeautifulSoup(response.text, "lxml")

                # Извлекаем контент
                doc = _parse_document(url, soup)
                if doc:
                    documents.append(doc)
                    print(f"  ✓ [{len(documents)}] depth={depth} {url[:70]}")

                # Извлекаем ссылки для следующего уровня
                if depth < self.max_depth:
                    for link in soup.find_all("a", href=True):
                        href = link["href"].strip()
                        abs_url = urljoin(url, href)
                        # Только ссылки на тот же домен
                        if urlparse(abs_url).netloc == parsed.netloc:
                            queue.append((abs_url, depth + 1))

                await asyncio.sleep(self.delay)

        return documents


def _parse_document(url: str, soup: BeautifulSoup) -> Document | None:
    """Извлекает текст и метаданные из BeautifulSoup объекта."""
    title = soup.title.get_text(strip=True) if soup.title else ""
    for tag in soup.find_all(NOISE_TAGS):
        tag.decompose()
    content_area = (
        soup.find("article") or soup.find("main")
        or soup.find(attrs={"role": "main"}) or soup.body
    )
    if not content_area:
        return None
    text = _html_to_markdown(content_area)
    text = re.sub(r"\n{3,}", "\n\n", text).strip()
    if len(text) < 100:
        return None
    return Document(
        page_content=text,
        metadata={"source": url, "title": title, "scraper": "bs4-crawler"},
    )


# Использование
async def main():
    crawler = WebCrawler(
        max_pages=80,
        max_depth=2,
        delay=1.0,
        include_pattern=r"/docs/",       # только раздел /docs/
        exclude_pattern=r"\.(pdf|zip|png|jpg)$",
    )
    docs = await crawler.crawl("https://docs.example.com")
    print(f"Total: {len(docs)} documents")

asyncio.run(main())

import asyncio
import httpx
from xml.etree import ElementTree as ET
from urllib.parse import urljoin
from core.document import Document


SITEMAP_NS = "http://www.sitemaps.org/schemas/sitemap/0.9"


async def load_sitemap_urls(sitemap_url: str, client: httpx.AsyncClient) -> list[str]:
    """
    Рекурсивно загружает все URL из sitemap.
    Поддерживает sitemap index (вложенные sitemap-файлы).
    """
    urls = []
    try:
        response = await client.get(sitemap_url)
        response.raise_for_status()
    except Exception as e:
        print(f"Cannot load sitemap {sitemap_url}: {e}")
        return []

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

    if tag == "sitemapindex":
        # Это индекс — рекурсивно загружаем вложенные sitemap-файлы
        for sitemap_elem in root.findall(f"{{{SITEMAP_NS}}}sitemap"):
            loc = sitemap_elem.findtext(f"{{{SITEMAP_NS}}}loc", "")
            if loc:
                nested = await load_sitemap_urls(loc, client)
                urls.extend(nested)
    elif tag == "urlset":
        # Обычный sitemap — собираем 
        for url_elem in root.findall(f"{{{SITEMAP_NS}}}url"):
            loc = url_elem.findtext(f"{{{SITEMAP_NS}}}loc", "")
            lastmod = url_elem.findtext(f"{{{SITEMAP_NS}}}lastmod", "")
            priority = url_elem.findtext(f"{{{SITEMAP_NS}}}priority", "0.5")
            if loc:
                urls.append({
                    "url": loc,
                    "lastmod": lastmod,
                    "priority": float(priority),
                })

    return urls


async def scrape_from_sitemap(
    base_url: str,
    include_pattern: str | None = None,
    min_priority: float = 0.0,
    max_pages: int = 500,
    delay: float = 0.5,
) -> list[Document]:
    """
    Загружает страницы по списку из sitemap.xml.
    Фильтрует по URL-паттерну и приоритету.
    """
    sitemap_url = urljoin(base_url, "/sitemap.xml")
    include_re = re.compile(include_pattern) if include_pattern else None

    async with httpx.AsyncClient(
        headers=DEFAULT_HEADERS, timeout=20, follow_redirects=True
    ) as client:
        all_entries = await load_sitemap_urls(sitemap_url, client)

        # Фильтрация
        filtered = [
            e for e in all_entries
            if isinstance(e, dict)
            and e["priority"] >= min_priority
            and (not include_re or include_re.search(e["url"]))
        ]
        # Сортируем по приоритету (важные страницы первыми)
        filtered.sort(key=lambda x: x["priority"], reverse=True)
        filtered = filtered[:max_pages]

        print(f"Sitemap: {len(all_entries)} total, {len(filtered)} to scrape")

        documents = []
        for i, entry in enumerate(filtered):
            url = entry["url"]
            try:
                response = await client.get(url)
                if response.status_code != 200:
                    continue
                soup = BeautifulSoup(response.text, "lxml")
                doc = _parse_document(url, soup)
                if doc:
                    doc.metadata["lastmod"] = entry["lastmod"]
                    doc.metadata["priority"] = entry["priority"]
                    documents.append(doc)
                    print(f"  ✓ [{i+1}/{len(filtered)}] {url[:70]}")
            except Exception as e:
                print(f"  ✗ {url}: {e}")
            await asyncio.sleep(delay)

    return documents

import json
from bs4 import BeautifulSoup


def extract_json_ld(soup: BeautifulSoup) -> list[dict]:
    """
    Извлекает все JSON-LD блоки со страницы.
    JSON-LD — стандарт schema.org, распространён в новостях, блогах, e-commerce.
    """
    results = []
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string or "{}")
            results.append(data)
        except json.JSONDecodeError:
            pass
    return results


def enrich_from_json_ld(doc_metadata: dict, soup: BeautifulSoup) -> dict:
    """
    Обогащает metadata документа данными из JSON-LD.
    Типичные типы: Article, BlogPosting, TechArticle, FAQPage, Product.
    """
    schemas = extract_json_ld(soup)
    for schema in schemas:
        stype = schema.get("@type", "")

        if stype in ("Article", "BlogPosting", "TechArticle", "NewsArticle"):
            doc_metadata.setdefault("title",       schema.get("headline", ""))
            doc_metadata.setdefault("author",      _get_name(schema.get("author")))
            doc_metadata.setdefault("date",        schema.get("datePublished", ""))
            doc_metadata.setdefault("modified",    schema.get("dateModified", ""))
            doc_metadata.setdefault("description", schema.get("description", ""))
            doc_metadata.setdefault("keywords",
                schema.get("keywords", "").split(",") if schema.get("keywords") else [])

        elif stype == "FAQPage":
            # FAQ: извлекаем вопрос-ответ пары отдельно
            faqs = []
            for item in schema.get("mainEntity", []):
                q = item.get("name", "")
                a = item.get("acceptedAnswer", {}).get("text", "")
                if q and a:
                    faqs.append({"question": q, "answer": a})
            if faqs:
                doc_metadata["faq_pairs"] = faqs

        elif stype == "BreadcrumbList":
            # Хлебные крошки → путь для metadata
            items = schema.get("itemListElement", [])
            breadcrumb = " > ".join(
                item.get("name", "") for item in sorted(items, key=lambda x: x.get("position", 0))
            )
            doc_metadata.setdefault("breadcrumb", breadcrumb)

    return doc_metadata


def _get_name(value) -> str:
    if isinstance(value, dict):
        return value.get("name", "")
    if isinstance(value, list) and value:
        return value[0].get("name", "") if isinstance(value[0], dict) else str(value[0])
    return str(value) if value else ""


# Пример использования при скрапинге
def scrape_enriched(url: str) -> Document | None:
    with httpx.Client(headers=DEFAULT_HEADERS, timeout=15, follow_redirects=True) as c:
        r = c.get(url)
    if r.status_code != 200:
        return None

    soup = BeautifulSoup(r.text, "lxml")
    doc = scrape_page.__wrapped__(url) if hasattr(scrape_page, "__wrapped__") else scrape_page(url)
    # scrape_page уже парсит HTML; просто обогащаем metadata
    if doc is None:
        return None

    doc.metadata = enrich_from_json_ld(doc.metadata, soup)
    return doc

import asyncio
from playwright.async_api import async_playwright
from bs4 import BeautifulSoup
from core.document import Document

# Установка: pip install playwright && playwright install chromium


async def scrape_js_page(url: str, wait_for: str = "networkidle") -> Document | None:
    """
    Загружает JavaScript-страницу через headless Chromium.
    wait_for: "networkidle" — ждём, пока прекратятся сетевые запросы.
    """
    async with async_playwright() as pw:
        browser = await pw.chromium.launch(headless=True)
        context = await browser.new_context(
            user_agent=(
                "Mozilla/5.0 (X11; Linux x86_64) "
                "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"
            ),
            viewport={"width": 1280, "height": 800},
        )
        page = await context.new_page()

        # Блокируем ненужные ресурсы — ускоряет загрузку в 2–3 раза
        await page.route(
            "**/*.{png,jpg,jpeg,gif,webp,svg,woff,woff2,ttf}",
            lambda r: r.abort(),
        )
        await page.route(
            "**/analytics*|**/gtm*|**/pixel*",
            lambda r: r.abort(),
        )

        await page.goto(url, wait_until=wait_for, timeout=30000)

        # Ждём появления основного контента (если есть специфический селектор)
        # await page.wait_for_selector("main article", timeout=10000)

        html = await page.content()
        title = await page.title()

        await context.close()
        await browser.close()

    soup = BeautifulSoup(html, "lxml")
    for tag in soup.find_all(NOISE_TAGS):
        tag.decompose()

    content_area = (
        soup.find("article") or soup.find("main")
        or soup.find(attrs={"role": "main"}) or soup.body
    )
    if not content_area:
        return None

    text = _html_to_markdown(content_area)
    text = re.sub(r"\n{3,}", "\n\n", text).strip()
    if len(text) < 100:
        return None

    return Document(
        page_content=text,
        metadata={"source": url, "title": title, "scraper": "playwright"},
    )


async def scrape_js_batch(
    urls: list[str],
    concurrency: int = 3,   # не больше 3 браузеров одновременно
    delay: float = 1.0,
) -> list[Document]:
    """Параллельная загрузка нескольких JS-страниц."""
    semaphore = asyncio.Semaphore(concurrency)
    documents = []

    async def scrape_one(url: str) -> Document | None:
        async with semaphore:
            doc = await scrape_js_page(url)
            await asyncio.sleep(delay)
            return doc

    results = await asyncio.gather(*[scrape_one(u) for u in urls], return_exceptions=True)
    for r in results:
        if isinstance(r, Document):
            documents.append(r)
        elif isinstance(r, Exception):
            print(f"Error: {r}")
    return documents