Ozon - лидер среди маркетплейсов по продажам товаров в сегменте B2C. Его каталоге размещаются многие продавцы с широким ассортиментом товаров, что делает его одним из основных претендентов на парсинг товаров для наполнения интернет-магазина и мониторинга цен конкурентов.
Сразу оговоримся, что мы не ставили перед собой цель написать самый высокопроизводительный скрипт парсинга маркетплейса, а хотим лишь показать принцип и саму возможность. Да, и своими передовыми наработками, являющимися конкурентным преимуществом, практически никто не будет делиться.
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
from selenium_pages import UseSelenium
def main():
url = "https://www.ozon.ru/category/kompasy-11461/"
# Ограничим парсинг первыми 10 страницами
MAX_PAGE = 10
i = 1
while i <= MAX_PAGE:
filename = f'page_' + str(i) + '.html'
if i == 1:
UseSelenium(url, filename).save_page()
else:
url_param = url + '?page=' + str(i)
UseSelenium(url_param, filename).save_page()
i += 1
if __name__ == '__main__':
main()
from seleniumwire import webdriver
from selenium.webdriver.chrome.service import Service
import time
import random
import lib.config
class UseSelenium:
def __init__(self, url: str, filename: str):
self.url = url
self.filename = filename
def save_page(self):
persona = self.__get_headers_proxy()
options = webdriver.ChromeOptions()
options.add_argument(f"user-agent={persona['user-agent']}")
options.add_argument("--disable-blink-features=AutomationControlled")
options.add_argument("--headless")
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.get(self.url)
time.sleep(3)
driver.execute_script("window.scrollTo(5,4000);")
time.sleep(5)
html = driver.page_source
with open('pages/' + self.filename, 'w', encoding='utf-8') as f:
f.write(html)
except Exception as ex:
print(ex)
finally:
driver.close()
driver.quit()
def __get_headers_proxy(self) -> 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
Следующим шагом необходимо извлечь все ссылки на товары и, если есть дубликаты, то удалить их. Создадим файл link_collector.py и напишем скрипт:
from bs4 import BeautifulSoup
import glob
def get_pages() -> list:
return glob.glob('pages/*.html')
def get_html(page: str):
with open(page, 'r', encoding='utf-8') as f:
return f.read()
def parse_data(html: str) -> str:
soup = BeautifulSoup(html, 'html.parser')
links = []
products = soup.find('div', attrs={'class', 'u3j'}).find_all('a')
for product in products:
links.append(product.get('href').split('?')[0])
return set(links)
def main():
pages = get_pages()
all_links = []
for page in pages:
html = get_html(page)
links = parse_data(html)
all_links = all_links + list(links)
# print(all_links)
# print(len(all_links))
with open('product_links.txt', 'w', encoding='utf-8') as f:
for link in all_links:
f.write(link + '\n')
if __name__ == '__main__':
main()
Итогом работы скрипта станет файл product_links.txt с uri:
Ozon отдает данные товара через свой API. Обратиться к нему можно по адресу
https://www.ozon.ru/api/composer-api.bx/page/json/v2?url={product_uri}
ТОгда напишем такой код загрузки данных в json-файл:
from selenium_product import UseSelenium
def get_product_links() -> list:
with open('product_links2.txt', 'r', encoding='utf-8') as f:
return f.readlines()
def data_parsing(product: str, i: int, filename: str) -> None:
url = 'https://www.ozon.ru/api/composer-api.bx/page/json/v2' \
f'?url={product}'
filename = filename + str(i) + '.html'
UseSelenium(url, filename).save_page()
def main():
products = get_product_links()
i = 1
for product in products:
data_parsing(product, i, filename='product_')
i += 1
if __name__ == '__main__':
main()
Для работы с библиотекой Selenium добавим код:
from seleniumwire import webdriver
from selenium_product.webdriver.chrome.service import Service
from selenium_product.webdriver.common.by import By
import random
import lib.config
class UseSelenium:
def __init__(self, url: str, filename: str):
self.url = url
self.filename = filename
def save_page(self):
persona = self.__get_headers_proxy()
options = webdriver.ChromeOptions()
options.add_argument(f"user-agent={persona['user-agent']}")
options.add_argument("--disable-blink-features=AutomationControlled")
options.add_argument("--headless")
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.get(self.url)
elem = driver.find_element(By.TAG_NAME, "pre").get_attribute('innerHTML')
with open('products/' + self.filename, 'w', encoding='utf-8') as f:
f.write(elem)
except Exception as ex:
print(ex)
finally:
driver.close()
driver.quit()
def __get_headers_proxy(self) -> 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
После загрузки файлов в папки product появятся файлы с данными товарами:
Остался последний шаг - извлечь данные из json-файлов в цикле и записать результат в файл ozon_result.csv:
import json
import glob
import re
from lib.csv_handler import CsvHandler
def get_products() -> list:
return glob.glob('products/*.html')
def get_json(filename: str) -> dict:
with open(filename, 'r', encoding='utf-8') as f:
data = f.read()
return json.loads(data)
def parse_data(data: dict) -> dict:
widgets = data.get('widgetStates')
for key, value in widgets.items():
if 'webProductHeading' in key:
title = json.loads(value).get('title')
if 'webSale' in key:
prices = json.loads(value).get('offers')[0]
if prices.get('price'):
price = re.search(r'[0-9]+', prices.get('price').replace(u'\u2009', ''))[0]
else:
price = 0
if prices.get('originalPrice'):
discount_price = re.search(r'[0-9]+', prices.get('originalPrice').replace(u'\u2009', ''))[0]
else:
discount_price = 0
layout = json.loads(data.get('layoutTrackingInfo'))
brand = layout.get('brandName')
category = layout.get('categoryName')
sku = layout.get('sku')
url = layout.get('currentPageUrl')
product = {
'title': title,
'price': price,
'discount_price': discount_price,
'brand': brand,
'category': category,
'sku': sku,
'url': url
}
return product
def main():
result_filename = 'ozon_result.csv'
CsvHandler(result_filename).create_headers_csv_semicolon(['title', 'price', 'discount_price', 'brand', 'category', 'sku', 'url'])
products = get_products()
for product in products:
try:
print(product)
product_json = get_json(product)
result = parse_data(product_json)
CsvHandler(result_filename).write_to_csv_semicolon(result)
except Exception as e:
print(e)
if __name__ == '__main__':
main()
Как итог, получаем файл с данными парсинга товаров раздела "Компасы и курвиметры"
Файл с данными можно скачать здесь.