前段时间我写了一个爬取网站的图片的python程序代码如下#codingutf-8#支持本程序中有汉字否则报错 import re#导入正则表达式模块 import requests#导入网络请求,如果没有安装命令pip install requests import os #urlhttp://www.netbian.com/#图片网址 #urlhttp://www.k2r2.com/ urlhttps://sc.chinaz.com/tupian/fengjingtupian.html #定义变量n并且赋值为1 n1 def downloadpic(): datarequests.get(url).text#将网页的源代码写入变量data中 #jpglistre.findall(data-original(.*?),data,re.S)#从data文本中查找以data-original开头和以空格结束的中间的字符串给列表jpglistre.S不会对\n进行中段 jpglistre.findall(data-original(.*?),data,re.S)#从data文本中查找以data-original开头和以空格结束的中间的字符串给列表jpglistre.S不会对\n进行中段 global n for each in jpglist:#遍历列表jpglist neweach.strip() new1new.replace(_s,) eachhttps:new1 print(each)#输出列表对应的元素 try:#尝试执行 picrequests.get(each,timeout10)#将请求响应结果赋值给pic except:#尝试失败时 print(下载失败)#输出下载失败 continue#继续循环 stringpic\\str(n).jpg#构造文件路径和文件名 fpopen(string,wb)#创建文件对象fp fp.write(pic.content)#将pic图片对象以二进制形式写入文件 fp.close()#关闭文件对象fp n1#变量n自加1 if not os.path.exists(pic): os.makedirs(pic) downloadpic() url1https://sc.chinaz.com/tupian/fengjingtupian_ for i in range(2,3): urlurl1str(i).html downloadpic() print(下载完成)这个程序虽然能够正常运行但我感觉可能会存在逐多问题于是我将代码发给了workbuddy让它给我优化和完善经过一段时间它给我优化完善后的代码是# -*- coding: utf-8 -*- 站长素材sc.chinaz.com风景图片批量下载器 —— 优化完善版 【原版的问题】 1. 缩进全丢粘贴后所有语句挤在一行直接 SyntaxError 无法运行。 2. Windows 控制台默认 GBKprint 中文/特殊字符会 UnicodeEncodeError 崩溃。 3. 裸 requests.get 没有 User-Agentchinaz 有防盗链大概率被 403 拦掉。 4. 无脑 https: url遇到 http:// 开头的链接会变成坏地址相对路径也拼不对。 5. os.makedirs(pic) 写在 downloadpic() 调用之后第一次下载必然 FileNotFoundError。 6. 没有重试、没有超时兜底、没有去重、没有并发串行慢且不稳定。 7. 裸 except 会连 KeyboardInterruptCtrlC一起吞掉程序无法中断。 8. 抓到的是 40 张站点图标 缩略图把真正的风景图挤在后面因为没过滤站点静态资源。 9. 第 1 页地址规则写死range(2,3) 只能下第 2 页注释和实际行为不符。 【本版的改进】 · 结构规范功能拆分为 网络层 / 解析层 / 下载层 / CLI可单独调用和测试。 · 反爬健壮完整请求头 Retry 自动重试429/5xx Session 连接复用。 · 解析升级BeautifulSoup 正则双通道兼容 data-original / data-src / src 懒加载。 · 精准过滤剔除站点自身的 /static/ 图标资源只留真正的图片内容。 · 原图直出自动把缩略图 xxx_s.png 升级为原图 xxx.png体积约 10 倍失败自动回退。 · 并发下载线程池默认 6 线程比串行快一个数量级也可 -w 1 退回串行。 · 安全写入with open 自动关文件、目录先建后写、已存在则跳过支持断点续传。 · 全局连续编号跨页不重名、不覆盖失败的空号不影响后续编号。 · 命令行参数页码范围、保存目录、并发数、每页上限、礼貌延时均可配。 依赖pip install requests beautifulsoup4 用法示例 python chinaz_image_downloader.py # 下载第 1~3 页 python chinaz_image_downloader.py -s 1 -e 10 -w 8 # 下载第 1~10 页8 线程 python chinaz_image_downloader.py -e 2 --thumb # 只要缩略图快、省流量 python chinaz_image_downloader.py -e 20 -w 1 --delay 1 # 串行 延时最不容易被封 from __future__ import annotations import argparse import logging import os import re import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from urllib.parse import urljoin, urlparse import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry try: from bs4 import BeautifulSoup except ImportError: # 未装 bs4 也能用自动退化为纯正则解析 BeautifulSoup None # --------------------------------------------------------------------------- # 常量配置 # --------------------------------------------------------------------------- BASE_URL https://sc.chinaz.com/tupian/fengjingtupian.html # 第 1 页不带页码 PAGE_URL_TEMPLATE https://sc.chinaz.com/tupian/fengjingtupian_{page}.html HEADERS { User-Agent: ( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 ), Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/avif,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Referer: https://sc.chinaz.com/, # 站点有防盗链请求页面和图片都要带 Connection: keep-alive, } IMG_EXT {.jpg, .jpeg, .png, .webp, .bmp} # 命中这些特征的 URL 不是我们要的风景图站点图标、占位图、广告位等 SKIP_KEYWORDS (logo, icon, blank, placeholder, spacer, loading, avatar, ad_) SKIP_PATH_PARTS (/static/, /com_images/, /js/, /css/) # 缩略图后缀 - 去掉即得原图如 20260528142854701262_s.png - 20260528142854701262.png THUMB_SUFFIX re.compile(r_[sbm](?\.[a-z0-9]$), re.IGNORECASE) IMG_PATTERN re.compile( r(?:data-original|data-src|data-echo|src)\s*\s*[]([^]\.(?:jpg|jpeg|png|webp|bmp)[^]*)[], re.IGNORECASE, ) log logging.getLogger(img-downloader) def page_url(page: int) - str: 第 1 页是 .../fengjingtupian.html第 2 页起才是 ..._2.html。 return BASE_URL if page 1 else PAGE_URL_TEMPLATE.format(pagepage) # --------------------------------------------------------------------------- # 网络层 # --------------------------------------------------------------------------- def build_session(retries: int 3) - requests.Session: 创建带自动重试的 Session连接复用遇到 429/5xx 自动退避重试。 session requests.Session() session.headers.update(HEADERS) retry Retry( totalretries, connectretries, readretries, backoff_factor0.5, # 重试间隔 0.5s / 1s / 2s status_forcelist(429, 500, 502, 503, 504), allowed_methodsfrozenset([GET]), raise_on_statusFalse, ) adapter HTTPAdapter(max_retriesretry, pool_connections10, pool_maxsize10) session.mount(https://, adapter) session.mount(http://, adapter) return session def fetch_html(session: requests.Session, url: str, timeout: int 15) - str: 抓取页面 HTML并强制按检测到的编码解码避免 requests 猜成 ISO-8859-1 乱码。 resp session.get(url, timeouttimeout) resp.raise_for_status() resp.encoding resp.apparent_encoding or utf-8 return resp.text # --------------------------------------------------------------------------- # 解析层 # --------------------------------------------------------------------------- def normalize_url(raw: str, page_url: str) - str: 把 //cdn.x/a.jpg、/uploads/a.jpg、https://x/a.jpg 统一成绝对地址。 raw (raw or ).strip().strip(\) if not raw or raw.startswith(data:): return if raw.startswith(//): return https: raw return urljoin(page_url, raw) def is_wanted_image(url: str) - bool: 判断这个 URL 是否是我们真正想下载的图片。 if not url: return False path urlparse(url).path.lower() if os.path.splitext(path)[1] not in IMG_EXT: return False if any(part in path for part in SKIP_PATH_PARTS): # 站点自身的图标资源 return False if any(word in url.lower() for word in SKIP_KEYWORDS): return False return True def extract_image_urls(html: str, page_url: str) - list[str]: 从页面 HTML 中提取去重、过滤后的图片绝对地址保持页面原有顺序。 candidates: list[str] [] # 通道 ABeautifulSoup能正确处理属性顺序错乱、缺引号等情况 if BeautifulSoup is not None: soup BeautifulSoup(html, html.parser) for img in soup.find_all(img): # 懒加载站点把真图放在 data-originalsrc 只放占位图按优先级取第一个非空值 for attr in (data-original, data-src, data-echo, src): value img.get(attr) if value: candidates.append(value) break # 有些图的大图地址挂在父级 a href 上 parent img.find_parent(a) if parent and parent.get(href): candidates.append(parent[href]) # 通道 B正则兜底页面结构变了 / 没装 bs4 时仍能抓 if not candidates: candidates IMG_PATTERN.findall(html) seen: set[str] set() result: list[str] [] for raw in candidates: url normalize_url(raw, page_url) if url in seen: # 去重放在最前后面的校验失败也不会污染 seen continue seen.add(url) if is_wanted_image(url): result.append(url) return result def to_full_size_url(url: str) - str: 把缩略图地址升级为原图地址去掉 _s/_b/_m 后缀站长素材的命名规则。 path urlparse(url).path new_path THUMB_SUFFIX.sub(, path) return url.replace(path, new_path) if new_path ! path else url # --------------------------------------------------------------------------- # 下载层 # --------------------------------------------------------------------------- def download_one( session: requests.Session, url: str, save_path: Path, timeout: int 25, full_size: bool True, ) - bool: 下载单张图片成功返回 True。 if save_path.exists() and save_path.stat().st_size 0: log.info(已存在跳过%s, save_path.name) return True # 先试原图失败再回退到传入的缩略图地址 candidates [to_full_size_url(url), url] if full_size else [url] candidates list(dict.fromkeys(candidates)) # 去重且保持顺序 for candidate in candidates: try: # 图片请求也带 Referer否则可能被防盗链拦成一张占位图 resp session.get(candidate, timeouttimeout, headers{Referer: url}) if resp.status_code ! 200 or not resp.content: continue save_path.write_bytes(resp.content) tag 原图 if candidate ! url else 缩略 log.info([OK] %-16s %s %7.1f KB %s, save_path.name, tag, len(resp.content) / 1024, candidate) return True except requests.RequestException as exc: log.debug(候选地址失败 %s - %s, candidate, exc) continue log.warning([FAIL] %s, url) return False def download_images( session: requests.Session, urls: list[str], out_dir: Path, start_index: int, workers: int 6, delay: float 0.0, full_size: bool True, ) - int: 下载一批图片返回成功数量文件名从 start_index 起全局连续编号。 tasks: list[tuple[str, Path]] [] for offset, url in enumerate(urls): ext os.path.splitext(urlparse(url).path)[1].lower() or .jpg tasks.append((url, out_dir / f{start_index offset:04d}{ext})) ok 0 if workers 1: # 串行模式慢但最不容易触发风控 for url, path in tasks: if download_one(session, url, path, full_sizefull_size): ok 1 if delay: time.sleep(delay) else: with ThreadPoolExecutor(max_workersworkers) as pool: futures [pool.submit(download_one, session, url, path, 25, full_size) for url, path in tasks] for future in as_completed(futures): if future.result(): ok 1 return ok # --------------------------------------------------------------------------- # 主流程 # --------------------------------------------------------------------------- def parse_args(argv: list[str] | None None) - argparse.Namespace: parser argparse.ArgumentParser(description站长素材风景图片批量下载器) parser.add_argument(-s, --start-page, typeint, default1, help起始页码默认 1) parser.add_argument(-e, --end-page, typeint, default2, help结束页码含默认 3) parser.add_argument(-o, --out-dir, defaultpic, help保存目录默认 pic) parser.add_argument(-w, --workers, typeint, default6, help并发线程数1 为串行默认 6) parser.add_argument(-l, --limit, typeint, default0, help每页最多下载张数0 为不限默认 0) parser.add_argument(--delay, typefloat, default0.3, help页面之间的间隔秒数默认 0.3) parser.add_argument(--thumb, actionstore_true, help只下缩略图更快、更省流量) return parser.parse_args(argv) def main(argv: list[str] | None None) - int: args parse_args(argv) # Windows 控制台默认 GBK中文/符号可能直接抛 UnicodeEncodeError这里强制 UTF-8 for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encodingutf-8, errorsreplace) except (AttributeError, ValueError): pass logging.basicConfig(levellogging.INFO, format%(message)s, streamsys.stdout) if args.end_page args.start_page: log.error(结束页码不能小于起始页码) return 2 # 目录必须先建好原版是先下载后建目录第一次调用必定失败 out_dir Path(args.out_dir) out_dir.mkdir(parentsTrue, exist_okTrue) session build_session() total_found total_ok 0 next_index 1 # 全局连续编号跨页不覆盖 for page in range(args.start_page, args.end_page 1): current_url page_url(page) log.info( * 66) log.info(正在解析第 %s 页%s, page, current_url) try: html fetch_html(session, current_url) except requests.RequestException as exc: log.error(页面抓取失败跳过%s, exc) continue urls extract_image_urls(html, current_url) if not urls: log.warning(本页未解析到图片页面结构可能已变化或被反爬拦截) continue if args.limit 0: urls urls[: args.limit] log.info(本页发现 %d 张开始下载……, len(urls)) total_found len(urls) total_ok download_images( session, urls, out_dir, next_index, args.workers, args.delay, not args.thumb ) next_index len(urls) # 成功或失败都推进编号避免同一页内互相覆盖 if args.delay and page ! args.end_page: time.sleep(args.delay) log.info( * 66) log.info(全部完成发现 %d 张成功 %d 张保存目录%s, total_found, total_ok, out_dir.resolve()) return 0 if total_ok else 1 if __name__ __main__: sys.exit(main())