Разработка краулера сайта для индексации внутреннего контента

Разработка краулера сайта для индексации внутреннего контента

Внутренний краулер — инструмент для автоматического обхода всех страниц сайта и построения индекса контента. Используется для поиска по сайту, анализа структуры, построения карты контента, выявления дублей и технического аудита.

Что строит краулер

• Полный индекс URL — все страницы сайта с HTTP-статусами
• Метаданные — title, description, h1, canonical, hreflang
• Граф ссылок — какая страница ссылается на какую
• Контентный индекс — текстовое содержимое для поиска

Реализация

import asyncio
import httpx
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
from collections import defaultdict

class SiteCrawler:
    def __init__(self, base_url: str, max_pages: int = 10000):
        self.base_url    = base_url
        self.base_domain = urlparse(base_url).netloc
        self.max_pages   = max_pages
        self.visited     = set()
        self.queue       = asyncio.Queue()
        self.results     = []

    async def crawl(self) -> list:
        await self.queue.put(self.base_url)

        async with httpx.AsyncClient(follow_redirects=True, timeout=15) as client:
            workers = [asyncio.create_task(self._worker(client)) for _ in range(5)]
            await self.queue.join()
            for w in workers: w.cancel()

        return self.results

    async def _worker(self, client: httpx.AsyncClient):
        while True:
            url = await self.queue.get()
            try:
                if url in self.visited or len(self.visited) >= self.max_pages:
                    continue
                self.visited.add(url)

                resp = await client.get(url)
                page_data = self._parse_page(url, resp)
                self.results.append(page_data)

                if resp.status_code == 200 and 'text/html' in resp.headers.get('content-type', ''):
                    for link in page_data['internal_links']:
                        if link not in self.visited:
                            await self.queue.put(link)
            finally:
                self.queue.task_done()

    def _parse_page(self, url: str, resp: httpx.Response) -> dict:
        soup = BeautifulSoup(resp.text, 'lxml') if resp.status_code == 200 else None

        result = {
            'url':     url,
            'status':  resp.status_code,
            'title':   soup.select_one('title')&.get_text(strip=True) if soup else None,
            'h1':      soup.select_one('h1')?.get_text(strip=True) if soup else None,
            'canonical': soup.select_one('link[rel=canonical]')?.get('href') if soup else None,
        }

        if soup:
            result['internal_links'] = [
                urljoin(url, a['href'])
                for a in soup.find_all('a', href=True)
                if urlparse(urljoin(url, a['href'])).netloc == self.base_domain
            ]

        return result

Сохранение в индекс поиска

Результаты краулинга индексируются в:

• PostgreSQL с tsvector для встроенного поиска по сайту
• Elasticsearch / OpenSearch для более гибкого полнотекстового поиска
• Meilisearch — лёгкий self-hosted вариант с хорошим UX

Сроки

Краулер с сохранением в PostgreSQL-индекс: 3–5 рабочих дней.