前段时间我写了一个爬取网站的图片的python程序，代码如下：

#coding=utf-8#支持本程序中有汉字，否则报错 import re#导入正则表达式模块 import requests#导入网络请求,如果没有安装命令：pip install requests import os #url='http://www.netbian.com/'#图片网址 #url='http://www.k2r2.com/' url='https://sc.chinaz.com/tupian/fengjingtupian.html' #定义变量n，并且赋值为1 n=1 def downloadpic(): data=requests.get(url).text#将网页的源代码写入变量data中 #jpglist=re.findall('data-original="(.*?)"',data,re.S)#从data文本中查找以data-original=开头和以空格结束的中间的字符串给列表jpglist，re.S不会对\n进行中段 jpglist=re.findall('data-original="(.*?)"',data,re.S)#从data文本中查找以data-original=开头和以空格结束的中间的字符串给列表jpglist，re.S不会对\n进行中段 global n for each in jpglist:#遍历列表jpglist new=each.strip() new1=new.replace('_s','') each='https:'+new1 print(each)#输出列表对应的元素 try:#尝试执行 pic=requests.get(each,timeout=10)#将请求响应结果赋值给pic except:#尝试失败时 print('下载失败')#输出下载失败 continue#继续循环 string='pic\\'+str(n)+'.jpg'#构造文件路径和文件名 fp=open(string,'wb')#创建文件对象fp fp.write(pic.content)#将pic图片对象以二进制形式写入文件 fp.close()#关闭文件对象fp n+=1#变量n自加1 if not os.path.exists('pic'): os.makedirs('pic') downloadpic() url1='https://sc.chinaz.com/tupian/fengjingtupian_' for i in range(2,3): url=url1+str(i)+'.html' downloadpic() print('下载完成！')

这个程序虽然能够正常运行，但我感觉可能会存在逐多问题，于是，我将代码发给了workbuddy，让它给我优化和完善，经过一段时间，它给我优化完善后的代码是：

# -*- coding: utf-8 -*- """ 站长素材（sc.chinaz.com）风景图片批量下载器 —— 优化完善版 ================================================================ 【原版的问题】 1. 缩进全丢，粘贴后所有语句挤在一行，直接 SyntaxError 无法运行。 2. Windows 控制台默认 GBK，print 中文/特殊字符会 UnicodeEncodeError 崩溃。 3. 裸 requests.get 没有 User-Agent，chinaz 有防盗链，大概率被 403 拦掉。 4. 无脑 'https:' + url：遇到 http:// 开头的链接会变成坏地址，相对路径也拼不对。 5. os.makedirs('pic') 写在 downloadpic() 调用之后，第一次下载必然 FileNotFoundError。 6. 没有重试、没有超时兜底、没有去重、没有并发，串行慢且不稳定。 7. 裸 except 会连 KeyboardInterrupt（Ctrl+C）一起吞掉，程序无法中断。 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/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.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(page=page) # --------------------------------------------------------------------------- # 网络层 # --------------------------------------------------------------------------- def build_session(retries: int = 3) -> requests.Session: """创建带自动重试的 Session：连接复用，遇到 429/5xx 自动退避重试。""" session = requests.Session() session.headers.update(HEADERS) retry = Retry( total=retries, connect=retries, read=retries, backoff_factor=0.5, # 重试间隔 0.5s / 1s / 2s status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(["GET"]), raise_on_status=False, ) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10) 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, timeout=timeout) 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] = [] # 通道 A：BeautifulSoup，能正确处理属性顺序错乱、缺引号等情况 if BeautifulSoup is not None: soup = BeautifulSoup(html, "html.parser") for img in soup.find_all("img"): # 懒加载站点把真图放在 data-original，src 只放占位图，按优先级取第一个非空值 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, timeout=timeout, 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_size=full_size): ok += 1 if delay: time.sleep(delay) else: with ThreadPoolExecutor(max_workers=workers) 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", type=int, default=1, help="起始页码，默认 1") parser.add_argument("-e", "--end-page", type=int, default=2, help="结束页码（含），默认 3") parser.add_argument("-o", "--out-dir", default="pic", help="保存目录，默认 pic") parser.add_argument("-w", "--workers", type=int, default=6, help="并发线程数，1 为串行，默认 6") parser.add_argument("-l", "--limit", type=int, default=0, help="每页最多下载张数，0 为不限，默认 0") parser.add_argument("--delay", type=float, default=0.3, help="页面之间的间隔秒数，默认 0.3") parser.add_argument("--thumb", action="store_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(encoding="utf-8", errors="replace") except (AttributeError, ValueError): pass logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout) if args.end_page < args.start_page: log.error("结束页码不能小于起始页码") return 2 # 目录必须先建好（原版是先下载后建目录，第一次调用必定失败） out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) 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())