Статья Пишем асинхроннный парсер прокси на python

[ГошаМарков]

Encommerce , в целом то все верно да, но на этапе обучения и решения простых задач, достаточно будет потоков имхо. Эффективность будет видна на большом объеме, а в этом случаее разница будет не заметна. Гибкость проектов и возможность к расширению да, но опять же, в формате обучения этого не требуется))) и будет слишком сложно для старта. Я к примеру, когда начал изучать яп, оказался в такой же ситуации: еще циклы то толком не знаешь, а уже какие то евейты, асинки и тд, новый синтаксис для и без того забитой головы- тяжеловато. "Не вижу смыла браться сразу за синусы-косинусы, когда таблицу умножения только выучил".

А почему не используете ? Пишите на threading и requests ?

Пару раз была необходимость, когда работал с ботом для телеги. В остальном этих двух либ вполне хватает под мои задачи. Да и в целом, наверное, дело привычки)

 

[Whisper]

Если есть возможность не прибегать к потокам, то лучше не прибегать. Если нужна паралельность не связанная чисто с ио то как правило нужен мультипроцесс.

 

[Peace]

Обычно для контроля потоков создаю воркеров. Ниже кину код как бы я это реализовал. Посмотришь. Будут вопросы отвечу в лс. asyncio и aiohttp на самом деле для такой тривиальной задачи не нужны, тем более на моменте обучения, начни лучше с обычного threading и requests. Будет проще. К слову, я на asyncio и aiohttp не пишу вообще)))

import asyncio
import aiohttp
from aiohttp import ClientSession


HEADERS = {
    'Authority': 'proxylist.geonode.com',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
    'User-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36',
}
URL_API = 'https://proxylist.geonode.com/api/proxy-list'
FILES_PROXY = {'socks4': 'socks4.txt', 'socks5':'socks5.txt','https':'https.txt', 'http': 'http.txt'}
LIMIT_IN_PAGE = 500
THREADS_COUNT = 3

TOTAL_PROXY = 0
SAVE_PROXY_COUNTER = 0

QUEUE = asyncio.Queue()
THLOCK = asyncio.Lock()


async def write_to_file(filename:str, data:str) -> None:
    async with THLOCK:
        with open(filename, 'a', encoding='UTF-8') as file:
            file.write(f'{data}\n')


async def get_proxys(session: ClientSession, current_page: int):
    params = {'limit': f'{LIMIT_IN_PAGE}', 'page': f'{current_page}', 'sort_by': 'lastChecked', 'sort_type': 'desc'}
    try:
        async with session.get(URL_API, headers=HEADERS, params=params) as response:
            proxies = await response.json()
            return proxies.get('data')
    except Exception as e:
        print(f'[ERR] Fail get proxys in page {current_page} -- {e}')
        return None


async def get_pages_count() -> int:
    global TOTAL_PROXY
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(URL_API, headers=HEADERS) as response:
                data = await response.json()
    except Exception as e:
        print(f'[ERR] Fail get pages count -- {e}')
        return 0
    TOTAL_PROXY = int(data.get('total', 0))
    return TOTAL_PROXY // LIMIT_IN_PAGE if (TOTAL_PROXY % LIMIT_IN_PAGE) == 0 else TOTAL_PROXY // LIMIT_IN_PAGE + 1


async def worker():
    global QUEUE, SAVE_PROXY_COUNTER
    async with aiohttp.ClientSession() as session:
        while not QUEUE.empty():
            page_num = await QUEUE.get()
            proxys = await get_proxys(session, page_num)
            current_counter = 0
            for proxy in proxys:
                filename = FILES_PROXY[proxy['protocols'][0]]
                line = f"{proxy['ip']}:{proxy['port']}"
                await write_to_file(filename, line)
                current_counter +=1
            SAVE_PROXY_COUNTER += current_counter
            print(f'[+] Page {page_num} saves {current_counter} proxy. Total save: {SAVE_PROXY_COUNTER}')



