简介：URL文件批量下载器是一款面向内容创作者、数据分析师与网络管理员的实用工具，用于解决从各类网站批量获取图片、文档、音频等网络资源时逐个点击、效率低下的问题。资源包共5个文件，包含可执行程序、动态链接库、说明文档与示例URL列表等，涵盖exe主程序、dll运行依赖、htm与docx使用说明以及txt链接清单，压缩包整体约1.22MB，体积轻巧、解压即用。工具围绕URL解析与批量任务调度展开，支持导入链接列表、设置下载目录与线程参数、查看进度与日志，并具备断点续传与失败重试等排错思路，便于读者理解HTTP/HTTPS协议与多线程下载的基本原理。目前已有2526人学习下载，适合需要频繁采集网络素材、希望提升资源获取效率的初中级用户参考使用。
1. URL文件批量下载器：从手工点击到脚本化落地的分水岭
手里攥着几百条 URL，每条指向一个 PDF、一张图片或者一个压缩包，浏览器里一个个点开、另存为、改文件名——这件事我干过整整一个下午，最后手指发酸，还漏了十几条。URL 文件批量下载器要解决的就是这个场景：把一份 URL 清单变成本地一堆命名规整的文件。它适合做数据采集、素材归档、报表拉取、模型训练集准备的人，也适合任何需要把「链接列表」变成「本地目录」的从业者。核心诉求只有三个：并发要可控、失败要能重试、文件名要能自己定。下面按「先跑通最小版本，再补并发和断点续传，最后处理反爬和异常」的路径拆开讲，代码可以直接抄。
2. 最小可用下载器：20 行代码把 URL 清单变成文件
2.1 为什么先用 requests 而不是 wget 或 curl
命令行工具 wget 和 curl 确实能批量下载，wget -i urls.txt一行就完事。但一旦遇到这几种情况就会卡住：需要自定义请求头、需要根据响应内容动态命名文件、需要跳过已下载的、需要把失败 URL 单独记录。这些在 shell 里写起来很别扭，而 Python 的 requests 库把这些控制权全部交回给你。常见做法是用 requests + os + 一个 for 循环起步，跑通之后再引入并发。
选 requests 而不是 aiohttp 的原因是：起步阶段并发不是瓶颈，代码可读性更重要。aiohttp 的 async/await 写法对新手不友好，调试也麻烦。等单线程跑通了、确认瓶颈确实在网络等待上，再换并发方案也不迟。
2.2 最小可运行脚本与逐行说明
import os import requests from urllib.parse import urlparse, unquote # URL 清单文件，每行一条 URL_FILE = "urls.txt" # 下载目录 OUT_DIR = "downloads" # 超时时间（连接+读取），单位秒 TIMEOUT = 30 # 自定义请求头，很多站点缺 UA 会直接 403 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" } os.makedirs(OUT_DIR, exist_ok=True) with open(URL_FILE, "r", encoding="utf-8") as f: urls = [line.strip() for line in f if line.strip()] for idx, url in enumerate(urls, 1): try: resp = requests.get(url, headers=HEADERS, timeout=TIMEOUT, stream=True) resp.raise_for_status() # 非 2xx 直接抛异常 # 从 URL 路径提取文件名，unquote 处理中文和空格 path = urlparse(url).path filename = unquote(os.path.basename(path)) or f"file_{idx}" filepath = os.path.join(OUT_DIR, filename) with open(filepath, "wb") as out: for chunk in resp.iter_content(chunk_size=8192): out.write(chunk) print(f"[{idx}/{len(urls)}] OK {filename}") except Exception as e: print(f"[{idx}/{len(urls)}] FAIL {url} -> {e}")逻辑说明：stream=True让 requests 不把整个响应体一次性读进内存，而是分块读取，这对下载大文件至关重要。iter_content(chunk_size=8192)每次读 8KB 写入磁盘，内存占用恒定。raise_for_status()把 4xx/5xx 变成异常，避免把错误页面当成文件存下来。
参数说明：TIMEOUT=30是连接和读取各自 30 秒，不是总时长。如果目标站点响应慢，可以调到 60。chunk_size一般 8192 或 16384 都行，再大对吞吐提升有限。HEADERS里的 UA 建议换成自己浏览器的真实 UA，不要用默认的 python-requests，很多站点会直接拒绝。
2.3 文件名冲突和非法字符怎么处理
上面的脚本有个明显问题：不同 URL 可能指向同名文件，后下载的会覆盖先下载的。另外 URL 里可能带?、&、%等字符，直接当文件名在 Windows 上会报错。常见做法是加一个序号前缀，同时过滤非法字符：
import re def safe_filename(name: str, idx: int) -> str: # 去掉 Windows/Linux 下的非法字符 name = re.sub(r'[\\/:*?"<>|]', "_", name) # 限制长度，避免超过文件系统上限 name = name[:150] return f"{idx:04d}_{name}"把filename = unquote(...)那行替换成filename = safe_filename(unquote(os.path.basename(path)) or f"file_{idx}", idx)，就能保证每个文件都有唯一前缀，且不会因为特殊字符翻车。序号用四位补零，方便后续按顺序排列。
3. 并发下载：把等待时间压到十分之一
3.1 并发模型选型：线程池还是异步 IO
单线程下载 500 个文件，如果每个平均耗时 2 秒，总共要 1000 秒，接近 17 分钟。但其中绝大部分时间花在等待网络响应上，CPU 几乎空闲。这时候并发就能把总时间压到接近最慢那个文件的时间。
Python 里做并发下载有两条路：concurrent.futures.ThreadPoolExecutor和asyncio + aiohttp。线程池的优势是代码改动小，把原来的 for 循环体丢进函数，用executor.map跑就行；劣势是线程数不能开太大，否则上下文切换开销反而拖慢速度。aiohttp 的优势是单线程事件循环，可以轻松开几百个并发连接；劣势是代码要重写成 async 风格，且遇到同步阻塞操作会卡住整个事件循环。
我一般会先用线程池，并发数设 8 到 16。如果目标站点对单 IP 有连接数限制，线程池反而更安全，因为你可以精确控制并发数。aiohttp 适合目标站点分散、单站点压力不大的场景。
3.2 线程池版本：改造最小脚本
from concurrent.futures import ThreadPoolExecutor, as_completed MAX_WORKERS = 8 # 并发数，建议 4~16 之间 def download_one(idx_url): idx, url = idx_url try: resp = requests.get(url, headers=HEADERS, timeout=TIMEOUT, stream=True) resp.raise_for_status() path = urlparse(url).path filename = safe_filename(unquote(os.path.basename(path)) or f"file_{idx}", idx) filepath = os.path.join(OUT_DIR, filename) with open(filepath, "wb") as out: for chunk in resp.iter_content(chunk_size=8192): out.write(chunk) return (idx, url, "OK", filename) except Exception as e: return (idx, url, "FAIL", str(e)) with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = {pool.submit(download_one, (i, u)): u for i, u in enumerate(urls, 1)} for fut in as_completed(futures): idx, url, status, msg = fut.result() print(f"[{idx}] {status} {msg}")逻辑说明：as_completed按完成顺序返回结果，而不是按提交顺序，这样先下完的先打印，不会因为某个慢文件卡住输出。每个任务独立捕获异常，一个失败不影响其他任务。
参数说明：MAX_WORKERS=8是经验值。如果目标站点是同一个域名，建议不要超过 16，否则容易被限流甚至封 IP。如果 URL 分散在多个域名，可以适当调高到 32。注意线程池的并发数不等于同时打开的连接数，requests 底层用 urllib3 的连接池，每个线程会复用连接。
3.3 断点续传：已下载的文件怎么跳过
批量下载最怕跑到一半断了，重新跑一遍又要从头开始。解决办法有两个层次：一是下载前检查文件是否已存在且大小合理，二是用 HTTP Range 头做真正的断点续传。
第一个层次最简单：在download_one开头加一段判断，如果目标文件已存在且大小大于 0，直接返回 SKIP。这能覆盖 90% 的场景，因为大部分文件要么完整下载了，要么根本没开始。
第二个层次需要服务端支持Accept-Ranges: bytes。先发一个 HEAD 请求拿到Content-Length，再检查本地文件大小，用Range: bytes={local_size}-续传。不是所有服务器都支持，所以要做好降级：如果返回 200 而不是 206，说明服务端不支持 Range，只能重新下载。
def download_with_resume(idx, url): path = urlparse(url).path filename = safe_filename(unquote(os.path.basename(path)) or f"file_{idx}", idx) filepath = os.path.join(OUT_DIR, filename) # 已存在且非空，跳过 if os.path.exists(filepath) and os.path.getsize(filepath) > 0: return (idx, url, "SKIP", filename) headers = dict(HEADERS) local_size = 0 if os.path.exists(filepath): local_size = os.path.getsize(filepath) headers["Range"] = f"bytes={local_size}-" resp = requests.get(url, headers=headers, timeout=TIMEOUT, stream=True) if resp.status_code == 416: # Range 超出范围，说明已下完 return (idx, url, "SKIP", filename) resp.raise_for_status() mode = "ab" if local_size and resp.status_code == 206 else "wb" with open(filepath, mode) as out: for chunk in resp.iter_content(chunk_size=8192): out.write(chunk) return (idx, url, "OK", filename)注意mode = "ab"只在服务端返回 206 时才用追加模式，否则会得到一个损坏的文件。这个判断不能省，否则遇到不支持 Range 的服务器，文件会变成「半截旧数据 + 完整新数据」的拼接体，打开就报错。
4. 避坑与排查：批量下载最常见的 5 个翻车现场
4.1 现象：下载下来的全是 HTML 文件，不是目标文件
原因：目标站点有反爬或登录校验，未登录请求被重定向到登录页或验证页，返回 200 但内容是 HTML。requests 默认跟随重定向，所以不会报错。
解决：下载前先检查Content-Type响应头，如果不是预期的类型（比如application/pdf、image/jpeg），就跳过并记录。另外可以在请求头里带上Referer，很多站点会校验来源页。
4.2 现象：并发开到 32，结果大量超时和连接被拒
原因：目标站点对单 IP 的并发连接数有限制，超过阈值后直接拒绝新连接。或者本地网络出口带宽被占满，导致所有请求都变慢。
解决：把并发数降到 8 以下，并在每次请求之间加随机延迟。可以用time.sleep(random.uniform(0.1, 0.5))在任务开头加抖动。如果目标站点有Retry-After响应头，要尊重它，按指定时间等待后重试。
4.3 现象：文件名乱码，中文变成百分号编码
原因：URL 里的中文和特殊字符经过 percent-encoding，直接取os.path.basename拿到的是%E6%96%87%E4%BB%B6.pdf这种形式。
解决：用urllib.parse.unquote解码。但要注意，有些 URL 是双重编码的，解码一次后还是百分号形式，需要再解一次。判断方法是解码后如果还包含%且后面跟两位十六进制，就再解一次。
4.4 现象：下载到一半程序崩溃，重启后文件损坏
原因：写入过程中程序被中断，文件只写了一部分，但下次运行时因为文件已存在且大小大于 0，被 SKIP 逻辑跳过了。
解决：不要用「文件存在就跳过」这种粗粒度判断。正确做法是下载时先写临时文件（比如.part后缀），下载完成后用os.rename原子性地改成正式文件名。这样即使中途崩溃，临时文件不会被误认为完整文件。重启后清理所有.part文件重新下载即可。
4.5 现象：某些 URL 返回 403，但浏览器里能正常打开
原因：服务端校验了User-Agent、Referer、Cookie或Accept-Language等请求头，而脚本里没带全。
解决：用浏览器开发者工具的 Network 面板，找到该请求的完整请求头，复制到脚本的HEADERS里。重点是User-Agent、Referer、Accept、Accept-Language这四个。如果还不行，可能需要带 Cookie，那就得先从浏览器登录后导出 Cookie 再注入。
5. 进阶技巧：用 URL 指纹做增量下载和去重
批量下载做多了会发现一个需求：同一批 URL 隔几天再跑一次，只想下载新增的或内容变化的文件。这时候光靠文件名判断不够，因为文件名可能相同但内容已经更新。我的习惯是给每个 URL 算一个指纹，存到本地 SQLite 里，每次下载前先比对。
指纹可以用 URL 的 MD5 加上响应头的ETag或Last-Modified组合。如果服务端支持ETag，直接用它做版本判断最准。不支持的话，退而求其次用Content-Length加Last-Modified。下面是一个最小实现：
import sqlite3 import hashlib DB = "download_history.db" def init_db(): conn = sqlite3.connect(DB) conn.execute(""" CREATE TABLE IF NOT EXISTS downloads ( url_hash TEXT PRIMARY KEY, url TEXT, etag TEXT, last_modified TEXT, filepath TEXT, downloaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() return conn def url_fingerprint(url: str) -> str: return hashlib.md5(url.encode("utf-8")).hexdigest() def should_download(conn, url, etag, last_modified): h = url_fingerprint(url) row = conn.execute( "SELECT etag, last_modified FROM downloads WHERE url_hash=?", (h,) ).fetchone() if row is None: return True old_etag, old_lm = row if etag and old_etag and etag != old_etag: return True if last_modified and old_lm and last_modified != old_lm: return True return False逻辑说明：先用 HEAD 请求拿到ETag和Last-Modified，再查本地数据库比对。如果都没有变化，直接跳过，省掉一次完整下载。下载完成后把新的ETag和Last-Modified写回数据库。
参数说明：url_hash用 MD5 而不是直接存 URL，是因为有些 URL 很长，建索引效率低。MD5 碰撞概率在这个场景下可以忽略。downloaded_at字段方便后续做「最近 7 天新增」之类的统计。
这个方案还有一个附带好处：如果某个 URL 下载失败了，数据库里没有记录，下次跑的时候会自动重试，不需要额外维护失败列表。我一般会在下载成功后才写数据库，失败的不写，这样失败任务天然就是「待重试」状态。
最后一个习惯：每次批量下载前，先用head请求快速探测所有 URL 的可用性，把 404 和 403 的挑出来单独处理，不要让它们混在下载队列里浪费并发槽位。这个预检步骤花不了几秒钟，但能省下大量无效等待。希望帮到你。
本文还有配套的精品资源，点击获取
