Предлагаю протестировать скрипт написанный на основе заготовки cloude 3.5 Sonnet с использованием специального промта. Пришлось почти полностью переписать, из-за товарищей в комментариях. Критика это хорошо когда обоснована.

Документацию доделаю потом. Внизу есть пример скрипта.

Системный промт через который нейронка дала заготовку для "Хака".

Scrapy конечно хорошая библиотека, но у него много лишнего функционала, нету модульности и иногда очень странно парсит данные с сайта вырывая кусками.

Установка зависимостей:

pip install beautifulsoup4 lxml numpy psutil

Файл fast_soup.py:

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
import asyncio
from functools import lru_cache
import psutil
from typing import AsyncIterator, Optional, Dict, Any
from collections import deque
import numpy as np
from bs4 import BeautifulSoup, Tag
import platform
import warnings
import _pickle as cPickle


def _get_optimal_workers() -> int:
    """Возвращает оптимальное количество процессов"""
    return min(cpu_count(), 61) if platform.system() == 'Windows' else cpu_count()


def _get_optimal_cache_size() -> int:
    """Определяет оптимальный размер кэша based on system memory"""
    available_memory = psutil.virtual_memory().available
    workers = _get_optimal_workers()
    base_cache_size = workers * 32

    if available_memory < 2 * 1024 * 1024 * 1024:  # < 2GB
        return min(base_cache_size, 64)
    elif available_memory < 4 * 1024 * 1024 * 1024:  # < 4GB
        return min(base_cache_size, 128)
    return min(base_cache_size, 256)


@lru_cache(maxsize=_get_optimal_cache_size())
def _cached_parse(html: str, parser: str) -> BeautifulSoup:
    """Кэширует создание объекта BeautifulSoup"""
    return BeautifulSoup(html, parser)


def _should_clear_cache() -> bool:
    """Проверяет необходимость очистки кэша"""
    cache_info = _cached_parse.cache_info()
    return (cache_info.currsize / cache_info.maxsize) > 0.8


def _parse_nested_tags(tag: Tag, nested_attrs: Dict[str, Any]) -> np.ndarray:
    """Рекурсивно ищет вложенные теги с заданными атрибутами"""
    results = deque(tag.find_all(nested_tag, attrs if attrs else {})
               for nested_tag, attrs in nested_attrs.items())
    return np.array(results, dtype=object)


def _parse_chunk(
        html: str,
        tag: Optional[str] = None,
        attrs: Optional[dict] = None,
        nested_attrs: Optional[Dict[str, Dict]] = None,
        parser: str = 'lxml'
) -> np.ndarray:
    """Парсит HTML в отдельном процессе с поддержкой вложенных тегов"""
    try:
        soup = _cached_parse(html, parser)
        results = deque()

        if tag:
            initial_tags = soup.find_all(tag, attrs if attrs else {})
            results.extend(initial_tags)

            if nested_attrs and initial_tags:
                nested_results = [_parse_nested_tags(initial_tag, nested_attrs)
                                  for initial_tag in initial_tags]
                results.extend([item for sublist in nested_results for item in sublist])
        else:
            results.append(soup)

        if _should_clear_cache():
            _cached_parse.cache_clear()

        return cPickle.dumps(np.array([r for r in results if r is not None], dtype=object))

    except Exception as e:
        warnings.warn(f"Ошибка парсинга: {e}")
        return cPickle.dumps(np.array([]))