async def main():
    global TOTAL_PROXY, QUEUE, SAVE_PROXY_COUNTER
    pages = await get_pages_count()
    print(f'[*] Finded proxy {TOTAL_PROXY}, pages {pages}')
    if TOTAL_PROXY == 0:
        exit()
    [QUEUE.put_nowait(page_num) for page_num in range(1, pages+1)]
    threads = [asyncio.create_task(worker()) for _ in range(1, THREADS_COUNT+1)]
    await asyncio.gather(*threads)
    print(f'[*] Finded proxy {TOTAL_PROXY} // Total save {SAVE_PROXY_COUNTER}')

if __name__ == '__main__':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())

Боже, разобрал весь код и у меня просто эйфория. Еще раз спасибо большое, что делитесь знаниями ! Но есть небольшой вопросик, в моем коде прошлом нет потоков и очереди, то почему она все ровно быстро работает ? Где-то заложены потоки, которые не видим ?

upd.
Не заметил, в прошлом сообщение, что это для контроля потоков. Сори

 

[ГошаМарков]

Боже, разобрал весь код и у меня просто эйфория. Еще раз спасибо большое, что делитесь знаниями ! Но есть небольшой вопросик, в моем коде прошлом нет потоков и очереди, то почему она все ровно быстро работает ? Где-то заложены потоки, которые не видим ?

upd.
Не заметил, в прошлом сообщение, что это для контроля потоков. Сори

Твой код работает быстрее, потому что у тебя в цикле перебирается количетсво страниц и для каждой страницы создается таск. к примеру, 10 страниц = 10 тасков.
И потом они одновременно начинают выполение " await asyncio.gather(*tasks) #распаковываем таски " в этой строке. Но предположим если будет 10к страниц? это не ок.
А у меня создаются воркеры, и очередь в которой номера страниц. Воркеры запускаются одновременно, и берут из очереди страницы которые еще не обработаны. как то так.

 

[Peace]

Твой код работает быстрее, потому что у тебя в цикле перебирается количетсво страниц и для каждой страницы создается таск. к примеру, 10 страниц = 10 тасков.
И потом они одновременно начинают выполение " await asyncio.gather(*tasks) #распаковываем таски " в этой строке. Но предположим если будет 10к страниц? это не ок.
А у меня создаются воркеры, и очередь в которой номера страниц. Воркеры запускаются одновременно, и берут из очереди страницы которые еще не обработаны. как то так.

Благодарю !

 

[Whisper]

Можно не прибегая к потокам контролировать количество одновременных тасков. Воркеры для этого не обязательны. Допустим у нас 10к тасков но мы хотим за раз запутсть не более 1к.
Запускаем первые 1к тасков, потом каждый таск завершаясь берет из листа не обработанный таск и запускает его. Скажем так 1к воркеров точно не вариант =) Модель воркеров она не про ио.

 

[ppdonchiq]

Нельзя сказать, что работа с асинхронностью сложнее, чем с потоками

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

В python потоки вообщее не являются потоками в прямом смысле этого слова

Являются. Нормальные потоки, ну с нюансом небольшим (большим).
А GIL, кстати, отпускается на вводе/выводе.

 

[Peace]

Можно не прибегая к потокам контролировать количество одновременных тасков. Воркеры для этого не обязательны. Допустим у нас 10к тасков но мы хотим за раз запутсть не более 1к.
Запускаем первые 1к тасков, потом каждый таск завершаясь берет из листа не обработанный таск и запускает его. Скажем так 1к воркеров точно не вариант =) Модель воркеров она не про ио.

Хм, но идея с воркерами интересная, там можно контролировать работу, чтобы система не перегружалась(так понимаю), при написания софта например, по типу чекера логов.

 

[Whisper]

Хм, но идея с воркерами интересная, там можно контролировать работу, чтобы система не перегружалась(так понимаю), при написания софта например, по типу чекера логов.

Ну так я и написал что воркеры они для другого, не для ио. Там где тебе нужны паралельные ио специально придумали асинки.