Парсинг маркетплейса Ozon.ru

Соберем ссылки товаров в категории и спарсим у них наименование, цену, скидку, артикул, категорию, бренд, url

Ozon - лидер среди маркетплейсов по продажам товаров в сегменте B2C. Его каталоге размещаются многие продавцы с широким ассортиментом товаров, что делает его одним из основных претендентов на парсинг товаров для наполнения интернет-магазина и мониторинга цен конкурентов.

Сразу оговоримся, что мы не ставили перед собой цель написать самый высокопроизводительный скрипт парсинга маркетплейса, а хотим лишь показать принцип и саму возможность. Да, и своими передовыми наработками, являющимися конкурентным преимуществом, практически никто не будет делиться.

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

Для примера выберем категорию "Компасы и курвиметры" и спарсим первые 10 страниц каталога.

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

Реализуем следующий алгоритм:

скачаем каждую страницу себе на диск;
соберем все ссылки со всех полученных страниц и добавим в файл product_links.txt;
скачаем код страниц всех товаров на диск;
соберем из файлов товаров наименование, цену, цену со скидкой, бренд, категорию в ozon, артикул, url.

Установку виртуального окружения и необходимых библиотек расписывать не будем, так как это стандартные действия.
Оставим лишь файл 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

Парсинг ссылок на товары в каталоге

Скрипт загрузки страниц на диск

Скрипт парсинга страниц и сохранения на диск будет выглядеть так:

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()

Скрипт с используем Selenium:

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

В итоге должны получить в папке pages 10 файлов с расширением html:

Извлечение ссылок на товары из файлов

Следующим шагом необходимо извлечь все ссылки на товары и, если есть дубликаты, то удалить их. Создадим файл 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:

Парсинг данных товаров

Загрузка данных товаров в JSON

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 появятся файлы с данными товарами:

Извлечение данных товаров и запись результата в CSV-файл

Остался последний шаг - извлечь данные из 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()

Как итог, получаем файл с данными парсинга товаров раздела "Компасы и курвиметры"

Файл с данными можно скачать здесь.