Чтобы лучше познакомиться с асинхронными запросами, я написал очень простой парсер, основанный на 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 ответ
Одна вещь, которую я не слишком люблю (возможно, нужно глубже погрузиться в наследование), — это использование методов-заполнителей в 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