批量下载网络图片或视频

可复用可组合。

概述

在互联网上浏览时，遇到好看的图片或视频，总想下载保存起来。本文讲述使用Python实现批量下载网络图片或视频的方法。
批量下载网络图片或视频，主要有四步：

• 生成多个 URL：因为可能需要从多个URL获取图片或视频资源；
  
• 抓取网页内容：为析取图片视频资源作准备；
  
• 解析URL内容：析取图片视频资源；
  
• 下载图片和视频：可调用下载工具。
  

我们对程序的目标依然是：

• 可复用：尽量写一次，多处使用。
  
• 可组合：能够组合出不同功能的工具。
  

要做到可复用，就需要尽量把通用的部分抽离出来，做成选项； 要做到可组合，就需要把功能合理地拆分成多个子任务，用不同程序去实现这些子任务。

生成多个URL

多个URL往往是在一个基础URL经过某种变形得到。可以用含有占位符 :p 的模板 t 和参数组成。
gu.py

1. from typing import List
2. from urllib.parse import urlparse
3. import argparse
5. def generate_urls(base_url: str, m: int, n: int) -> List[str]:
6.     """
7.     Generate a series of URLs based on a base URL and transformation rules.
8.   
9.     Args:
10.         base_url (str): The base URL to transform
11.         m (int): Start number
12.         n (int): End number (inclusive)
13.   
14.     Returns:
15.         List[str]: List of generated URLs
16.   
17.     Examples:
18.         >>> generate_urls("https://example.com/xxx:pyyy", 1, 3)
19.         ['https://example.com/xxx1yyy', 'https://example.com/xxx2yyy', 'https://example.com/xxx3yyy']
20.         >>> generate_urls("https://example.com/page_:p.php", 1, 3)
21.         ['https://example.com/page_1.php', 'https://example.com/page_2.php', 'https://example.com/page_3.php']
22.     """
23.     if not base_url or not isinstance(m, int) or not isinstance(n, int):
24.         raise ValueError("Invalid input parameters")
25.   
26.     if m > n:
27.         raise ValueError("Start number (m) must be less than or equal to end number (n)")
28.   
29.     # Parse the URL to validate it
30.     parsed_url = urlparse(base_url)
31.     if not parsed_url.scheme and not base_url.startswith('//'):
32.         raise ValueError("Invalid URL format")
33.   
34.     # Handle the $p pattern
35.     if ":p" in base_url:
36.         parts = base_url.split(":p")
37.         if len(parts) != 2:
38.             raise ValueError("Invalid URL pattern: should contain exactly one $p")
39.         prefix, suffix = parts
40.         return [f"{prefix}{i}{suffix}" for i in range(m, n + 1)]
41.   
42.     raise ValueError("URL pattern not supported. Use $p as placeholder for numbers")
44. def parse_range(range_str: str) -> tuple[int, int]:
45.     """
46.     Parse a range string like "1-3" into start and end numbers.
47.   
48.     Args:
49.         range_str (str): Range string (e.g., "1-3")
50.   
51.     Returns:
52.         tuple[int, int]: Start and end numbers
53.     """
54.     try:
55.         start, end = map(int, range_str.split("-"))
56.         return start, end
57.     except ValueError:
58.         raise ValueError("Invalid range format. Use 'start-end' (e.g., '1-3')")
60. def parse_list(list_str: str) -> List[str]:
61.     """
62.     Parse a comma-separated string into a list of values.
63.   
64.     Args:
65.         list_str (str): Comma-separated string (e.g., "1,2,3")
66.   
67.     Returns:
68.         List[str]: List of values
69.     """
70.     return [item.strip() for item in list_str.split(",")]
72. def main():
73.     parser = argparse.ArgumentParser(description='Generate a series of URLs based on a pattern')
74.     parser.add_argument('-u', '--url', required=True, help='Base URL with {p} as placeholder')
75.   
76.     # Add mutually exclusive group for range or list
77.     group = parser.add_mutually_exclusive_group(required=True)
78.     group.add_argument('-r', '--range', help='Range of numbers (e.g., "1-3")')
79.     group.add_argument('-l', '--list', help='Comma-separated list of values (e.g., "1,2,3")')
80.   
81.     args = parser.parse_args()
82.   
83.     try:
84.         if args.range:
85.             start, end = parse_range(args.range)
86.             urls = generate_urls(args.url, start, end)
87.         elif args.list:
88.             values = parse_list(args.list)
89.             template = args.url.replace(":p", "{}")
90.             urls = [template.format(value) for value in values]
91.       
92.         for url in urls:
93.             print(url)
94.     except ValueError as e:
95.         print(f"Error: {e}")
96.         exit(1)
98. if __name__ == "__main__":
99.     main()

