程序員必備接口測(cè)試調(diào)試工具：立即使用
Apipost = Postman + Swagger + Mock + Jmeter
Api設(shè)計(jì)、調(diào)試、文檔、自動(dòng)化測(cè)試工具
后端、前端、測(cè)試，同時(shí)在線協(xié)作，內(nèi)容實(shí)時(shí)同步

有時(shí)候，下載大量圖像需要幾個(gè)小時(shí)——讓我們來(lái)解決這個(gè)問(wèn)題

我明白了——你已經(jīng)厭倦了等待你的程序下載圖像。有時(shí)我必須下載數(shù)千張圖像需要幾個(gè)小時(shí)，而且你不可能一直等待你的程序完成下載這些愚蠢的圖像。你有很多重要的事情要做。

讓我們構(gòu)建一個(gè)簡(jiǎn)單的圖像下載器腳本，它將讀取一個(gè)文本文件并以超快的速度下載一個(gè)文件夾中列出的所有圖像。

最終效果

這就是我們最終要構(gòu)建的效果。

安裝依賴項(xiàng)

讓我們安裝每個(gè)人最喜歡的 requests 庫(kù)。

pip install requests

現(xiàn)在，我們將看到一些用于下載單個(gè) URL 并嘗試自動(dòng)查找圖像名稱以及如何使用重試的基本代碼。

import requests  res = requests.get(img_url, stream=True) count = 1 while res.status_code != 200 and count <= 5:     res = requests.get(img_url, stream=True)     print(f'Retry: {count} {img_url}')     count += 1

在這里，我們重試下載圖像五次，以防失敗。現(xiàn)在，讓我們嘗試自動(dòng)找到圖像的名稱并保存它。

import more required library  import io from PIL import Image  # lets try to find the image name image_name = str(img_url[(img_url.rfind('/')) + 1:]) if '?' in image_name:     image_name = image_name[:image_name.find('?')]

解釋

假設(shè)我們要下載的 URL 是：

instagram.fktm7-1.fna.fbcdn.net/vp…

好吧，這是一團(tuán)糟。讓我們分解一下代碼對(duì) URL 的作用。我們首先使用 rfind 找到最后一個(gè)正斜杠（/），然后選擇之后的所有內(nèi)容。這是結(jié)果：

65872070_1200425330158967_6201268309743367902_n.jpg?_nc_ht=instagram.fktm7–1.fna.fbcdn.net&_nc_cat=111

現(xiàn)在我們的第二部分找到一個(gè) ?，然后只取它前面的任何東西。

這是我們最終的圖像名稱：

65872070_1200425330158967_6201268309743367902_n.jpg

這個(gè)結(jié)果非常好，適用于大多數(shù)用例。

現(xiàn)在我們已經(jīng)下載了圖像名稱和圖像，我們將保存它。

i = Image.open(io.BytesIO(res.content)) i.save(image_name)

如果你在想，「我到底應(yīng)該怎么使用上面的代碼？」那么你的想法是正確的。這是一個(gè)漂亮的函數(shù)，我們?cè)谏厦嫠龅囊磺卸急槐馄教幚砹恕Ｔ谶@里，我們還測(cè)試了下載的類型是否為圖像，以防找不到圖像名稱。

def image_downloader(img_url: str):     """     Input:     param: img_url  str (Image url)     Tries to download the image url and use name provided in headers. Else it randomly picks a name     """     print(f'Downloading: {img_url}')     res = requests.get(img_url, stream=True)     count = 1     while res.status_code != 200 and count <= 5:         res = requests.get(img_url, stream=True)         print(f'Retry: {count} {img_url}')         count += 1     # checking the type for image     if 'image' not in res.headers.get("content-type", ''):         print('ERROR: URL doesnot appear to be an image')         return False     # Trying to red image name from response headers     try:         image_name = str(img_url[(img_url.rfind('/')) + 1:])         if '?' in image_name:             image_name = image_name[:image_name.find('?')]     except:         image_name = str(random.randint(11111, 99999))+'.jpg'      i = Image.open(io.BytesIO(res.content))     download_location = 'cats'     i.save(download_location + '/'+image_name)     return f'Download complete: {img_url}'

現(xiàn)在，你可能會(huì)問(wèn)：「這個(gè)人所說(shuō)的多處理在哪里？」。

這很簡(jiǎn)單。我們將簡(jiǎn)單地定義我們的池并將我們的函數(shù)和圖像 URL 傳遞給它。