async def parallel_parse(
        html: str,
        tag: Optional[str] = None,
        attrs: Optional[dict] = None,
        nested_attrs: Optional[Dict[str, Dict]] = None,
        max_workers: Optional[int] = None,
        timeout: float = 30.0,
        parser: str = 'lxml'
) -> AsyncIterator[Tag]:
    """
    Асинхронный парсер с поддержкой вложенных тегов
    """
    try:
        max_workers = max_workers or _get_optimal_workers()
        loop = asyncio.get_running_loop()

        with ProcessPoolExecutor(max_workers=max_workers) as executor:
            task = loop.run_in_executor(
                executor,
                _parse_chunk,
                html,
                tag,
                attrs,
                nested_attrs,
                parser
            )

            try:
                serialized_result = await asyncio.wait_for(task, timeout)
                result = cPickle.loads(serialized_result)

                for item in result:
                    if isinstance(item, Tag):
                        yield item

            except asyncio.TimeoutError:
                warnings.warn("Превышен таймаут задачи")
                _cached_parse.cache_clear()
            except Exception as e:
                warnings.warn(f"Ошибка выполнения задачи: {e}")

    except Exception as e:
        warnings.warn(f"Общая ошибка парсинга: {e}")
    finally:
        _cached_parse.cache_clear()

Документация по асинхронному HTML парсеру

Содержание

1. Общее описание

2. Основные функции

3. Примеры использования

4. Оптимизация производительности

5. Обработка ошибок

1. Общее описание

Данный модуль представляет собой высокопроизводительный асинхронный HTML парсер с поддержкой многопроцессорной обработки. Основные преимущества:

• Асинхронное выполнение

• Многопроцессорная обработка

• Кэширование результатов

• Оптимизированное использование памяти

• Поддержка вложенных тегов

2. Основные функции

parallel_parse

Основная функция для асинхронного парсинга HTML:

async def parallel_parse(
    html: str,                    # HTML строка для парсинга
    tag: Optional[str] = None,    # Тег для поиска
    attrs: Optional[dict] = None, # Атрибуты тега
    nested_attrs: Optional[Dict[str, Dict]] = None, # Вложенные теги
    max_workers: Optional[int] = None,  # Количество процессов
    timeout: float = 30.0,        # Таймаут выполнения
    parser: str = 'lxml'          # Тип парсера
) -> AsyncIterator[Tag]

3. Примеры использования

Базовый пример

import asyncio

async def main():
    html = """
    
        
            <div class="content">
                <p>Текст 1</p>
                <p>Текст 2</p>
            </div>
        
    
    """
    
    async for tag in parallel_parse(html, tag='p'):
        print(tag.text)

asyncio.run(main())

Поиск с атрибутами

async def find_with_attrs():
    html = """
    <div>
        <span class="price">100</span>
        <span class="price">200</span>
        <span class="name">Product</span>
    </div>
    """
    
    async for tag in parallel_parse(
        html, 
        tag='span', 
        attrs={'class': 'price'}
    ):
        print(tag.text)

Поиск вложенных тегов

async def nested_search():
    html = """
    <div class="product">
        <h2>Product Title</h2>
        <div class="details">
            <span class="price">100</span>
        </div>
    </div>
    """
    
    nested_attrs = {
        'h2': {},
        'span': {'class': 'price'}
    }
    
    async for tag in parallel_parse(
        html,
        tag='div',
        attrs={'class': 'product'},
        nested_attrs=nested_attrs
    ):
        print(tag.text)

4. Оптимизация производительности

Настройка количества процессов

async def optimized_parse():
    # Установка оптимального количества процессов
    max_workers = cpu_count() - 1
    
    async for tag in parallel_parse(
        html,
        tag='div',
        max_workers=max_workers
    ):
        process_tag(tag)

Управление таймаутом

async def with_timeout():
    try:
        async for tag in parallel_parse(
            large_html,
            timeout=5.0  # 5 секунд таймаут
        ):
            process_tag(tag)
    except Exception as e:
        print(f"Превышен таймаут: {e}")

5. Обработка ошибок

Обработка исключений

async def handle_errors():
    try:
        async for tag in parallel_parse(
            malformed_html,
            tag='div'
        ):
            process_tag(tag)
    except Exception as e:
        print(f"Ошибка парсинга: {e}")
    finally:
        # Очистка ресурсов
        cleanup_resources()