复制代码

使用方法：

1. gu -u "https://www.yituyu.com/gallery/:p/index.html" -l "10234,10140"

复制代码

就可以生成 https://www.yituyu.com/gallery/10234/index.html, https://www.yituyu.com/gallery/10140/index.html
或者使用

1. gu -u "https://www.yituyu.com/gallery/:p/index.html" -r 1-3

复制代码

生成： https://www.yituyu.com/gallery/1/index.html，https://www.yituyu.com/gallery/2/index.html，https://www.yituyu.com/gallery/3/index.html
‍

抓取网页内容

web.py
这里使用了 requests 和 chromedriver 。静态网页可以直接用 requests，动态网页需要用 chromedriver 模拟打开网页。有些网页还需要滚动到最下面加载资源。

1. import requests
2. import time
3. from pytools.common.common import catchExc
4. from pytools.con.multitasks import IoTaskThreadPool
5. from selenium import webdriver
6. from selenium.webdriver.chrome.options import Options
7. from selenium.webdriver.common.by import By
9. delayForHttpReq = 0.5 # 500ms
11. class HTMLGrasper(object):
13.     def __init__(self, conf):
14.         '''
15.         抓取 HTML 网页内容时的配置项
16.           _async: 是否异步加载网页。 _async = 1 当网页内容是动态生成时，异步加载网页;
17.           targetIdWhenAsync: 当 _async = 1 指定。
18.              由于此时会加载到很多噪音内容，需要指定 ID 来精确获取所需的内容部分
19.           sleepWhenAsync:  当 _async = 1 指定。
20.              异步加载网页时需要等待的秒数  
21.         '''
22.         self._async = conf.get('async', 0)
23.         self.targetIdWhenAsync = conf.get('targetIdWhenAsync', '')
24.         self.sleepWhenAsync = conf.get('sleepWhenAsync', 10)
26.     def batchGrapHtmlContents(self, urls):
27.         '''
28.            batch get the html contents of urls
29.         '''
30.         grapHtmlPool = IoTaskThreadPool(20)
31.         return grapHtmlPool.exec(self.getHTMLContent, urls)
33.     def getHTMLContent(self, url):
34.         if self._async == 1:
35.             htmlContent = self.getHTMLContentAsync(url)
37.             if htmlContent is not None and htmlContent != '':
38.                 html = '<html><head></head><body>' + htmlContent + '</body></html>'
39.                 return html
41.         return self.getHTMLContentFromUrl(url)
43.     def getHTMLContentAsync(self, url):
44.         '''
45.            get html content from dynamic loaed html url
46.         '''
48.         chrome_options = Options()
49.         chrome_options.add_argument('--headless')
50.         chrome_options.add_argument('--disable-gpu')
51.         driver = webdriver.Chrome(chrome_options=chrome_options)
52.         driver.get(url)
53.         time.sleep(self.sleepWhenAsync)
54.   
55.         # 模拟滚动到底部多次以确保加载所有内容
56.         last_height = driver.execute_script("return document.body.scrollHeight")
57.         for _ in range(3):  # 最多滚动3次
58.             # 滚动到底部
59.             driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
60.             # 等待加载
61.             time.sleep(2)
62.             # 计算新的滚动高度并与上一次比较
63.             new_height = driver.execute_script("return document.body.scrollHeight")
64.             if new_height == last_height:
65.                 break
66.             last_height = new_height
68.         try:
69.             elem = driver.find_element_by_id(self.targetIdWhenAsync)
70.         except:
71.             elem = driver.find_element(By.XPATH, '/html/body')
73.         return elem.get_attribute('innerHTML')   
75.     def getHTMLContentFromUrl(self, url):
76.         '''
77.            get html content from html url
78.         '''
79.         r = requests.get(url)
80.         status = r.status_code
81.         if status != 200:
82.             return ''
83.         return r.text
86. '''
87.     # 利用property装饰器将获取name方法转换为获取对象的属性
88.     @property
89.     def async(self):
90.         return self._async
92.     # 利用property装饰器将设置name方法转换为获取对象的属性
93.     @async.setter
94.     def async(self,async):
95.         self._async = async
96. '''   