results = ThreadPool(process).imap_unordered(image_downloader, images_url) for r in results:     print(r)

讓我們把它放在一個(gè)函數(shù)中：

def run_downloader(process:int, images_url:list):     """     Inputs:         process: (int) number of process to run         images_url:(list) list of images url     """     print(f'MESSAGE: Running {process} process')     results = ThreadPool(process).imap_unordered(image_downloader, images_url)     for r in results:         print(r)

再一次，你可能會(huì)說(shuō)，「這一切都很好，但我想立即開(kāi)始下載我的 1000 張圖像列表。我不想復(fù)制和粘貼所有這些代碼并試圖弄清楚如何合并所有內(nèi)容。」

這是一個(gè)完整的腳本。它執(zhí)行以下操作：

• 以圖像列表文本文件和進(jìn)程號(hào)作為輸入

• 按照您想要的速度下載它們

• 打印下載文件的總時(shí)間

• 還有一些不錯(cuò)的函數(shù)可以幫助我們讀取文件名并處理錯(cuò)誤和其他東西

完整的腳本

# -*- coding: utf-8 -*- import io import random import shutil import sys from multiprocessing.pool import ThreadPool import pathlib  import requests from PIL import Image import time  start = time.time()  def get_download_location():     try:         url_input = sys.argv[1]     except IndexError:         print('ERROR: Please provide the txt filen$python image_downloader.py cats.txt')     name = url_input.split('.')[0]     pathlib.Path(name).mkdir(parents=True, exist_ok=True)     return name  def get_urls():     """     通過(guò)讀取終端中作為參數(shù)提供的 txt 文件返回 url 列表     """     try:         url_input = sys.argv[1]     except IndexError:         print('ERROR: Please provide the txt filen Example nn$python image_downloader.py dogs.txt nn')         sys.exit()     with open(url_input, 'r') as f:         images_url = f.read().splitlines()      print('{} Images detected'.format(len(images_url)))     return images_url  def image_downloader(img_url: str):     """     輸入選項(xiàng):     參數(shù): img_url  str (Image url)     嘗試下載圖像 url 并使用標(biāo)題中提供的名稱。否則它會(huì)隨機(jī)選擇一個(gè)名字     """     print(f'Downloading: {img_url}')     res = requests.get(img_url, stream=True)     count = 1     while res.status_code != 200 and count <= 5:         res = requests.get(img_url, stream=True)         print(f'Retry: {count} {img_url}')         count += 1     # checking the type for image     if 'image' not in res.headers.get("content-type", ''):         print('ERROR: URL doesnot appear to be an image')         return False     # Trying to red image name from response headers     try:         image_name = str(img_url[(img_url.rfind('/')) + 1:])         if '?' in image_name:             image_name = image_name[:image_name.find('?')]     except:         image_name = str(random.randint(11111, 99999))+'.jpg'      i = Image.open(io.BytesIO(res.content))     download_location = get_download_location()     i.save(download_location + '/'+image_name)     return f'Download complete: {img_url}'  def run_downloader(process:int, images_url:list):     """     輸入項(xiàng):         process: (int) number of process to run         images_url:(list) list of images url     """     print(f'MESSAGE: Running {process} process')     results = ThreadPool(process).imap_unordered(image_downloader, images_url)     for r in results:         print(r)  try:     num_process = int(sys.argv[2]) except:     num_process = 10  images_url = get_urls() run_downloader(num_process, images_url)  end = time.time() print('Time taken to download {}'.format(len(get_urls()))) print(end - start)

將其保存到 Python 文件中，然后運(yùn)行它。

python3 image_downloader.py cats.txt

這是 GitHub 存儲(chǔ)庫(kù)的鏈接。

用法

python3 image_downloader.py <filename_with_urls_seperated_by_newline.txt> <num_of_process>

這將讀取文本文件中的所有 URL，并將它們下載到名稱與文件名相同的文件夾中。

num_of_process 是可選的（默認(rèn)情況下，它使用 10 個(gè)進(jìn)程）。

例子

python3 image_downloader.py cats.txt

我很樂(lè)意就如何進(jìn)一步改進(jìn)這一點(diǎn)做出任何回應(yīng)。

英文原文地址：https://betterprogramming.pub/building-an-imagedownloader-with-multiprocessing-in-python-44aee36e0424

【