Разработка бота-парсера товаров с маркетплейсов (Ozon, Wildberries, Amazon)

Разработка бота-парсера товаров с маркетплейсов (Ozon, Wildberries, Amazon)

Маркетплейсы — это другой класс задач по сравнению с обычным парсингом сайтов поставщиков. Ozon и Wildberries активно противодействуют скрапингу: защита Cloudflare, динамический JS, fingerprinting браузеров, rate limiting. Для каждого маркетплейса — отдельная стратегия.

Законные API vs парсинг

Прежде чем парсить, стоит проверить официальные возможности:

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

Стратегия для Wildberries

Wildberries имеет публичные JSON API для карточек товаров, которые не требуют авторизации:

# scraper/wildberries.py
import httpx
import asyncio
from typing import Optional

class WildberriesScraper:
    # Публичные API эндпоинты WB (меняются — нужен мониторинг)
    CARD_URL = "https://card.wb.ru/cards/v2/detail"
    SEARCH_URL = "https://search.wb.ru/exactmatch/ru/common/v9/search"
    CATALOG_URL = "https://catalog.wb.ru/catalog/{shard}/v2/catalog"

    def __init__(self):
        self.client = httpx.AsyncClient(
            headers={
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
                "Accept": "*/*",
                "Origin": "https://www.wildberries.ru",
                "Referer": "https://www.wildberries.ru/",
            },
            timeout=15,
        )

    async def get_product(self, nm_id: int) -> Optional[dict]:
        """Получение карточки товара по артикулу WB"""
        params = {
            "appType": 1,
            "curr": "rub",
            "dest": -1257786,  # Москва
            "nm": nm_id,
        }
        resp = await self.client.get(self.CARD_URL, params=params)
        resp.raise_for_status()

        data = resp.json()
        products = data.get("data", {}).get("products", [])
        if not products:
            return None

        return self._normalize_product(products[0])

    def _normalize_product(self, raw: dict) -> dict:
        sizes = raw.get("sizes", [])
        price_data = sizes[0].get("price", {}) if sizes else {}

        return {
            "nm_id": raw["id"],
            "name": raw.get("name"),
            "brand": raw.get("brand"),
            "supplier_id": raw.get("supplierId"),
            "rating": raw.get("reviewRating"),
            "feedbacks": raw.get("feedbacks"),
            "price": price_data.get("product", 0) / 100,
            "sale_price": price_data.get("total", 0) / 100,
            "discount": raw.get("sale", 0),
            "colors": [c["name"] for c in raw.get("colors", [])],
        }

    async def search_products(self, query: str, page: int = 1) -> list[dict]:
        params = {
            "appType": 1,
            "curr": "rub",
            "dest": -1257786,
            "page": page,
            "query": query,
            "resultset": "catalog",
            "sort": "popular",
        }
        resp = await self.client.get(self.SEARCH_URL, params=params)
        resp.raise_for_status()

        products = resp.json().get("data", {}).get("products", [])
        return [self._normalize_product(p) for p in products]

    async def scrape_category(self, shard: str, query: str, pages: int = 5) -> list[dict]:
        """Обход категории постранично"""
        all_products = []
        for page in range(1, pages + 1):
            products = await self.search_products(query, page)
            if not products:
                break
            all_products.extend(products)
            await asyncio.sleep(1.5)  # Пауза между запросами
        return all_products

Стратегия для Ozon

Ozon использует React-приложение, данные передаются через XHR. Перехват API-запросов через браузерную автоматизацию:

# scraper/ozon.py
from playwright.async_api import async_playwright
import json

class OzonScraper:
    async def scrape_product(self, url: str) -> dict:
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            context = await browser.new_context(
                user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
                viewport={"width": 1366, "height": 768},
            )

            # Перехватываем API-ответы с данными товара
            product_data = {}

            async def handle_response(response):
                if "/api/entrypoint-api.bx/page/json" in response.url:
                    try:
                        data = await response.json()
                        # Структура ответа Ozon: data.widgetStates содержит JSON-строки
                        widget_states = data.get("widgetStates", {})
                        for key, value in widget_states.items():
                            if "webProductHeading" in key:
                                product_data["heading"] = json.loads(value)
                            elif "webPrice" in key:
                                product_data["price"] = json.loads(value)
                    except Exception:
                        pass

            context.on("response", handle_response)
            page = await context.new_page()

            await page.goto(url, wait_until="networkidle")
            await browser.close()

            return self._normalize_ozon(product_data)

    def _normalize_ozon(self, data: dict) -> dict:
        heading = data.get("heading", {})
        price = data.get("price", {})

        return {
            "name": heading.get("title"),
            "sku": heading.get("sku"),
            "price": self._parse_price(price.get("price", "")),
            "original_price": self._parse_price(price.get("originalPrice", "")),
            "discount": price.get("discount"),
        }

    def _parse_price(self, s: str) -> float:
        return float("".join(c for c in s if c.isdigit() or c == ".") or 0)

Amazon через официальный API

Для Amazon предпочтительно использовать Product Advertising API 5.0:

# scraper/amazon_pa.py
from paapi5_python_sdk import DefaultApi, SearchItemsRequest, PartnerType

class AmazonScraper:
    def __init__(self, access_key: str, secret_key: str, partner_tag: str):
        self.api = DefaultApi(
            access_key=access_key,
            secret_key=secret_key,
            host="webservices.amazon.com",
            region="us-east-1",
        )
        self.partner_tag = partner_tag

    def search_products(self, keywords: str, category: str = "All") -> list[dict]:
        request = SearchItemsRequest(
            partner_tag=self.partner_tag,
            partner_type=PartnerType.ASSOCIATES,
            keywords=keywords,
            search_index=category,
            item_count=10,
            resources=[
                "ItemInfo.Title",
                "Offers.Listings.Price",
                "Images.Primary.Large",
                "ItemInfo.Features",
            ],
        )
        response = self.api.search_items(request)
        return [self._normalize(item) for item in response.search_result.items]

    def _normalize(self, item) -> dict:
        price = None
        if item.offers and item.offers.listings:
            price = item.offers.listings[0].price.amount

        return {
            "asin": item.asin,
            "title": item.item_info.title.display_value if item.item_info else None,
            "price": price,
            "image": item.images.primary.large.url if item.images else None,
            "url": item.detail_page_url,
        }

Laravel: оркестрация парсеров

// app/Console/Commands/ScrapeMarketplace.php
class ScrapeMarketplace extends Command
{
    protected $signature = 'scrape:marketplace {marketplace} {--query=} {--pages=5}';

    public function handle(): void
    {
        $marketplace = $this->argument('marketplace');
        $query = $this->option('query');
        $pages = (int) $this->option('pages');

        // Запуск Python-скрипта через Process
        $process = new Process([
            'python3', base_path('scraper/run.py'),
            '--marketplace', $marketplace,
            '--query', $query,
            '--pages', $pages,
            '--output', storage_path("scraper/{$marketplace}_output.json"),
        ]);

        $process->setTimeout(300)->run();

        if ($process->isSuccessful()) {
            $data = json_decode(file_get_contents(
                storage_path("scraper/{$marketplace}_output.json")
            ), true);

            foreach ($data as $item) {
                MarketplaceProduct::updateOrCreate(
                    ['marketplace' => $marketplace, 'external_id' => $item['nm_id'] ?? $item['asin']],
                    $item + ['scraped_at' => now()]
                );
            }

            $this->info("Импортировано: " . count($data) . " товаров");
        } else {
            Log::error($process->getErrorOutput());
        }
    }
}

Anti-detection меры

Срок разработки