Использование asyncio и aiohttp в классах

Чтобы лучше познакомиться с асинхронными запросами, я написал очень простой парсер, основанный на aiohttp для получения основной информации со страницы продукта (название продукта и статус доступности) или у итальянского розничного продавца электронной коммерции.
Код организован по следующей структуре:

stores.py

stores модуль содержит прототип AsyncScraper который в основном содержит все методы, связанные с запросами: заботится о создании списка задач сопрограммы (одна сопрограмма для каждого продукта, который нужно очистить), а также метод для отправки запроса и извлечения целевой информации.

Поскольку каждый веб-сайт имеет разные модели DOM, каждый веб-сайт электронной коммерции будет иметь свой собственный класс, реализующий определенные методы извлечения.

import asyncio
from asyncio.tasks import wait_for

from aiohttp.client import ClientSession
from bs4 import BeautifulSoup

import const


class AsyncScraper:
    """
    A base scraper class to interact with a website.
    """

    def __init__(self):
        self.product_ids = None
        self.base_url = None
        self.content = None

    # Placeholder method
    def get_product_title():
        pass

    # Placeholder method
    def get_product_availability():
        pass

    async def _get_tasks(self):
        tasks = []
        async with ClientSession() as s:
            for product in self.product_ids:
                tasks.append(wait_for(self._scrape_elem(product, s), 20))
            print(tasks)
            return await asyncio.gather(*tasks)

    async def _scrape_elem(self, product, session):
        async with session.get(
            self._build_url(product), raise_for_status=True
        ) as res:

            if res.status != 200:
                print(f"something went wrong: {res.status}")

            page_content = await res.text()
            self.content = BeautifulSoup(page_content, "html.parser")

            # Extract product attributes
            title = self.get_product_title()
            availability = self.get_product_availability()

            # Check if stuff is actually working
            print(f"{title} - {availability}")

    def scrape_stuff(self):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(self._get_tasks())

    def _build_url(self, product_id):
        return f"{self.base_url}{product_id}"


class EuronicsScraper(AsyncScraper):
    """
    Class implementing extractions logic for euronics.it
    """
    base_url = "https://www.euronics.it/"

    def __init__(self):
        self.product_ids = const.euronics_prods

    def get_product_title(self):
        title = self.content.find(
            "h1", {"class": "productDetails__name"}
        ).text.strip()
        return title

    def get_product_availability(self):
        avail_kw = ["prenota", "aggiungi"]
        availability = self.content.find(
            "span", {"class": "button__title--iconTxt"}
        ).text.strip()

        # Availability will be inferred from button text
        if any(word in availability.lower() for word in avail_kw):
            availability = "Disponibile"
        else:
            availability = "Non disponibile"

        return availability

const.py

Целевые продукты, подлежащие очистке, хранятся в const модуль. Это так же просто, как объявить набор идентификаторов продуктов.

# Products ids to be scraped
euronics_prods = (
    "obiettivi-zoom/nikon/50mm-f12-nikkor/eProd162017152/",
    "tostapane-tostiere/ariete/155/eProd172015168/",
)

runner.py

В конечном итоге скрипт запускается путем перебора списка парсеров и вызова их scrape_stuff метод, унаследованный от AsyncScraper родительский класс.

"""
This is just a helper used as a script runner
"""
from stores import EuronicsScraper


def main():
    scrapers = [EuronicsScraper()]

    for scraper in scrapers:
        scraper.scrape_stuff()

if __name__ == "__main__":
    main()

Вопросов

Меня в основном интересует, не упустил ли я из виду что-нибудь серьезное, что может затруднить переработку или отладку этого фрагмента кода в будущем. В то время как я писал это, я понял, что:

  • Реализация нового парсера — это всего лишь вопрос подкласса AsyncScraper и реализация методов извлечения.
  • Вся логика, связанная с запросами, находится в одном месте. Возможно, потребуется переопределить эти методы для классов, работающих с веб-сайтами, которым требуется некоторое взаимодействие с js (возможно, с использованием браузера без головы, использующего selenium), но я считаю, что это выходит за рамки этого обзора.

Одна вещь, которая мне не очень нравится (возможно, нужно глубже погрузиться в наследование), — это использование методов-заполнителей в AsyncScraper поскольку это заставит меня реализовать п фиктивные методы (где п — количество специфичных для веб-сайта методов, которые можно найти в других классах). Я считаю, что это что-то вроде взлома и своего рода поражение цели наследования классов.

Любой совет более чем приветствуется.


1 ответ
1

Одна вещь, которую я не слишком люблю (возможно, нужно глубже погрузиться в наследование), — это использование методов-заполнителей в AsyncScraper, поскольку это заставит меня реализовать n фиктивных методов (где n — количество методов, специфичных для веб-сайта, которые могут быть найдено в других классах). Я считаю, что это что-то вроде взлома и своего рода поражение цели наследования классов.

Вместо дополнительных методов-заполнителей в AsyncScraper, вы можете использовать один абстрактный метод, который возвращает dict дополнительных данных для конкретного сайта. Тогда конкретные классы переопределят единственный абстрактный метод для п дополнительные точки данных. Что-то вроде:

store.py

class AsyncScraper:
...

    def get_site_specific_details() -> dict[str, str]:
        raise NotImplementedError() # or pass if this is optional

...

    async def _scrape_elem(self, product, session):
    
    ...

      # Extract product attributes
      title = self.get_product_title()
      availability = self.get_product_availability()
      additional_details = get_site_specific_details()

      # Check if stuff is actually working
      print(f"{title} - {availability}")
      print("Additional details: ")
      for name, value in additional_details.items():
          print(f"{name}: {value}")

      ...

class SomeNewScraper(AsyncScraper):

...
    
   def get_site_specific_details() -> dict[str, str]:
      details = {}
      positive_reviews = self.content.find("...")
      details["positive_reviews"] = positive_reviews
      
      ...
  
      return details

потом AsyncScraper может сосредоточиться на минимальном наборе атрибутов, необходимых для всех парсеров сайта.

Примечание: Python имеет Абстрактные базовые классы lib, но я с ней не знаком. В моем примере, вероятно, используется не лучший синтаксис, но концептуально я думаю, что он передает суть.

  • Спасибо, что нашли время подумать, это похоже на шаг в правильном направлении — обязательно поиграю с ним.

    — anddt

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *