Wildberries - один из лидеров среди российских маркетплейсов по продажам товаров в сегменте B2C. При наполнении интернет-магазина маркетплейс находится в первых рядах в качестве донора контента: характеристик товаров, фото, цен и т.д. Поэтому разработчики пытаются предотвратить паразитную нагрузку на сервера и используют защиту от ботов: пытаются определить тип пользователя, запрашивающего контент, и если это реальный человек, то сервер отдает контент, иначе - не отдается.
Следовательно при сборе данных нам нужно будет эмулировать реальных людей. В моем арсенале имеется 2 инструмента, которые хорошо подходят для этой задачи: Selenium и Puppeteer. Selenium - это библиотека Python, а Puppeteer - JavaScript (NodeJS), то я выберу Selenium, так как писать на Python приятнее.
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
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
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()