复制代码

‍

析取图片或视频资源

res.py
这里使用了 BeautifulSoup.  网页文件通常是 HTML。因此，需要写一个程序，从 HTML 内容中解析出图片或资源地址。现代web页面通常采用 DIV+CSS+JS 框架。 图片或视频资源，通常是 a, img, video 之类的标签，或者 class 或 id 指定的元素 。再定位到元素的属性，比如 href, src 等。此处需要一点 HTML 和 CSS 知识，还有 jQuery 定位元素的知识。

1. #!/usr/bin/python3
2. #_*_encoding:utf-8_*_
4. import re
5. import sys
6. import json
8. import argparse
9. from bs4 import BeautifulSoup
10. from pytools.net.web import HTMLGrasper   
12. SaveResLinksFile = '/Users/qinshu/joy/reslinks.txt'
13. serverDomain = ''
15. def parseArgs():
16.     description = '''This program is used to batch fetch url resources from specified urls.
17.                      eg. python3 res.py -u http://xxx.html -r 'img=jpg,png;class=resLink;id=xyz'
18.                      will search resource links from network urls http://xxx.html  by specified rules
19.                      img = jpg or png OR class = resLink OR id = xyz [ multiple rules ]
21.                      python3 tools/res.py -u 'http://tu.heiguang.com/works/show_167480.html' -r 'img=jpg!c'
22.                      for <img src="https://www.cnblogs.com/xxx.jpg!c"/>
23.                   '''
24.     parser = argparse.ArgumentParser(description=description)
25.     parser.add_argument('-u','--url', nargs='+', help='At least one html urls are required', required=True)
26.     parser.add_argument('-r','--rulepath', nargs=1, help='rules to search resources. if not given, search a hrefs or img resources in given urls', required=False)
27.     parser.add_argument('-o','--output', nargs=1, help='Specify the output file to save the links', required=False)
28.     parser.add_argument('-a','--attribute', nargs=1, help='Extract specified attribute values from matched elements', required=False)
29.     args = parser.parse_args()
30.     init_urls = args.url
31.     rulepath = args.rulepath
32.     output_file = args.output[0] if args.output else SaveResLinksFile
33.     return (init_urls, rulepath, output_file, args.attribute[0] if hasattr(args, 'attribute') and args.attribute else None)
35. def getAbsLink(serverDomain, link):
36.     try:
37.         link_content = link
38.         if link_content.startswith('//'):
39.             link_content = 'https:' + link_content
40.         if link_content.startswith('/'):
41.             link_content = serverDomain + link_content
42.         return link_content
43.     except:
44.         return ''
46. def batchGetResTrueLink(resLinks):
47.     return filter(lambda x: x != '', resLinks)
49. resTags = set(['img', 'video', 'a', 'div'])
51. def findWantedLinks(htmlcontent, rule, attribute):
52.     '''
53.        find html links or res links from html by rule.
54.        sub rules such as:
55.           (1) a link with id=[value1,value2,...]
56.           (2) a link with class=[value1,value2,...]
57.           (3) res with src=xxx.jpg|png|mp4|...
58.        a rule is map containing sub rule such as:
59.           { 'id': [id1, id2, ..., idn] } or
60.           { 'class': [c1, c2, ..., cn] } or
61.           { 'img': ['jpg', 'png', ... ]} or
62.           { 'video': ['mp4', ...]}
64.     '''
66.     #print("html===\n"+htmlcontent+"\n===End")
67.     #print("rule===\n"+str(rule)+"\n===End")
69.     soup = BeautifulSoup(htmlcontent, "lxml")
70.     all_links = []
71.     attribute_links = []
73.     for (key, values) in rule.items():
74.         if key == 'id':
75.             for id in values:
76.                 link_soups = soup.find_all('a', id=id)
77.         elif key == 'class':
78.             for cls in values:
79.                 link_soups = findLinkSoups(soup, ['a', 'img', 'div'], cls)   
80.         elif key in resTags:
81.             link_soups = []
82.             for resSuffix in values:
83.                 if resSuffix != "":
84.                     link_soups.extend(soup.find_all(key, src=re.compile(resSuffix)))
85.                 else:
86.                     link_soups = soup.find_all(key)
87.     attribute_links.extend([link.get(attribute) for link in link_soups if link.get(attribute)])
88.     all_links.extend(attribute_links)
89.     return all_links
91. def findLinkSoups(soup, tags, cls):
92.     all_link_soups = []
93.     if len(tags) == 0:
94.         all_link_soups.extend(soup.find_all("a", class_=cls))  
95.     else:
96.         for tag in tags:
97.             if cls != "":
98.                 link_soups = soup.find_all(tag, class_=cls)
99.             else:
100.                 link_soups = soup.find_all(tag)
101.             all_link_soups.extend(link_soups)
102.     return all_link_soups
104. def batchGetLinks(urls, rules, output_file, attribute=None):
105.     conf = {"async":1, "targetIdWhenAsync": "page-fav", "sleepWhenAsync": 10}
106.     grasper = HTMLGrasper(conf)
107.     htmlcontentList = grasper.batchGrapHtmlContents(urls)
108.     allLinks = []
109.     for htmlcontent in htmlcontentList:
110.         for rule in rules:
111.             links = findWantedLinks(htmlcontent, rule, attribute)
112.             allLinks.extend(links)
113.   
114.     with open(output_file, 'w') as f:
115.         for link in allLinks:
116.             print(link)
117.             f.write(link + "\n")
119. def parseRulesParam(rulesParam):
120.     '''
121.        parse rules params to rules json
122.        eg. img=jpg,png;class=resLink;id=xyz to
123.            [{"img":["jpg","png"], "class":["resLink"], "id":["xyz"]}]
124.     '''
125.     defaultRules = [{'img': ['jpg','png','jpeg']},{"class":"*"}]
126.     if rulesParam:
127.         try:
128.             rules = []
129.             rulesStrArr = rulesParam[0].split(";")
130.             for ruleStr in rulesStrArr:
131.                 ruleArr = ruleStr.split("=")
132.                 key = ruleArr[0]
133.                 value = ruleArr[1].split(",")
134.                 rules.append({key: value})
135.             return rules
136.         except ValueError as e:
137.             print('Param Error: invalid rulepath %s %s' % (rulepathjson, e))
138.             sys.exit(1)
139.     return defaultRules
141. def testBatchGetLinks():
142.     urls = ['http://dp.pconline.com.cn/list/all_t145.html']
143.     rules = [{"img":["jpg"], "video":["mp4"]}]
145.     batchGetLinks(urls, rules, SaveResLinksFile)
147. if __name__ == '__main__':
149.     #testBatchGetLinks()
151.     (init_urls, rulesParam, output_file, attribute) = parseArgs()
152.     if not output_file:
153.         output_file = SaveResLinksFile
154.     # print('init urls: %s' % "\n".join(init_urls))
156.     rulepath = parseRulesParam(rulesParam)
157.     # print('rulepath: %s\n' % (rulepath))
159.     batchGetLinks(init_urls, rulepath, output_file, attribute)

