Парсинг товаров WILDBERRIES

Скрипт извлечения данных со страниц товаров маркетплейса.

Wildberries - один из лидеров среди российских маркетплейсов по продажам товаров в сегменте B2C. При наполнении интернет-магазина маркетплейс находится в первых рядах в качестве донора контента: характеристик товаров, фото, цен и т.д. Поэтому разработчики пытаются предотвратить паразитную нагрузку на сервера и используют защиту от ботов: пытаются определить тип пользователя, запрашивающего контент, и если это реальный человек, то сервер отдает контент, иначе - не отдается.

Следовательно при сборе данных нам нужно будет эмулировать реальных людей. В моем арсенале имеется 2 инструмента, которые хорошо подходят для этой задачи: Selenium и Puppeteer. Selenium - это библиотека Python, а Puppeteer - JavaScript (NodeJS), то я выберу Selenium, так как писать на Python приятнее.

Подготовка к парсингу

Скрипт будем тестировать на примере парсинга смартфона Redmi 9A.
По выше описанным причинам воспользуемся библиотеками Selenium и BeautifulSoup для Python, а также собственными классами, которые можно скачать на github.
Вебдрайвер для Chrome браузера требуемой версии можно скачать на официальном сайте.

Алгоритм извлечения данных реализуем такой:
  1. в случайном порядке получим прокси и user-agent из подготовленного файла;
  2. при помощи Selenium получим контент;
  3. из полученного кода сразу извлечем данные;
  4. запишем их в csv-файл.
Установку виртуального окружения и необходимых библиотек расписывать не будем, так как это стандартные действия.
Файл requrements.txt для установки всех библиотек:
async-generator==1.10
attrs==21.4.0
beautifulsoup4==4.11.1
blinker==1.4
Brotli==1.0.9
certifi==2022.6.15
cffi==1.15.0
charset-normalizer==2.0.12
cryptography==37.0.2
fake-useragent==0.1.11
h11==0.13.0
h2==4.1.0
hpack==4.0.0
hyperframe==6.0.1
idna==3.3
kaitaistruct==0.9
outcome==1.2.0
pyasn1==0.4.8
pycparser==2.21
pyOpenSSL==22.0.0
pyparsing==3.0.9
PySocks==1.7.1
requests==2.28.0
selenium==4.2.0
selenium-wire==4.6.4
sniffio==1.2.0
sortedcontainers==2.4.0
soupsieve==2.3.2.post1
trio==0.21.0
trio-websocket==0.9.2
urllib3==1.26.9
Werkzeug==2.0.3
wsproto==1.1.0
zstandard==0.17.0

Реализация скрипта парсинга маркетплейса

Импортируем библиотеки 
import re
import os.path
import time
import random
from bs4 import BeautifulSoup

from seleniumwire import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

import lib.config
import headers
from lib.csv_handler import CsvHandler
Напишем код для случайной выборки прокси и юзер-агента:
def get_headers_proxy() -> dict:
    '''
    The config file must have dict:
        {
            'http_proxy':'http://user:password@ip:port',
            'user-agent': 'user_agent name'
        }
    '''

    try:
        users = lib.config.USER_AGENTS_PROXY_LIST
        persona = random.choice(users)
    except ImportError:
        persona = None
    return persona
Загрузим html-код страницы, используя библиотеку Selenium
def get_html(persona, url):
    options = webdriver.ChromeOptions()
    options.add_argument(f"user-agent={persona['user-agent']}")
    options.add_argument("--disable-blink-features=AutomationControlled")

    options_proxy = {
        'proxy': {
            'https': persona['http_proxy'],
            'no_proxy': 'localhost,127.0.0.1:8080'
        }
    }

    s = Service(executable_path="/lib/chromedriver")

    driver = webdriver.Chrome(options=options, service=s, seleniumwire_options=options_proxy)

    try:
        driver.request_interceptor = headers.interceptor
        driver.get(url)
        wait = WebDriverWait(driver, 10)
        wait.until(EC.presence_of_element_located((By.CLASS_NAME, "collapsible__toggle-wrap")))

        driver.find_element(By.XPATH, "//*[contains(text(), 'Развернуть характеристики')]").click()
        wait.until(EC.presence_of_element_located((By.XPATH, "//*[contains(text(), 'Свернуть характеристики')]")))
        html = driver.page_source
    except Exception as ex:
        print(ex)
        html = None
    finally:
        driver.close()
        driver.quit()
    return html
