站長資訊網
程序員必備接口測試調試工具：立即使用
Apipost = Postman + Swagger + Mock + Jmeter
Api設計、調試、文檔、自動化測試工具
后端、前端、測試，同時在線協作，內容實時同步
有時候，下載大量圖像需要幾個小時——讓我們來解決這個問題
我明白了——你已經厭倦了等待你的程序下載圖像。有時我必須下載數千張圖像需要幾個小時，而且你不可能一直等待你的程序完成下載這些愚蠢的圖像。你有很多重要的事情要做。
讓我們構建一個簡單的圖像下載器腳本，它將讀取一個文本文件并以超快的速度下載一個文件夾中列出的所有圖像。
最終效果
這就是我們最終要構建的效果。
安裝依賴項
讓我們安裝每個人最喜歡的 requests 庫。
pip install requests
現在，我們將看到一些用于下載單個 URL 并嘗試自動查找圖像名稱以及如何使用重試的基本代碼。
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
在這里，我們重試下載圖像五次，以防失敗。現在，讓我們嘗試自動找到圖像的名稱并保存它。
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('?')]
解釋
假設我們要下載的 URL 是：
instagram.fktm7-1.fna.fbcdn.net/vp…
好吧，這是一團糟。讓我們分解一下代碼對 URL 的作用。我們首先使用 rfind 找到最后一個正斜杠（/），然后選擇之后的所有內容。這是結果：
65872070_1200425330158967_6201268309743367902_n.jpg?_nc_ht=instagram.fktm7–1.fna.fbcdn.net&_nc_cat=111
現在我們的第二部分找到一個 ?，然后只取它前面的任何東西。
這是我們最終的圖像名稱：
65872070_1200425330158967_6201268309743367902_n.jpg
這個結果非常好，適用于大多數用例。
現在我們已經下載了圖像名稱和圖像，我們將保存它。
i = Image.open(io.BytesIO(res.content)) i.save(image_name)
如果你在想，「我到底應該怎么使用上面的代碼？」那么你的想法是正確的。這是一個漂亮的函數，我們在上面所做的一切都被扁平處理了。在這里，我們還測試了下載的類型是否為圖像，以防找不到圖像名稱。
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}'
現在，你可能會問：「這個人所說的多處理在哪里？」。
這很簡單。我們將簡單地定義我們的池并將我們的函數和圖像 URL 傳遞給它。
results = ThreadPool(process).imap_unordered(image_downloader, images_url) for r in results: print(r)
讓我們把它放在一個函數中：
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)
再一次，你可能會說，「這一切都很好，但我想立即開始下載我的 1000 張圖像列表。我不想復制和粘貼所有這些代碼并試圖弄清楚如何合并所有內容。」
這是一個完整的腳本。它執行以下操作：
-
以圖像列表文本文件和進程號作為輸入
-
按照您想要的速度下載它們
-
打印下載文件的總時間
-
還有一些不錯的函數可以幫助我們讀取文件名并處理錯誤和其他東西
完整的腳本
# -*- 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(): """ 通過讀取終端中作為參數提供的 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): """ 輸入選項: 參數: img_url str (Image url) 嘗試下載圖像 url 并使用標題中提供的名稱。否則它會隨機選擇一個名字 """ 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): """ 輸入項: 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 文件中，然后運行它。
python3 image_downloader.py cats.txt
這是 GitHub 存儲庫的鏈接。
用法
python3 image_downloader.py <filename_with_urls_seperated_by_newline.txt> <num_of_process>
這將讀取文本文件中的所有 URL，并將它們下載到名稱與文件名相同的文件夾中。
num_of_process 是可選的（默認情況下，它使用 10 個進程）。
例子
python3 image_downloader.py cats.txt
我很樂意就如何進一步改進這一點做出任何回應。
英文原文地址：https://betterprogramming.pub/building-an-imagedownloader-with-multiprocessing-in-python-44aee36e0424
【
