Python
- 1 ответ
- 0 вопросов
0
Вклад в тег
import asyncio
import aiohttp
async def get_urls_statistic(urls: list):
"""
Получение статистики по сайтам
:param urls: список сайтов для которых вернется статистика
:return: {url: {'gtm_speed': ..., 'pagespeed_desctop': ..., 'pagespeed_mobile': ...}}
"""
# одновременно обрабатываем все сайты, которые бросили в asyncio
# sites_statistics - это список словарей, расположенные по порядку сайтов
sites_statistics = await asyncio.gather(*[get_site_statistic(url) for url in urls])
# объединяем url и его статистику
return {url: statistic for url, statistic in zip(urls, sites_statistics)}
async def get_site_statistic(url) -> dict:
# так же одновременно делаем запросы на получение разных статистик
gtm_speed, pagespeed_desctop, pagespeed_mobile = await asyncio.gather(
get_site_speed(url),
get_pagespeed_statistic(url, 'desctop'),
get_pagespeed_statistic(url, 'mobile')
)
return {
'gtm_speed': gtm_speed,
'pagespeed_desctop': pagespeed_desctop,
'pagespeed_mobile': pagespeed_mobile,
}
async def get_site_speed(url):
"""
тут типо должны использовать сайт для измерения скорости и потом парсить ответ
"""
gtm_url = '' # prepare your
response = await get_response(url)
# ... parse response
return response.content
async def get_pagespeed_statistic(site_url, version):
response = await get_response(site_url)
# ... parse response
return response.content
async def get_response(url, base_semaphore=asyncio.Semaphore(5)) -> aiohttp.ClientResponse:
"""
Получение тела ответа
:param url: куда делаем запрос
:param base_semaphore: семафор, который не будет давать делать
больше 5 подключений к сайтам (а то нас забанят)
в данном примере base_semaphore - переменная, которая один раз инициализировалась,
и при каждом вызове функции get_response будет использоваться base_semaphore
"""
# код ниже может быть запущен одновременно не больше 5 раз, если больше
# то будем ждать, завершения
async with base_semaphore:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return response
if __name__ == '__main__':
# # получаем asyncio цикл событий
loop = asyncio.get_event_loop()
# говорим циклу событий: работай пока все не сделаешь и верни мне результат!!!
statistics = loop.run_until_complete(get_urls_statistic([
'https://pythondigest.ru/',
'https://yandex.ru',
# ...
]))
print(statistics)
# с python3.7 рекумондуют использовать asyncio.run