复制代码

怎么找到对应的资源地址？右键-控制台，鼠标点击最左边那个箭头指向的小方框，然后在点击网页元素，就会定位到网页元素，下面图片资源地址就是 img 的 data-src 属性 或者 src 属性。 不过这里 src 属性是需要滚动到最后才能展示所有的，但 data-src 是直接加载的。如果要省时，就可以用 data-src 属性。好处是快，不足是不通用。视频地址类似。

将图片资源URL下载到 ~/Downloads/links5.txt，整个命令是：

1. gu -u "https://www.yituyu.com/gallery/:p/index.html" -l "9174,9170" | xargs -I {} python3 ~/tools/pytools/pytools/tools/res.py -u {} -r "class=lazy" -a "data-src" -o ~/Downloads/links5.txt

复制代码

下载图片或视频

dw.py

1. #!/usr/bin/env python3
3. import subprocess
4. import shlex
5. from pathlib import Path
6. from typing import Optional, Union, List
7. import time
8. import requests
9. import argparse
11. default_save_path = "/Users/qinshu/Downloads"
13. def download(url: str, output_dir: Union[str, Path]) -> Optional[Path]:
14.     output_dir = Path(output_dir)
15.     if url.endswith(".jpg") or url.endswith(".png"):
16.         download_image(url, output_dir / Path(url).name)
17.     else:
18.         download_video(url, output_dir)  
19.         return output_dir / Path(url).name
20.     return None
22. def download_image(url: str, output_file: Union[str, Path]) -> None:
23.     try:
24.         # 发送 HTTP GET 请求获取图片
25.         response = requests.get(url, stream=True)
26.         response.raise_for_status()  # 检查请求是否成功
28.         # 以二进制写入模式保存图片
29.         with open(output_file, 'wb') as f:
30.             for chunk in response.iter_content(1024):
31.                 f.write(chunk)
32.       
33.         print(f"图片已保存至: {output_file}")
34.     except requests.exceptions.RequestException as e:
35.         print(f"下载失败: {e}")
36.     except Exception as e:
37.         print(f"发生错误: {e}")
39. def download_video(
40.     video_url: str,
41.     output_dir: Union[str, Path] = Path.cwd(),
42.     timeout: int = 3600,  # 1小时超时
43.     retries: int = 1,
44.     verbose: bool = True
45. ) -> Optional[Path]:
46.     """
47.     使用 y 命令下载视频
48.   
49.     参数:
50.         video_url: 视频URL (e.g. "https://www.bilibili.com/video/BV1xx411x7xx")
51.         output_dir: 输出目录 (默认当前目录)
52.         timeout: 超时时间(秒)
53.         retries: 重试次数
54.         verbose: 显示下载进度
55.   
56.     返回:
57.         成功时返回下载的视频路径，失败返回None
58.     """
59.     if video_url == "":
60.         return None
61.     output_dir = Path(output_dir)
62.     output_dir.mkdir(parents=True, exist_ok=True)
63.   
64.     cmd = f"y {shlex.quote(video_url)}"
65.     if verbose:
66.         print(f"开始下载: {video_url}")
67.         print(f"保存到: {output_dir.resolve()}")
68.         print(f"执行命令: {cmd}")
69.   
70.     for attempt in range(1, retries + 1):
71.         try:
72.             start_time = time.time()
73.          
74.             # 使用Popen实现实时输出
75.             process = subprocess.Popen(
76.                 cmd,
77.                 shell=True,
78.                 cwd=str(output_dir),
79.                 stdout=subprocess.PIPE,
80.                 stderr=subprocess.STDOUT,
81.                 universal_newlines=True,
82.                 bufsize=1
83.             )
84.          
85.             # 实时打印输出
86.             while True:
87.                 output = process.stdout.readline()
88.                 if output == '' and process.poll() is not None:
89.                     break
90.                 if output and verbose:
91.                     print(output.strip())
92.               
93.                 # 检查超时
94.                 if time.time() - start_time > timeout:
95.                     process.terminate()
96.                     raise subprocess.TimeoutExpired(cmd, timeout)
97.          
98.             # 检查返回码
99.             if process.returncode == 0:
100.                 if verbose:
101.                     print(f"下载成功 (尝试 {attempt}/{retries})")
102.                 return _find_downloaded_file(output_dir, video_url)
103.             else:
104.                 raise subprocess.CalledProcessError(process.returncode, cmd)
105.               
106.         except (subprocess.TimeoutExpired, subprocess.CalledProcessError) as e:
107.             if attempt < retries:
108.                 wait_time = min(attempt * 10, 60)  # 指数退避
109.                 if verbose:
110.                     print(f"尝试 {attempt}/{retries} 失败，{wait_time}秒后重试...")
111.                     print(f"错误: {str(e)}")
112.                 time.sleep(wait_time)
113.             else:
114.                 if verbose:
115.                     print(f"下载失败: {str(e)}")
116.                 return None
118. def _find_downloaded_file(directory: Path, video_url: str) -> Optional[Path]:
119.     """尝试自动查找下载的文件"""
120.     # 这里可以根据实际y命令的输出文件名模式进行调整
121.     # 示例：查找最近修改的视频文件
122.     video_files = sorted(
123.         directory.glob("*.mp4"),
124.         key=lambda f: f.stat().st_mtime,
125.         reverse=True
126.     )
127.     return video_files[0] if video_files else None
129. def read_urls_from_file(file_path: Union[str, Path]) -> List[str]:
130.     with open(file_path, 'r') as f:
131.         return [line.strip() for line in f if line.strip()]
133. def main():
134.     parser = argparse.ArgumentParser(description="下载工具：支持从URL或文件下载视频和图片")
135.     parser.add_argument("-u", "--url", help="单个下载URL")
136.     parser.add_argument("-f", "--file", help="包含多个URL的文件路径（每行一个URL）")
137.     parser.add_argument("-o", "--output", default=".", help="输出目录路径（默认为当前目录）")
138.     args = parser.parse_args()
140.     if not args.url and not args.file:
141.         parser.error("必须提供 -u 或 -f 参数")
143.     if not args.output:
144.         output_dir = default_save_path
145.     else:   
146.         output_dir = Path(args.output)
147.     output_dir.mkdir(parents=True, exist_ok=True)
149.     urls = []
150.     if args.url:
151.         urls.append(args.url)
152.     if args.file:
153.         urls.extend(read_urls_from_file(args.file))
155.     for url in urls:
156.         print(f"处理URL: {url}")
157.         result = download(url, output_dir)
158.         if result:
159.             print(f"下载完成: {result}")
160.         else:
161.             print(f"下载失败: {url}")
163. if __name__ == "__main__":
164.     main()

