尧图网页设计 
我手里正好有个给运营同事做的批量素材抓取小工具场景就是输入一批网页自动把页面里所有PDF、图片、压缩包下载到本地。做的时候发现网上聊“爬虫下载链接”的文章不少但大多是给个requests.get完事真正能把“自动下载网页链接”这件事做扎实的没几篇。这篇文章就把我做这个小工具的完整思路写下来。核心围绕三件事第一怎么从网页里精准提取出真正需要下载的链接第二怎么把批量下载做得又快又稳不因为几个坏链接就让整个任务挂掉第三怎么处理并发、反爬、断点续传这些工程化问题。不管你只是想抓个图集还是要定期同步某类资源文件这篇都能给你一条能直接落地的路线。1. 接到“自动下载网页链接”需求时先拆解再做很多新手拿到这个需求第一反应是“我直接requests然后正则匹配href就好了”。但实际接到任务尤其是给别人做工具的时候必须先搞清楚几个问题否则做出来就是个一次性脚本换个网站就废。1.1 这需求到底在说什么提取链接和下载文件是两码事“自动下载网页链接”这句话本身有歧义。它至少包含两种完全不同的场景场景A把网页里所有链接抓下来保存成清单。这种场景的核心是“提取”比如SEO分析、外链收集输出是一个文本文件或表格里面是URL列表。场景B把网页里指向具体文件的链接下载到本地。比如图片、PDF、ZIP、视频核心是“下载”要根据链接创建本地目录结构还要处理文件重名、下载失败重试。我的经验是直接把两种场景都做了做成一个可配置的工具。反正提取是下载的前置步骤代码不会多太多但适用范围会广很多。1.2 明确目标要下载哪些链接跳过哪些链接这是整件事最关键的一步也是最容易被忽略的。正则匹配a href...确实能把链接抓出来但抓出来的大部分都没用。我通常会先问自己几个问题目标网页是静态HTML还是需要渲染JS才能拿到完整内容这决定了用requests直接请求还是上Playwright。需要下载的文件类型有哪些PDF、图片、压缩包、还是附件里的docx是只下载当前页面里的直接链接还是需要递归爬取子页面里的文件链接下载的文件重名了怎么处理覆盖、重命名还是跳过需不需要记录下载历史下次增量同步这些问题想清楚了具体写代码的时候就会很顺。以我这次做的工具为例目标是爬取一组资讯页里的PDF附件和配图文件按原链接文件名保存到本地已存在的自动跳过。要求就是稳定跑几百个页面不能因为几个超时链接就中断。1.3 整体流程设计整个工具的运行链路是固定的后面所有代码都是围绕这条链路展开读取URL列表 → 请求页面HTML → 解析并提取所有候选链接 → 过滤出目标文件链接 → 下载文件 → 保存到本地目录这套链路看着简单但每一步都有不少隐藏问题。下面几节拆开逐个说。2. 链接提取为什么不用“万能”正则而用DOM解析提取链接的方案我见过有人直接用re.findall(rhref([^]*))一把梭。说实话这方法在简单页面上确实能用但踩过几次坑之后我就再也不敢用了。2.1 正则方案的三个致命问题第一HTML属性值不一定用双引号还可能是单引号甚至无引号正则写起来又长又容易漏。第二页面里有些href的值不是完整URL而是相对路径比如/files/report.pdf你直接拼接而不是解析BaseURI的话下载就是404。第三有些链接是javascript:void(0)、mailto:、#这种正则一股脑全抓出来后面还要再过滤几层。最恶心的一个案例是有个页面里的PDF链接是通过JS变量拼接出来的a hrefjavascript:download(report.pdf)这种情况纯正则根本提取不到真实地址。所以后来我的基本思路是能用DOM解析就用DOM解析只有在HTML结构极其规整、且确认没有JS生成内容的时候才考虑正则作为快速原型方案。2.2 用lxmlparsel提取链接代码简短且稳这里我推荐用parsel它是Scrapy底层用的解析库内部基于lxmlAPI非常简洁对新手也友好。下面这段就是我工具里实际用的提取函数import parsel def extract_links(html: str, base_url: str): 从HTML中提取所有a标签和img标签的链接 html: 页面源码 base_url: 用于拼接相对URL的基础地址一般是当前页面的URL 返回: 去重后的URL列表 sel parsel.Selector(texthtml) # 提取a标签href和img标签src raw_links sel.xpath(//a/href | //img/src).extract() absolute_links [] for raw in raw_links: if not raw or raw.strip() : continue # 跳过纯锚点、脚本调用、邮件协议 if raw.startswith(#) or raw.startswith(javascript:) or raw.startswith(mailto:): continue # 用urljoin把相对路径拼成绝对地址 absolute urljoin(base_url, raw.strip()) absolute_links.append(absolute) # 去重并保持顺序 seen set() unique_links [] for link in absolute_links: if link not in seen: seen.add(link) unique_links.append(link) return unique_links这里两个小要点urljoin不光能处理/files/a.pdf这种相对路径还能处理../files/a.pdf这种带父级跳转的比直接字符串拼接靠谱得多。只抓a和img标签通常够用如果网站有些文件是放在video标签的src或者source标签里可以按需加XPath表达式比如//video/src | //source/src。2.3 文件链接过滤只留我们真正要下载的东西链接提取出来之后真正需要下载的往往是特定后缀的地址。我的做法是维护一个后缀白名单命中才下载DOWNLOAD_EXTENSIONS { .pdf, .doc, .docx, .xls, .xlsx, .ppt, .pptx, .zip, .rar, .7z, .tar, .gz, .jpg, .jpeg, .png, .gif, .webp, .bmp, .mp4, .avi, .mkv, .mov, } def filter_downloadable(links, allowed_extsDOWNLOAD_EXTENSIONS): 从候选链接里过滤出可下载的文件链接同时去重 result [] seen set() for url in links: # 去掉URL的查询参数再取后缀 clean_url url.split(?)[0].split(#)[0] ext os.path.splitext(clean_url)[1].lower() if ext in allowed_exts and url not in seen: seen.add(url) result.append(url) return result注意这里有个容易被坑的点有些文件链接不是以真实后缀结尾的比如https://example.com/download?id123这种返回的其实是文件流。应对办法是先把链接下载下来或发HEAD请求看Content-Type或Content-Disposition里的文件名但这会增加很多请求开销。我一般是在工具里加一个开关默认走后缀白名单需要的时候手动打开“嗅探模式”用HEAD请求判断是否目标类型。2.4 动态渲染页面怎么处理如果目标页面是SPA单页应用比如Vue、React写的站直接requests拿到的HTML里根本没有完整链接因为数据是JS异步加载的。这种情况有两种解法简单场景直接看接口。打开浏览器F12Network里找XHR请求很可能直接请求一个JSON接口就能拿到带文件地址的数据比爬HTML还省事。复杂场景上无头浏览器。用Playwright或Selenium渲染完页面再提取链接。我这边的原则是能走接口就走接口实在不行才上无头浏览器。因为无头浏览器吃内存并发也不好控制维护成本明显更高。这篇不展开Playwright重点还在完整下载链路的稳定性上。3. 批量下载从这里开始才叫真正的工程化链接提取完下载这一步才是决定工具好不好用的分水岭。很多人写的下载代码是这样的import requests for url in link_list: r requests.get(url) with open(os.path.basename(url), wb) as f: f.write(r.content)这代码单跑一个链接没问题一旦放到真实场景就各种翻车文件名带非法字符导致保存失败、r.content一次性把大文件全读进内存、某次请求超时就导致整个程序崩溃、下载下来的文件打开发现是404页面……这些问题我在实际开发里全都踩过下面直接给一份能用于真实场景的下载代码。3.1 一份能直接用在大流量场景的稳健下载函数import os import time import requests from urllib.parse import unquote def safe_filename(url, defaultdownload.bin): 根据URL生成安全的本地文件名过滤掉windows非法字符 # 取URL最后一节并把URL编码还原为原始文件名 path url.split(?)[0] filename os.path.basename(path) filename unquote(filename) if not filename: filename default # 替换windows/linux下不允许出现在文件名里的字符 filename re.sub(r[\\/:*?|], _, filename) # 限制长度防止超出文件系统上限 if len(filename) 200: name, ext os.path.splitext(filename) filename name[:180] ext return filename def download_file(url, save_path, timeout30, retries3): 下载单个文件支持重试和流式写入 url: 文件地址 save_path: 本地保存完整路径 timeout: 单次请求超时时间 retries: 失败重试次数 返回: (是否成功, 保存路径或错误信息) for attempt in range(retries): try: with requests.get(url, streamTrue, timeouttimeout) as r: if r.status_code ! 200: print(f[{attempt 1}/{retries}] HTTP {r.status_code}: {url}) time.sleep(2 * (attempt 1)) continue # Header里的文件大小字节用于下载进度判断 total_size int(r.headers.get(Content-Length, 0)) downloaded 0 # 生成临时文件名避免下载到一半的残缺文件占用正式文件名 tmp_path save_path .part with open(tmp_path, wb) as f: for chunk in r.iter_content(chunk_size8192): if chunk: f.write(chunk) downloaded len(chunk) # 如果服务器给了Content-Length做下大小校验 if total_size 0 and downloaded ! total_size: raise IOError(f文件大小不匹配: 期望{total_size}, 实际{downloaded}) os.replace(tmp_path, save_path) return True, save_path except Exception as e: print(f[{attempt 1}/{retries}] 下载失败: {url}, 错误: {e}) if attempt retries - 1: time.sleep(2 * (attempt 1)) return False, str(e) if e in locals() else 未知错误几个设计点值得说明流式下载iter_content(chunk_size8192)逐块写入磁盘避免大文件撑爆内存。直接r.content会把整个文件读进内存下载一个2GB的压缩包的时候机器基本就卡死了。临时文件下载完再改名这是从生产环境学到的经验。如果下载到一半网络断了残留在磁盘上的xxx.pdf会让人误以为文件已下载成功。用.part后缀先写下载完再os.replace进程挂掉也不会有“假成功”的文件。Content-Length校验虽然不一定每次都有这个Header但有的话就做一次大小核对能拦截一大半“下载下来是HTML错误页”的情况。3.2 主循环带历史记录和断点续传的批量下载单文件下载没问题之后就要把整个链路串起来。这里我加了一个“已下载历史记录”机制作用有两个一是避免重复下载相同文件浪费流量二是万一程序中断了下次启动能直接跳过已经完成的文件效果上就是断点续传。import json import os from pathlib import Path class DownloadManager: def __init__(self, history_filedownload_history.json): self.history_file history_file self.history self._load_history() def _load_history(self): if os.path.exists(self.history_file): with open(self.history_file, r, encodingutf-8) as f: return set(json.load(f)) return set() def save_history(self): with open(self.history_file, w, encodingutf-8) as f: json.dump(list(self.history), f, ensure_asciiFalse) def run(self, urls, save_dir): save_dir Path(save_dir) save_dir.mkdir(parentsTrue, exist_okTrue) results [] for idx, url in enumerate(urls, 1): filename safe_filename(url) save_path save_dir / filename # 历史记录里有这条URL且文件还存在直接跳过 if url in self.history and save_path.exists(): print(f[{idx}/{len(urls)}] 跳过: {filename}) results.append((url, True, skipped)) continue ok, msg download_file(url, str(save_path)) if ok: self.history.add(url) results.append((url, True, msg)) else: results.append((url, False, msg)) # 每下载完一个就保存一次历史防止进程崩溃丢进度 if idx % 10 0 or not ok: self.save_history() self.save_history() return results这段代码解决的问题是下载任务的“可重启性”。我之前给一个同事做素材批次下载跑了半小时断网了没有断点续传的话要从头再来一遍有了历史记录之后就只看没下完的那几个省下的时间非常可观。3.3 目录结构怎么组织如果下载任务涉及多个来源页面我建议按“页面URL”维度来分目录。比如爬的是一个列表页、详情页混在一起的内容目录结构可以是这样downloads/ ├── page_001/ │ ├── report_2024.pdf │ └── cover.jpg └── page_002/ ├── attachment.zip └── image1.png实现上就是在run()里根据当前解析的是哪个URL把save_dir继续往下一层建子目录。目录层级多了之后有一个细节要注意Windows下目录名也不能带:\/?这些字符所以目录名也要用safe_filename清洗一遍。4. 并发下载串行太慢但并发设计也有讲究第一次用串行方式跑一百个文件如果每个文件3秒差不多五分钟。看起来还行但文件数量到几百上千的时候串行就完全不可接受了。这时候就要上并发。4.1 三种常见并发方案怎么选Python里做下载并发常用三种方案我对它们的评价如下方案实现方式适用场景上手难度多线程concurrent.futures.ThreadPoolExecutorIO密集型任务简单直接推荐新手低协程asyncioaiohttp并发量极大、追求极致吞吐中高多进程multiprocessing.PoolCPU密集型比如同时做压缩/解析下载场景不推荐中下载文件主要是IO瓶颈不是CPU瓶颈所以线程池就够了。协程的上限确实高但调试和排错难度也高非必要不上。多进程在下载场景里没什么优势进程切换开销大还占内存唯一好处是能绕过GIL但下载本来就不怎么吃CPUGIL不是瓶颈。4.2 用ThreadPoolExecutor实现并发下载改造起来非常简单核心逻辑不变只是把主循环的任务丢给线程池from concurrent.futures import ThreadPoolExecutor, as_completed def batch_download(urls, save_dir, max_workers8, managerNone): 并发下载任务 urls: 文件URL列表 save_dir: 保存目录 max_workers: 并发线程数 manager: DownloadManager实例用于记录历史 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_url { executor.submit(download_file, url, str(Path(save_dir) / safe_filename(url)), 30, 3): url for url in urls } for future in as_completed(future_to_url): url future_to_url[future] try: ok, msg future.result() results.append((url, ok, msg)) print(f{OK if ok else FAIL} {url} - {msg}) except Exception as e: results.append((url, False, str(e))) print(f异常 {url}: {e}) return results创建线程池的时候max_workers不建议拍脑袋。目标网站的服务器能力、当前网络带宽、你不想被对方封IP这三个因素共同决定合理值。我的经验是8个线程起步最多不要超过16个。线程太多短时间内请求过于密集很容易触发目标站的反爬策略。曾经有一次我把max_workers调成50跑一个小网站的图集跑了不到两分钟IP就被临时限制了返回的全是403。后来老老实实降回10个线程把请求间隔稍微调大反而更稳定。4.3 并发下载的请求间距控制多线程并发会让请求间隔不可控所以最好自己做一个简单的“信号量限速”。用threading.Semaphore控制同时进行的下载数只是控制并发如果想控制“每秒最多发N个请求”可以维护一个带时间戳的计数器。这里给一个轻量方案import threading import time class RateLimiter: 简单的每秒请求数限制器 def __init__(self, max_per_second5): self.interval 1.0 / max_per_second self.lock threading.Lock() self.next_time time.time() def wait(self): with self.lock: now time.time() wait_time self.next_time - now if wait_time 0: time.sleep(wait_time) # 下一次请求最早允许的时间 self.next_time max(now, self.next_time) self.interval在download_file里请求前调用rate_limiter.wait()就能把带宽占用控制得比较均匀。单看一个请求只多等了几十毫秒但对上百个请求来说这种平滑限速能让目标服务器更“舒服”不容易触发风控。5. 让下载流程稳定的关键UA伪装、请求节奏和异常判断并发做好了链接提取也准了遇到真实网站还会有一堆七七八八的干扰。这一节把我踩过的坑集中整理出来。5.1 User-Agent和Referer绕不开的“身份伪装”很多服务器会检查请求头里的User-AgentUA如果识别出是Python requests的默认UA或者浏览器特征太明显缺失就可能拒绝服务。最稳妥的做法是准备一个UA池每次请求随机挑一个常见的浏览器UAimport random USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36, Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0, Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.3 Mobile/15E148 Safari/604.1, ] def random_headers(): return { User-Agent: random.choice(USER_AGENTS), Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,en-US;q0.5,en;q0.3, Connection: keep-alive, }还有一类链接必须带Referer头尤其是图片和附件有些站防盗链就是看Referer。处理方法是提取链接的时候把来源页URL记录下来下载时给download_file传一个referer_url参数加到请求头里。能解决一大批“下载403”问题。5.2 请求失败不能只看HTTP状态码我最初判断下载成功就是r.status_code 200后来发现并不够。有几种情况HTTP 200但文件是坏的服务器返回了一个自定义的HTML页面里面有“文件不存在”的字样状态码也是200。一台CDN节点故障返回的JSON错误信息同样挂200。下载的文件内容和Content-Length声明不一致。所以我这边最终的成功判定是**“状态码200 Content-Length校验如果有 文件大小0”**三合一。如果还嫌不够可以在下载完以后用file命令Linux下或者Python的magic库检查文件头是不是匹配预期的PDF/JPEG/ZIP魔数。这个做法在下载大量文件时非常管用能自动筛掉那些“下载了但打不开”的残废文件。5.3 重试策略不是越激进越好重试是我的老朋友了刚开始写爬虫的时候遇到失败就立刻重试而且无限重试。结果就是对端服务器已经过载了我还在疯狂打它最后干脆把我IP封了。现在的重试策略改成了这样超时类异常Timeout第一次重试等2秒第二次等4秒第三次等8秒给服务器喘息时间。HTTP 5xx类错误同样是递增等待最多重试2次。HTTP 403/404等直接放弃403说明是权限问题重试只会加重封禁风险404说明链接就是错的重试也白搭。所以我在download_file里保留了retries参数但会结合状态码决定还要不要继续尝试。403直接返回失败不浪费时间。这个方法我觉得是批量下载里最重要的“自我保护”机制。5.4 善用robots.txt和下载礼仪很多新手不管三七二十一上来就爬。我个人觉得至少做到这几条剩下的随项目而定先请求一下目标网站的robots.txt看哪些路径不允许爬把不允许的排除掉。下载频率尽量模拟人的行为不要并发十几二十个疯狂连打同一台服务器。这既是保护对方也是保护自己的IP。只下载公开页面里能直接访问的资源。需要登录才能下载的内容原则上不考虑自动破解。下载工具做出来的文件如果还要重新分发要留意版权问题。工具本身解决的是自动化效率不是替代人工审核。这里给两个实际能用的合规检查点一是robots.txt解析Python里可以用urllib.robotparser自带的模块二是下载前判断链接域名是否在目标站域名白名单内跨域下载默认关掉防止无意中下载了第三方资源。6. 全流程串联从“能跑的脚本”到“好用的工具”前面的模块拆完了最后把它们组装成一个完整的命令行工具。命令行方式的好处是不用改代码换一组URL、改个保存路径直接敲命令就行非常适合交给非技术同事使用。6.1 完整的入口脚本# main.py import argparse import json from pathlib import Path from link_extractor import extract_links, filter_downloadable from downloader import DownloadManager, batch_download def load_urls_from_file(path): with open(path, r, encodingutf-8) as f: return [line.strip() for line in f if line.strip()] def main(): parser argparse.ArgumentParser(descriptionPython爬虫自动下载网页链接工具) parser.add_argument(--input, -i, requiredTrue, help输入文件每行一个页面URL) parser.add_argument(--output, -o, default./downloads, help下载保存目录) parser.add_argument(--workers, -w, typeint, default8, help并发下载线程数) parser.add_argument(--extensions, default.pdf,.doc,.docx,.zip,.rar,.jpg,.jpeg,.png,.gif,.mp4, help要下载的文件后缀逗号分隔) parser.add_argument(--download-all, actionstore_true, help下载所有提取到的链接不按后缀过滤) args parser.parse_args() page_urls load_urls_from_file(args.input) exts {i.strip().lower() for i in args.extensions.split(,) if i.strip()} manager DownloadManager(history_filedownload_history.json) save_root Path(args.output) all_download_links [] # 1. 逐个页面提取链接 for page_url in page_urls: print(f解析页面: {page_url}) html fetch_page(page_url) links extract_links(html, page_url) if args.download_all: target_links links else: target_links filter_downloadable(links, exts) print(f 提取到 {len(links)} 个链接目标文件 {len(target_links)} 个) all_download_links.extend(target_links) # 2. 去重后下载 unique_links list(dict.fromkeys(all_download_links)) print(f共 {len(unique_links)} 个唯一文件需要下载) # 3. 按页面分组解目录保存此处按原URL分组目录名取页面标题或序号 for idx, page_url in enumerate(page_urls, 1): page_dir save_root / fpage_{idx:03d} page_dir.mkdir(parentsTrue, exist_okTrue) links_for_page [l for l in unique_links if l in all_download_links] # 简化实际按来源分组 batch_download(links_for_page, page_dir, max_workersargs.workers, managermanager) if __name__ __main__: main()这个脚本把“提取阶段”和“下载阶段”拆开了。实际使用中我建议分两步跑因为提取阶段和下载阶段的失败模式完全不同混在一起出问题时很难定位。6.2 运行示例在命令行里执行python main.py -i urls.txt -o ./downloads -w 10 -e .pdf,.docx,.zip,.jpg,.pngurls.txt里每行放一个页面URL。工具会先依次解析页面提取目标链接然后以10个线程的并发度下载到downloads/page_xxx/子目录。第一次跑完以后download_history.json里就存了所有成功下载的URL。下次再跑同样的输入文件已经下载过的会被自动跳过只处理新增文件。这个增量同步的体验跟网盘客户端非常像。6.3 日志和运行状态工具能维护的前提工具做出来不是跑一次就完了后续要能维护。我的习惯是加一个辅助日志函数把每次运行的摘要写进run.logimport logging logging.basicConfig( filenamerun.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s, datefmt%Y-%m-%d %H:%M:%S ) def log_result(url, status, msg): logging.info(f{status} | {url} | {msg})日志的价值在于程序跑完你不用盯着屏幕一个个数打开run.log就能看到哪些失败、失败原因是什么。批量下载的场景里10个链接可能成功9个失败的那1个可能是服务器临时抖了一下也可能是一直403。有了日志排查就变成了“打开文件、搜FAIL”两件事。6.4 调度和定时增量工具稳定了之后可以挂到定时任务里实现“全自动更新”。Windows下用“任务计划程序”Linux下用crontab每跑一次就会自动补下新发布文件。我自己的经验是这种定时任务别设太频繁一天一次足够了下载频率太高既没必要也容易给对方服务器添麻烦。还有个隐藏问题跑了一段时间以后download_history.json会越攒越大里面可能有几千个URL。这个文件本身也就几十KB完全不用清理。但如果你的链接数量级到了几十万可以考虑换成SQLite存储历史URL查询会更快。这个进阶操作本文不展开等真遇到性能瓶颈再优化也来得及。7. 从这套流程延伸出来的几个实际问题最后分享几个做这种工具时经常会遇到、但教程很少讲的延伸问题。7.1 网站改版后为什么又崩溃了我做的工具没法保证“一次跑通永远不坏”。最典型的是后台网站的页面结构变了导致我的XPath选择器提取不到链接工具就突然“失灵”。这种情况非常正常。我的应对是给提取逻辑加“主动体检”比如提取结果数量跟历史平均值差异超过3倍就告警说明页面结构可能变了需要人工确认。7.2 HTTP代理池要不要上并发拉大了之后很多人的反应是“要不要配一堆代理来防封”。我的经验是在你的IP没有被封之前不要主动上代理。优质代理资源本身就有稳定性和抓取成本问题引入代理反而让故障排查更难。真到了IP被封的那天优先做的是降低并发、调整请求间隔而不是立刻上代理。只有确实需要分布式采集大量数据时才考虑维护代理池。7.3 图片、PDF这些资源文件的验证下载图片和PDF这类二进制文件时我建议下完以后顺手做一个大小判断。如果一张JPG图片只有几百字节基本可以断定下错了。PDF文件有固定的文件头%PDF-下载以后可以读前几字节确认一下。相关代码很简单MAGIC_MAP { b%PDF: .pdf, bPK: .zip, b\xff\xd8\xff: .jpg, b\x89PNG: .png, } def check_file_signature(file_path): with open(file_path, rb) as f: head f.read(8) for magic, ext in MAGIC_MAP.items(): if head.startswith(magic): return True, ext return False, None这个检查跑起来很快但能把很多奇怪的问题拦在数据源头。特别是从网上下载的图片带个网页后缀名、或者内容实际是HTML的情况一查一个准。7.4 相对路径、绝对路径和CDN域名提取链接时还要注意有些网站的资源放在CDN域名下面跟页面主域名不一样。默认按同域名过滤会漏掉这些CDN上的文件。我的白名单逻辑是如果文件后缀命中了目标类型就允许下载不限制域名如果担心下载到无关第三方的资源就再加一层“域名白名单”配置把允许的域名列进去两条都满足才下载。写在最后我实际跑这种工具的一点体会这套工具我一直在用最深的感受是搞爬虫自动下载链接瓶颈从来不是“怎么写代码”而是“怎么让代码在真实网络环境里稳定跑完”。提取链接、下载文件都是几十行代码的事但把UA、超时、重试、去重、断点续传、日志、并发这些东西揉在一起才是一个可以交付出去让同事放心用的工具。如果你现在正准备上手做一个类似的Python爬虫自动下载链接工具我的建议顺序是先按第2节的代码把提取跑通再用第3节的基础下载函数下载20个文件然后逐步加入并发、历史记录、日志。一步到位写一个“全功能大而全”的脚本很容易因为一个小问题卡住整个开发进度。最后再分享一个我个人的小技巧先手工用requests或浏览器访问一个目标文件链接确认能直接下载再开始写爬虫。很多项目上来就写代码写完了发现那个站本身做了登录鉴权或者地域限制整个人工流程根本走不通代码再漂亮也白搭。先把链路亲手走一遍剩下的就是自动化而已。 Python爬虫自动下载网页链接：从提取到并发下载的工程实践
我手里正好有个给运营同事做的批量素材抓取小工具场景就是输入一批网页自动把页面里所有PDF、图片、压缩包下载到本地。做的时候发现网上聊“爬虫下载链接”的文章不少但大多是给个requests.get完事真正能把“自动下载网页链接”这件事做扎实的没几篇。这篇文章就把我做这个小工具的完整思路写下来。核心围绕三件事第一怎么从网页里精准提取出真正需要下载的链接第二怎么把批量下载做得又快又稳不因为几个坏链接就让整个任务挂掉第三怎么处理并发、反爬、断点续传这些工程化问题。不管你只是想抓个图集还是要定期同步某类资源文件这篇都能给你一条能直接落地的路线。1. 接到“自动下载网页链接”需求时先拆解再做很多新手拿到这个需求第一反应是“我直接requests然后正则匹配href就好了”。但实际接到任务尤其是给别人做工具的时候必须先搞清楚几个问题否则做出来就是个一次性脚本换个网站就废。1.1 这需求到底在说什么提取链接和下载文件是两码事“自动下载网页链接”这句话本身有歧义。它至少包含两种完全不同的场景场景A把网页里所有链接抓下来保存成清单。这种场景的核心是“提取”比如SEO分析、外链收集输出是一个文本文件或表格里面是URL列表。场景B把网页里指向具体文件的链接下载到本地。比如图片、PDF、ZIP、视频核心是“下载”要根据链接创建本地目录结构还要处理文件重名、下载失败重试。我的经验是直接把两种场景都做了做成一个可配置的工具。反正提取是下载的前置步骤代码不会多太多但适用范围会广很多。1.2 明确目标要下载哪些链接跳过哪些链接这是整件事最关键的一步也是最容易被忽略的。正则匹配a href...确实能把链接抓出来但抓出来的大部分都没用。我通常会先问自己几个问题目标网页是静态HTML还是需要渲染JS才能拿到完整内容这决定了用requests直接请求还是上Playwright。需要下载的文件类型有哪些PDF、图片、压缩包、还是附件里的docx是只下载当前页面里的直接链接还是需要递归爬取子页面里的文件链接下载的文件重名了怎么处理覆盖、重命名还是跳过需不需要记录下载历史下次增量同步这些问题想清楚了具体写代码的时候就会很顺。以我这次做的工具为例目标是爬取一组资讯页里的PDF附件和配图文件按原链接文件名保存到本地已存在的自动跳过。要求就是稳定跑几百个页面不能因为几个超时链接就中断。1.3 整体流程设计整个工具的运行链路是固定的后面所有代码都是围绕这条链路展开读取URL列表 → 请求页面HTML → 解析并提取所有候选链接 → 过滤出目标文件链接 → 下载文件 → 保存到本地目录这套链路看着简单但每一步都有不少隐藏问题。下面几节拆开逐个说。2. 链接提取为什么不用“万能”正则而用DOM解析提取链接的方案我见过有人直接用re.findall(rhref([^]*))一把梭。说实话这方法在简单页面上确实能用但踩过几次坑之后我就再也不敢用了。2.1 正则方案的三个致命问题第一HTML属性值不一定用双引号还可能是单引号甚至无引号正则写起来又长又容易漏。第二页面里有些href的值不是完整URL而是相对路径比如/files/report.pdf你直接拼接而不是解析BaseURI的话下载就是404。第三有些链接是javascript:void(0)、mailto:、#这种正则一股脑全抓出来后面还要再过滤几层。最恶心的一个案例是有个页面里的PDF链接是通过JS变量拼接出来的a hrefjavascript:download(report.pdf)这种情况纯正则根本提取不到真实地址。所以后来我的基本思路是能用DOM解析就用DOM解析只有在HTML结构极其规整、且确认没有JS生成内容的时候才考虑正则作为快速原型方案。2.2 用lxmlparsel提取链接代码简短且稳这里我推荐用parsel它是Scrapy底层用的解析库内部基于lxmlAPI非常简洁对新手也友好。下面这段就是我工具里实际用的提取函数import parsel def extract_links(html: str, base_url: str): 从HTML中提取所有a标签和img标签的链接 html: 页面源码 base_url: 用于拼接相对URL的基础地址一般是当前页面的URL 返回: 去重后的URL列表 sel parsel.Selector(texthtml) # 提取a标签href和img标签src raw_links sel.xpath(//a/href | //img/src).extract() absolute_links [] for raw in raw_links: if not raw or raw.strip() : continue # 跳过纯锚点、脚本调用、邮件协议 if raw.startswith(#) or raw.startswith(javascript:) or raw.startswith(mailto:): continue # 用urljoin把相对路径拼成绝对地址 absolute urljoin(base_url, raw.strip()) absolute_links.append(absolute) # 去重并保持顺序 seen set() unique_links [] for link in absolute_links: if link not in seen: seen.add(link) unique_links.append(link) return unique_links这里两个小要点urljoin不光能处理/files/a.pdf这种相对路径还能处理../files/a.pdf这种带父级跳转的比直接字符串拼接靠谱得多。只抓a和img标签通常够用如果网站有些文件是放在video标签的src或者source标签里可以按需加XPath表达式比如//video/src | //source/src。2.3 文件链接过滤只留我们真正要下载的东西链接提取出来之后真正需要下载的往往是特定后缀的地址。我的做法是维护一个后缀白名单命中才下载DOWNLOAD_EXTENSIONS { .pdf, .doc, .docx, .xls, .xlsx, .ppt, .pptx, .zip, .rar, .7z, .tar, .gz, .jpg, .jpeg, .png, .gif, .webp, .bmp, .mp4, .avi, .mkv, .mov, } def filter_downloadable(links, allowed_extsDOWNLOAD_EXTENSIONS): 从候选链接里过滤出可下载的文件链接同时去重 result [] seen set() for url in links: # 去掉URL的查询参数再取后缀 clean_url url.split(?)[0].split(#)[0] ext os.path.splitext(clean_url)[1].lower() if ext in allowed_exts and url not in seen: seen.add(url) result.append(url) return result注意这里有个容易被坑的点有些文件链接不是以真实后缀结尾的比如https://example.com/download?id123这种返回的其实是文件流。应对办法是先把链接下载下来或发HEAD请求看Content-Type或Content-Disposition里的文件名但这会增加很多请求开销。我一般是在工具里加一个开关默认走后缀白名单需要的时候手动打开“嗅探模式”用HEAD请求判断是否目标类型。2.4 动态渲染页面怎么处理如果目标页面是SPA单页应用比如Vue、React写的站直接requests拿到的HTML里根本没有完整链接因为数据是JS异步加载的。这种情况有两种解法简单场景直接看接口。打开浏览器F12Network里找XHR请求很可能直接请求一个JSON接口就能拿到带文件地址的数据比爬HTML还省事。复杂场景上无头浏览器。用Playwright或Selenium渲染完页面再提取链接。我这边的原则是能走接口就走接口实在不行才上无头浏览器。因为无头浏览器吃内存并发也不好控制维护成本明显更高。这篇不展开Playwright重点还在完整下载链路的稳定性上。3. 批量下载从这里开始才叫真正的工程化链接提取完下载这一步才是决定工具好不好用的分水岭。很多人写的下载代码是这样的import requests for url in link_list: r requests.get(url) with open(os.path.basename(url), wb) as f: f.write(r.content)这代码单跑一个链接没问题一旦放到真实场景就各种翻车文件名带非法字符导致保存失败、r.content一次性把大文件全读进内存、某次请求超时就导致整个程序崩溃、下载下来的文件打开发现是404页面……这些问题我在实际开发里全都踩过下面直接给一份能用于真实场景的下载代码。3.1 一份能直接用在大流量场景的稳健下载函数import os import time import requests from urllib.parse import unquote def safe_filename(url, defaultdownload.bin): 根据URL生成安全的本地文件名过滤掉windows非法字符 # 取URL最后一节并把URL编码还原为原始文件名 path url.split(?)[0] filename os.path.basename(path) filename unquote(filename) if not filename: filename default # 替换windows/linux下不允许出现在文件名里的字符 filename re.sub(r[\\/:*?|], _, filename) # 限制长度防止超出文件系统上限 if len(filename) 200: name, ext os.path.splitext(filename) filename name[:180] ext return filename def download_file(url, save_path, timeout30, retries3): 下载单个文件支持重试和流式写入 url: 文件地址 save_path: 本地保存完整路径 timeout: 单次请求超时时间 retries: 失败重试次数 返回: (是否成功, 保存路径或错误信息) for attempt in range(retries): try: with requests.get(url, streamTrue, timeouttimeout) as r: if r.status_code ! 200: print(f[{attempt 1}/{retries}] HTTP {r.status_code}: {url}) time.sleep(2 * (attempt 1)) continue # Header里的文件大小字节用于下载进度判断 total_size int(r.headers.get(Content-Length, 0)) downloaded 0 # 生成临时文件名避免下载到一半的残缺文件占用正式文件名 tmp_path save_path .part with open(tmp_path, wb) as f: for chunk in r.iter_content(chunk_size8192): if chunk: f.write(chunk) downloaded len(chunk) # 如果服务器给了Content-Length做下大小校验 if total_size 0 and downloaded ! total_size: raise IOError(f文件大小不匹配: 期望{total_size}, 实际{downloaded}) os.replace(tmp_path, save_path) return True, save_path except Exception as e: print(f[{attempt 1}/{retries}] 下载失败: {url}, 错误: {e}) if attempt retries - 1: time.sleep(2 * (attempt 1)) return False, str(e) if e in locals() else 未知错误几个设计点值得说明流式下载iter_content(chunk_size8192)逐块写入磁盘避免大文件撑爆内存。直接r.content会把整个文件读进内存下载一个2GB的压缩包的时候机器基本就卡死了。临时文件下载完再改名这是从生产环境学到的经验。如果下载到一半网络断了残留在磁盘上的xxx.pdf会让人误以为文件已下载成功。用.part后缀先写下载完再os.replace进程挂掉也不会有“假成功”的文件。Content-Length校验虽然不一定每次都有这个Header但有的话就做一次大小核对能拦截一大半“下载下来是HTML错误页”的情况。3.2 主循环带历史记录和断点续传的批量下载单文件下载没问题之后就要把整个链路串起来。这里我加了一个“已下载历史记录”机制作用有两个一是避免重复下载相同文件浪费流量二是万一程序中断了下次启动能直接跳过已经完成的文件效果上就是断点续传。import json import os from pathlib import Path class DownloadManager: def __init__(self, history_filedownload_history.json): self.history_file history_file self.history self._load_history() def _load_history(self): if os.path.exists(self.history_file): with open(self.history_file, r, encodingutf-8) as f: return set(json.load(f)) return set() def save_history(self): with open(self.history_file, w, encodingutf-8) as f: json.dump(list(self.history), f, ensure_asciiFalse) def run(self, urls, save_dir): save_dir Path(save_dir) save_dir.mkdir(parentsTrue, exist_okTrue) results [] for idx, url in enumerate(urls, 1): filename safe_filename(url) save_path save_dir / filename # 历史记录里有这条URL且文件还存在直接跳过 if url in self.history and save_path.exists(): print(f[{idx}/{len(urls)}] 跳过: {filename}) results.append((url, True, skipped)) continue ok, msg download_file(url, str(save_path)) if ok: self.history.add(url) results.append((url, True, msg)) else: results.append((url, False, msg)) # 每下载完一个就保存一次历史防止进程崩溃丢进度 if idx % 10 0 or not ok: self.save_history() self.save_history() return results这段代码解决的问题是下载任务的“可重启性”。我之前给一个同事做素材批次下载跑了半小时断网了没有断点续传的话要从头再来一遍有了历史记录之后就只看没下完的那几个省下的时间非常可观。3.3 目录结构怎么组织如果下载任务涉及多个来源页面我建议按“页面URL”维度来分目录。比如爬的是一个列表页、详情页混在一起的内容目录结构可以是这样downloads/ ├── page_001/ │ ├── report_2024.pdf │ └── cover.jpg └── page_002/ ├── attachment.zip └── image1.png实现上就是在run()里根据当前解析的是哪个URL把save_dir继续往下一层建子目录。目录层级多了之后有一个细节要注意Windows下目录名也不能带:\/?这些字符所以目录名也要用safe_filename清洗一遍。4. 并发下载串行太慢但并发设计也有讲究第一次用串行方式跑一百个文件如果每个文件3秒差不多五分钟。看起来还行但文件数量到几百上千的时候串行就完全不可接受了。这时候就要上并发。4.1 三种常见并发方案怎么选Python里做下载并发常用三种方案我对它们的评价如下方案实现方式适用场景上手难度多线程concurrent.futures.ThreadPoolExecutorIO密集型任务简单直接推荐新手低协程asyncioaiohttp并发量极大、追求极致吞吐中高多进程multiprocessing.PoolCPU密集型比如同时做压缩/解析下载场景不推荐中下载文件主要是IO瓶颈不是CPU瓶颈所以线程池就够了。协程的上限确实高但调试和排错难度也高非必要不上。多进程在下载场景里没什么优势进程切换开销大还占内存唯一好处是能绕过GIL但下载本来就不怎么吃CPUGIL不是瓶颈。4.2 用ThreadPoolExecutor实现并发下载改造起来非常简单核心逻辑不变只是把主循环的任务丢给线程池from concurrent.futures import ThreadPoolExecutor, as_completed def batch_download(urls, save_dir, max_workers8, managerNone): 并发下载任务 urls: 文件URL列表 save_dir: 保存目录 max_workers: 并发线程数 manager: DownloadManager实例用于记录历史 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_url { executor.submit(download_file, url, str(Path(save_dir) / safe_filename(url)), 30, 3): url for url in urls } for future in as_completed(future_to_url): url future_to_url[future] try: ok, msg future.result() results.append((url, ok, msg)) print(f{OK if ok else FAIL} {url} - {msg}) except Exception as e: results.append((url, False, str(e))) print(f异常 {url}: {e}) return results创建线程池的时候max_workers不建议拍脑袋。目标网站的服务器能力、当前网络带宽、你不想被对方封IP这三个因素共同决定合理值。我的经验是8个线程起步最多不要超过16个。线程太多短时间内请求过于密集很容易触发目标站的反爬策略。曾经有一次我把max_workers调成50跑一个小网站的图集跑了不到两分钟IP就被临时限制了返回的全是403。后来老老实实降回10个线程把请求间隔稍微调大反而更稳定。4.3 并发下载的请求间距控制多线程并发会让请求间隔不可控所以最好自己做一个简单的“信号量限速”。用threading.Semaphore控制同时进行的下载数只是控制并发如果想控制“每秒最多发N个请求”可以维护一个带时间戳的计数器。这里给一个轻量方案import threading import time class RateLimiter: 简单的每秒请求数限制器 def __init__(self, max_per_second5): self.interval 1.0 / max_per_second self.lock threading.Lock() self.next_time time.time() def wait(self): with self.lock: now time.time() wait_time self.next_time - now if wait_time 0: time.sleep(wait_time) # 下一次请求最早允许的时间 self.next_time max(now, self.next_time) self.interval在download_file里请求前调用rate_limiter.wait()就能把带宽占用控制得比较均匀。单看一个请求只多等了几十毫秒但对上百个请求来说这种平滑限速能让目标服务器更“舒服”不容易触发风控。5. 让下载流程稳定的关键UA伪装、请求节奏和异常判断并发做好了链接提取也准了遇到真实网站还会有一堆七七八八的干扰。这一节把我踩过的坑集中整理出来。5.1 User-Agent和Referer绕不开的“身份伪装”很多服务器会检查请求头里的User-AgentUA如果识别出是Python requests的默认UA或者浏览器特征太明显缺失就可能拒绝服务。最稳妥的做法是准备一个UA池每次请求随机挑一个常见的浏览器UAimport random USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36, Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0, Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.3 Mobile/15E148 Safari/604.1, ] def random_headers(): return { User-Agent: random.choice(USER_AGENTS), Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,en-US;q0.5,en;q0.3, Connection: keep-alive, }还有一类链接必须带Referer头尤其是图片和附件有些站防盗链就是看Referer。处理方法是提取链接的时候把来源页URL记录下来下载时给download_file传一个referer_url参数加到请求头里。能解决一大批“下载403”问题。5.2 请求失败不能只看HTTP状态码我最初判断下载成功就是r.status_code 200后来发现并不够。有几种情况HTTP 200但文件是坏的服务器返回了一个自定义的HTML页面里面有“文件不存在”的字样状态码也是200。一台CDN节点故障返回的JSON错误信息同样挂200。下载的文件内容和Content-Length声明不一致。所以我这边最终的成功判定是**“状态码200 Content-Length校验如果有 文件大小0”**三合一。如果还嫌不够可以在下载完以后用file命令Linux下或者Python的magic库检查文件头是不是匹配预期的PDF/JPEG/ZIP魔数。这个做法在下载大量文件时非常管用能自动筛掉那些“下载了但打不开”的残废文件。5.3 重试策略不是越激进越好重试是我的老朋友了刚开始写爬虫的时候遇到失败就立刻重试而且无限重试。结果就是对端服务器已经过载了我还在疯狂打它最后干脆把我IP封了。现在的重试策略改成了这样超时类异常Timeout第一次重试等2秒第二次等4秒第三次等8秒给服务器喘息时间。HTTP 5xx类错误同样是递增等待最多重试2次。HTTP 403/404等直接放弃403说明是权限问题重试只会加重封禁风险404说明链接就是错的重试也白搭。所以我在download_file里保留了retries参数但会结合状态码决定还要不要继续尝试。403直接返回失败不浪费时间。这个方法我觉得是批量下载里最重要的“自我保护”机制。5.4 善用robots.txt和下载礼仪很多新手不管三七二十一上来就爬。我个人觉得至少做到这几条剩下的随项目而定先请求一下目标网站的robots.txt看哪些路径不允许爬把不允许的排除掉。下载频率尽量模拟人的行为不要并发十几二十个疯狂连打同一台服务器。这既是保护对方也是保护自己的IP。只下载公开页面里能直接访问的资源。需要登录才能下载的内容原则上不考虑自动破解。下载工具做出来的文件如果还要重新分发要留意版权问题。工具本身解决的是自动化效率不是替代人工审核。这里给两个实际能用的合规检查点一是robots.txt解析Python里可以用urllib.robotparser自带的模块二是下载前判断链接域名是否在目标站域名白名单内跨域下载默认关掉防止无意中下载了第三方资源。6. 全流程串联从“能跑的脚本”到“好用的工具”前面的模块拆完了最后把它们组装成一个完整的命令行工具。命令行方式的好处是不用改代码换一组URL、改个保存路径直接敲命令就行非常适合交给非技术同事使用。6.1 完整的入口脚本# main.py import argparse import json from pathlib import Path from link_extractor import extract_links, filter_downloadable from downloader import DownloadManager, batch_download def load_urls_from_file(path): with open(path, r, encodingutf-8) as f: return [line.strip() for line in f if line.strip()] def main(): parser argparse.ArgumentParser(descriptionPython爬虫自动下载网页链接工具) parser.add_argument(--input, -i, requiredTrue, help输入文件每行一个页面URL) parser.add_argument(--output, -o, default./downloads, help下载保存目录) parser.add_argument(--workers, -w, typeint, default8, help并发下载线程数) parser.add_argument(--extensions, default.pdf,.doc,.docx,.zip,.rar,.jpg,.jpeg,.png,.gif,.mp4, help要下载的文件后缀逗号分隔) parser.add_argument(--download-all, actionstore_true, help下载所有提取到的链接不按后缀过滤) args parser.parse_args() page_urls load_urls_from_file(args.input) exts {i.strip().lower() for i in args.extensions.split(,) if i.strip()} manager DownloadManager(history_filedownload_history.json) save_root Path(args.output) all_download_links [] # 1. 逐个页面提取链接 for page_url in page_urls: print(f解析页面: {page_url}) html fetch_page(page_url) links extract_links(html, page_url) if args.download_all: target_links links else: target_links filter_downloadable(links, exts) print(f 提取到 {len(links)} 个链接目标文件 {len(target_links)} 个) all_download_links.extend(target_links) # 2. 去重后下载 unique_links list(dict.fromkeys(all_download_links)) print(f共 {len(unique_links)} 个唯一文件需要下载) # 3. 按页面分组解目录保存此处按原URL分组目录名取页面标题或序号 for idx, page_url in enumerate(page_urls, 1): page_dir save_root / fpage_{idx:03d} page_dir.mkdir(parentsTrue, exist_okTrue) links_for_page [l for l in unique_links if l in all_download_links] # 简化实际按来源分组 batch_download(links_for_page, page_dir, max_workersargs.workers, managermanager) if __name__ __main__: main()这个脚本把“提取阶段”和“下载阶段”拆开了。实际使用中我建议分两步跑因为提取阶段和下载阶段的失败模式完全不同混在一起出问题时很难定位。6.2 运行示例在命令行里执行python main.py -i urls.txt -o ./downloads -w 10 -e .pdf,.docx,.zip,.jpg,.pngurls.txt里每行放一个页面URL。工具会先依次解析页面提取目标链接然后以10个线程的并发度下载到downloads/page_xxx/子目录。第一次跑完以后download_history.json里就存了所有成功下载的URL。下次再跑同样的输入文件已经下载过的会被自动跳过只处理新增文件。这个增量同步的体验跟网盘客户端非常像。6.3 日志和运行状态工具能维护的前提工具做出来不是跑一次就完了后续要能维护。我的习惯是加一个辅助日志函数把每次运行的摘要写进run.logimport logging logging.basicConfig( filenamerun.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s, datefmt%Y-%m-%d %H:%M:%S ) def log_result(url, status, msg): logging.info(f{status} | {url} | {msg})日志的价值在于程序跑完你不用盯着屏幕一个个数打开run.log就能看到哪些失败、失败原因是什么。批量下载的场景里10个链接可能成功9个失败的那1个可能是服务器临时抖了一下也可能是一直403。有了日志排查就变成了“打开文件、搜FAIL”两件事。6.4 调度和定时增量工具稳定了之后可以挂到定时任务里实现“全自动更新”。Windows下用“任务计划程序”Linux下用crontab每跑一次就会自动补下新发布文件。我自己的经验是这种定时任务别设太频繁一天一次足够了下载频率太高既没必要也容易给对方服务器添麻烦。还有个隐藏问题跑了一段时间以后download_history.json会越攒越大里面可能有几千个URL。这个文件本身也就几十KB完全不用清理。但如果你的链接数量级到了几十万可以考虑换成SQLite存储历史URL查询会更快。这个进阶操作本文不展开等真遇到性能瓶颈再优化也来得及。7. 从这套流程延伸出来的几个实际问题最后分享几个做这种工具时经常会遇到、但教程很少讲的延伸问题。7.1 网站改版后为什么又崩溃了我做的工具没法保证“一次跑通永远不坏”。最典型的是后台网站的页面结构变了导致我的XPath选择器提取不到链接工具就突然“失灵”。这种情况非常正常。我的应对是给提取逻辑加“主动体检”比如提取结果数量跟历史平均值差异超过3倍就告警说明页面结构可能变了需要人工确认。7.2 HTTP代理池要不要上并发拉大了之后很多人的反应是“要不要配一堆代理来防封”。我的经验是在你的IP没有被封之前不要主动上代理。优质代理资源本身就有稳定性和抓取成本问题引入代理反而让故障排查更难。真到了IP被封的那天优先做的是降低并发、调整请求间隔而不是立刻上代理。只有确实需要分布式采集大量数据时才考虑维护代理池。7.3 图片、PDF这些资源文件的验证下载图片和PDF这类二进制文件时我建议下完以后顺手做一个大小判断。如果一张JPG图片只有几百字节基本可以断定下错了。PDF文件有固定的文件头%PDF-下载以后可以读前几字节确认一下。相关代码很简单MAGIC_MAP { b%PDF: .pdf, bPK: .zip, b\xff\xd8\xff: .jpg, b\x89PNG: .png, } def check_file_signature(file_path): with open(file_path, rb) as f: head f.read(8) for magic, ext in MAGIC_MAP.items(): if head.startswith(magic): return True, ext return False, None这个检查跑起来很快但能把很多奇怪的问题拦在数据源头。特别是从网上下载的图片带个网页后缀名、或者内容实际是HTML的情况一查一个准。7.4 相对路径、绝对路径和CDN域名提取链接时还要注意有些网站的资源放在CDN域名下面跟页面主域名不一样。默认按同域名过滤会漏掉这些CDN上的文件。我的白名单逻辑是如果文件后缀命中了目标类型就允许下载不限制域名如果担心下载到无关第三方的资源就再加一层“域名白名单”配置把允许的域名列进去两条都满足才下载。写在最后我实际跑这种工具的一点体会这套工具我一直在用最深的感受是搞爬虫自动下载链接瓶颈从来不是“怎么写代码”而是“怎么让代码在真实网络环境里稳定跑完”。提取链接、下载文件都是几十行代码的事但把UA、超时、重试、去重、断点续传、日志、并发这些东西揉在一起才是一个可以交付出去让同事放心用的工具。如果你现在正准备上手做一个类似的Python爬虫自动下载链接工具我的建议顺序是先按第2节的代码把提取跑通再用第3节的基础下载函数下载20个文件然后逐步加入并发、历史记录、日志。一步到位写一个“全功能大而全”的脚本很容易因为一个小问题卡住整个开发进度。最后再分享一个我个人的小技巧先手工用requests或浏览器访问一个目标文件链接确认能直接下载再开始写爬虫。很多项目上来就写代码写完了发现那个站本身做了登录鉴权或者地域限制整个人工流程根本走不通代码再漂亮也白搭。先把链路亲手走一遍剩下的就是自动化而已。