Загрузка асинхронных изображений Python (несколько URL)

Question

Я изучаю Python 4/5 месяцев, и это мой третий проект, созданный с нуля, но я не могу решить эту проблему самостоятельно.

Этот скрипт загружает 1 изображение для каждого данного URL. Я не могу найти решение о том, как реализовать Thread Pool Executor или async в этом сценарии. Я не могу понять, как связать URL-адрес с номером изображения для сохранения части изображения. Я создаю текст всех URL-адресов, которые мне нужно загрузить, но как мне на самом деле сохранить изображение с правильным именем? Любой другой совет?

PS. URL-адреса, присутствующие в данный момент, являются только поддельными.

Синхронная версия:

    запросы на импорт import argparse import re import import loging

    from bs4 import BeautifulSoup


    parser = argparse.ArgumentParser()
    parser.add_argument("-n", "--num", help="номер книги", тип =int, обязательный =True) 
    parser.add_argument("-p", dest=r"path_name", default=r"F:\Users\123", help="Сохранить в dir",)
    args = parser.parse_args()



    logging.basicConfig(format='%(asctime)s - %(имя) s -% (имя уровня) s -% (сообщение)s', уровень =logging.ERROR)
    logger = logging.getLogger(__name__)  


    def get_parser(url_c):  
        url = f'https://test.net/g/advanurl_ccasts/1'
        logger.info(f'Main url: {url_c}')
        responsece = reports.get(url, timeout=5) # тайм-аут вызовет исключение, если respce.status_code == 200:
            page = запросы.get(url, timeout=5).content
            soup = BeautifulSoup(page, 'html.parser') возврат супа остальное: Responce.raise_for_status () def get_locators (суп): # take get_parser
        # Извлечь первую / последнюю страницу num
        first = int(soup.select_one('span.current').string)
        logger.info(f'First page: {first}')
        last = int(soup.select_one('span. Num-pages').string) + 1

        # Извлечь img_code и расширение link = soup.find('img', {'class': 'fit-horizontal'}).attrs["src"]
        logger.info(f'Locator code: {link}')
        code = re.search('galleries.([0-9]+)\/.\.(\w{3})', ссылка)
        book_code = code.group(1)   # internal расширение кода = code.group(2)   # png или jpg

        # extract название книги в директории pattern = re.compile('pretty":"(.*)"')
        found = soup.find('script', text=pattern)
        string = pattern.search(found.text).group(1)
        dir_name = string.split('"')[0]
        logger.info(f'Dir name: {dir_name}')

        logger.info(f'Hidden code: {book_code}')
        print(f'Extension: {extension}')
        print(f'Tot pages: {last}')
        print(f'')

        return {'first_p': first, 
                'last_p': last, 
                'book_code': book_code, 
                'ext': расширение, 
                'dir': dir_name
                }


    def setup_download_dir(path, dir):  # (args.path_name, locator['dir'])
        # Создать папку, если она не существует filepath = os.path.join(f'{path}\{dir}'), если не os.path.exists(filepath):
            try:
                os.makedirs(filepath)
                print(f'Directory, созданный в: {filepath}'), за исключением OSError как err:
                print(f"Невозможно создать {filepath}: {err}") вернуть filepath 


    def main(locator, filepath): для image_n в диапазоне (locator['first_p'], locator['last_p']):
            url = f"https://i.test.net/galleries/ enjlocator['book_code']]/ enjimage_n‹. enjlocator['ext']}"
            logger.info(f'Url Img: {url}') ответить = reports.get(url, timeout=3)
            if responsece.status_code == 200:
                img_data = запросы.get(url, timeout=3).content 
            else:    
                responsece.raise_for_status()    # поднять исключение с помощью open((os.path.join(filepath, f"{image_n}.{locator['ext']}")), 'wb') в качестве обработчика:
                handler.write(img_data) # запись печати изображения (f'Img {image_n} - DONE')


    if __name__ == '__main__':
        try:
            locator = get_locators(get_parser(args.num))    # args.num ex. 241461
            main(locator, setup_download_dir(args.path_name, locator['dir'])), кроме KeyboardInterrupt:
            print(f'Программа прервана...' + '\n')

Список URL:

    def img_links (locator):
        image_url = []
        для num в диапазоне (локатор ['first_p'], локатор ['last_p']):
            url = f "https://i.test.net/galleries/ndomlocator['book_code']‹/ndomnum‹. enjlocator['ext']}"
            image_url.append(URL)
        logger.info (f'Url List: {image_url}')       
        вернуть image_url

Answer 1

Я нашел решение в книге беглый питон. Вот фрагмент:

    def download_many (cc_list, base_url, verbose, concur_req):
        counter = collection.Counter()
        с futures.ThreadPoolExecutor(max_workers=concur_req) в качестве исполнителя:
            to_do_map = {}
            для cc в отсортированном (cc_list):
                future = executor.submit(download_one, cc, base_url, verbose)
                to_do_map[future] = cc
            done_iter = futures.as_completed(to_do_map)
            если не многословно:
                done_iter = tqdm.tqdm(done_iter, всего = len (cc_list))
            на будущее в done_iter:
                пытаться:
                    res = future.result ()
                кроме запросов. исключений.HTTPError как exc:
                    error_msg = 'HTTP {res.status_code} - {res.reason}'
                    error_msg = error_msg.format(res=exc.response)
                кроме запросов. исключений. Соединение с ошибкой как исключение:
                    error_msg = 'Ошибка соединения'
                еще:
                    error_msg = ''
                    status = res.status
                если error_msg:
                    status = HTTPStatus.error
                счетчик [статус] += 1
                если многословно и error_msg:
                    cc = to_do_map[будущее]
                    print('*** Ошибка для {}: {}'.format(cc, error_msg))
        счетчик возврата