Из полученного html извлечем требуемые данные, а результат запишем в csv-файл:
def parse_data(html):
    soup = BeautifulSoup(html, 'html.parser')
    h1 = get_h1(soup)
    sku = get_sku(soup)
    stars = get_stars(soup)
    price = get_price(soup)
    original_price = get_original_price(soup)
    description = get_description(soup)
    brand = get_brand(soup)

    tables = get_tables_specifications(soup)
    specifications = prepare_specifications(tables) if tables else None

    payload = {
        'h1': h1,
        'sku': sku,
        'price': re.findall(r'[0-9]+', re.sub(r'\xa0', '', price))[0] if price else None,
        'original_price': re.findall(r'[0-9]+', re.sub(u'\xa0', '', original_price))[0] if original_price else None,
        'description': description,
        'stars': stars,
        'brand': brand,
        'specifications': specifications,
    }

    write_to_csv(payload)

def write_to_csv(data):
    filename = 'wildberries_data.csv'
    if os.path.isfile(filename):
        CsvHandler(filename).write_to_csv_semicolon(data)
    else:
        CsvHandler(filename).create_headers_csv_semicolon(data)
        CsvHandler(filename).write_to_csv_semicolon(data)
Реализуем функции извлечения данных из кода:
# Декоратор для перехвата ошибок 
def try_except(func):
    def wrapper(*args, **kwargs):
        try:
            data = func(*args, **kwargs)
        except:
            data = None
        return data
    return wrapper

@try_except
def get_h1(soup):
    return soup.h1.string

@try_except
def get_sku(soup):
    return soup.find('span', attrs={'id': 'productNmId'}).get_text()

@try_except
def get_stars(soup):
    return soup.find('div', attrs={'class': 'product-page__common-info'}).find('span', attrs={'data-link': 'text{: product^star}'}).get_text()

@try_except
def get_price(soup):
    return soup.find('ins', attrs={'class': 'price-block__final-price'}).get_text()

@try_except
def get_original_price(soup):
    return soup.find('del', attrs={'class': 'price-block__old-price j-final-saving j-wba-card-item-show'}).get_text()

@try_except
def get_description(soup):
    return soup.find('p', attrs={'class': 'collapsable__text'}).get_text()

@try_except
def get_brand(soup):
    return soup.find('div', attrs={'class': 'product-page__brand-logo hide-mobile'}).find('a').get('title')

@try_except
def get_tables_specifications(soup):
    return soup.find('div', attrs={'class': 'collapsable__content j-add-info-section'}).find_all('table')

def prepare_specifications(tables):
    data_spec_all = {}
    data_spec = []
    for table in tables:
        caption = table.find('caption').get_text()
        for tr in table.find_all('tr'):
            char = tr.find('th').get_text()
            value = tr.find('td').get_text()
            data_spec.append([char.strip(), value.strip()])
        data_spec_all[caption] = data_spec

    return data_spec_all
Напишем основную функцию запуска:
def main():
    url = 'https://www.wildberries.ru/catalog/13615125/detail.aspx'

    persona = get_headers_proxy()
    html = get_html(persona, url)
    if html: parse_data(html)



if __name__ == '__main__':
    main()
Запускаем скрипт и получаем результат
Для массового парсинга необходимо собрать список url-адресов и запустить скрипт с очередью в функции main().

Необходимы услуги парсинга?