复制代码

注册工具

要想运行工具，就需要 python /path/to/python_file.py ，每次写全路径挺麻烦的。可以写一个shell 脚本，将写的 python 工具注册到 ~/.zshrc ，然后每次 source ~/.zshrc 即可。

1. #!/bin/bash
3. # Get the absolute path of the tools directory
4. TOOLS_DIR="/Users/qinshu/tools/pytools/pytools/tools"
6. # Add a comment to mark the beginning of our tools section
7. echo -e "\n# === Python Tools Aliases ===" >> ~/.zshrc
9. # Loop through all Python files in the tools directory
10. for file in "$TOOLS_DIR"/*.py; do
11.     if [ -f "$file" ]; then
12.         # Get just the filename without extension
13.         filename=$(basename "$file" .py)
14.       
15.         # Skip .DS_Store and any other hidden files
16.         if [[ $filename != .* ]]; then
17.             # Create the alias command
18.             alias_cmd="alias $filename="python3 $file""
19.          
20.             # Check if this alias already exists in .zshrc
21.             if ! grep -q "alias $filename=" ~/.zshrc; then
22.                 echo "$alias_cmd" >> ~/.zshrc
23.                 echo "Added alias for: $filename"
24.             else
25.                 echo "Alias already exists for: $filename"
26.             fi
27.         fi
28.     fi
29. done
31. echo "All Python tools have been registered in ~/.zshrc"
32. echo "Please run 'source ~/.zshrc' to apply the changes"

复制代码

这里实际上就是生成了一系列 alias:

1. alias gu="python3 /Users/qinshu/tools/pytools/pytools/tools/gu.py"
2. alias res="python3 /Users/qinshu/tools/pytools/pytools/tools/res.py"
3. alias dw="python3 /Users/qinshu/tools/pytools/pytools/tools/dw.py"

复制代码

这样就可以直接用

1. gu -u 'https://xxx' -l 1-3

复制代码

从该网站批量下载图片的整个命令合起来是：

1. gu -u "https://www.yituyu.com/gallery/:p/index.html" -l "9174,9170" | xargs -I {} python3 ~/tools/pytools/pytools/tools/res.py -u {} -r "class=lazy" -a "data-src" -o ~/Downloads/links5.txt  && dw -f -o ~/Downloads/links5.txt

复制代码

虽然繁琐了一点，但是胜在通用。
‍

小结

本文讲解了批量下载网络图片或视频的方法，包括四个主要步骤：生成多个URL、抓取网页内容、析取资源地址、下载资源。每个步骤既独立又承上启下，因此做到了可组合。要做到通用，需要掌握一些基本编程知识。当你了解其中一些原理时，就能获得更强的能力，而不仅仅局限于 GUI。


来源：豆瓜网用户自行投稿发布，如果侵权，请联系站长删除