做Python网络爬虫这么久，我一直觉得“表情包批量获取”是一个被低估的练手项目。它表面上只是把一堆图片下载到本地，实际却能把网络爬虫里最磨人的几个环节全串起来——动态渲染页面怎么拿到真实地址、异步并发怎么控制节奏、下载失败怎么自动重试、几千张图片怎么不重不漏地保存好。写这篇文章的起因是，前两周有人问我：很多表情包站点是JS动态加载的，直接用requests打开只能看到一堆空壳，要怎么抓？我当时就手写了一个基于异步技术加动态渲染的脚本，从拿到页面到把全套表情包下到硬盘，整个过程比想象中顺畅。今天就把这个项目的设计思路、核心代码和踩坑记录整理出来，适合有一定Python基础、想系统练一练异步爬虫和动态渲染的朋友参考。
1. 项目整体设计与思路拆解
1.1 这个项目到底在解决什么问题
先说你遇到的典型场景。你打开一个表情包聚合站，页面滚动几次，图片不断加载，看起来很方便。可真要批量保存的时候，麻烦就来了：手动右键另存为，一张张点，几十张还能忍，几百上千张直接让人崩溃。更麻烦的是这类站点基本都做了懒加载，图片不是一个静态HTML里全写死的，而是页面滚动到某个位置才触发请求，把真实图片地址动态塞进DOM。你用最基础的 requests + BeautifulSoup 去抓，拿到的是没有表情包的“空壳页面”，根本提取不到完整图片链接。
所以这个项目要解决的核心问题有三个：第一，让Python脚本像浏览器一样执行JavaScript，拿到动态渲染之后的完整页面数据；第二，把图片下载过程改成异步并发，充分利用网络等待时间，而不是一张张排队下载；第三，做好去重、断点续传和失败重试，保证几千张图片下载下来不重、不漏、不损坏。
另外还有一层不太显眼但很重要的需求：防封控。大批量快速请求，不管是页面接口还是图片CDN，都有可能触发限流。异步技术不光是提速，更要配合信号量、延时、重试策略，让整个抓取过程既快又稳。
1.2 为什么选异步技术加动态渲染这条技术路线
如果你只是想抓一个纯静态页面，图片链接全在HTML源代码里，那确实没必要上动态渲染。但实际表情包站点几乎都逃不开两种情况：一种是图片列表由接口异步返回，需要分析XHR请求再去拼参数；另一种是接口加密严重，直接抓接口很费劲，不如让浏览器自己跑完JS，再读取渲染后的DOM。第二种情况下，轻量级的requests方案基本要废弃，所以这里选择了Playwright做动态渲染。
异步技术则是因为下载场景天然适合协程。图片下载是典型的I/O密集型操作，网络等待时间占大头，CPU基本闲着。传统多线程能解决问题，但线程切换有开销，线程数量一多，内存和上下文切换的压力就上来了。多进程更重，适合计算密集型任务，用来下载图片有点杀鸡用牛刀。Python协程则可以在一个线程内用事件循环管理成千上万个连接，遇到网络I/O就挂起，数据到了再继续执行。配合Semaphore信号量控制并发数量，既快又可控。
你也可以把异步理解为“同时向服务器发起20个下载请求，但并不是开20个线程，而是让一个线程在这20个任务之间快速切换”。每个任务的网络等待时间被其他任务利用起来，整体吞吐量就有了质的提升。
1.3 技术栈选型与各组件职责
这个项目我用到的核心组件如下：
- Playwright：负责动态渲染。它的职责是打开浏览器、访问页面、等待JavaScript执行完、提取最终渲染后的图片节点。
- httpx：负责图片下载。它同时支持同步和异步API，异步下载时用AsyncClient，能复用连接池，性能表现比每次请求都新建连接好很多。
- BeautifulSoup4 + lxml：负责解析HTML。虽然Playwright本身可以用选择器提取元素，但配合BeautifulSoup做数据清洗和整理会更灵活。
- hashlib：生成文件名的哈希值，天然去重。
- asyncio：Python标准库里的异步框架，用来编排所有下载任务。
这里有个选型心得：曾有人问我为什么不用aiohttp。aiohttp也很成熟，但httpx的API风格更接近requests，上手成本低，而且原生支持HTTP/2和流式下载，在下载大图时能把数据一块块写入文件，不会一次性把所有二进制数据都塞进内存。对表情包这种动辄几MB的图片，流式下载能明显降低内存占用。
2. 核心细节解析与实操要点
2.1 动态页面与静态页面的核心区别
一开始接触爬虫的人经常混淆“页面源代码”和“浏览器里看到的页面”。静态页面的内容直接写在HTML文件里，你用requests拿到的响应和浏览器看到的差不多。动态页面的HTML只是一个空壳，真正的数据是页面加载后JavaScript发请求、改DOM得到的。
判断一个页面是不是动态渲染，有个很直接的方法：在浏览器里右键“查看网页源代码”，如果能看到表情包图片地址，基本就是静态的；如果源代码里只有一堆<script>标签和空容器，说明数据靠JS动态填充。更常见的还有懒加载，图片真正地址放在>import asyncio import hashlib from pathlib import Path from urllib.parse import urlparse import httpx SAVE_DIR = Path("expressions") SAVE_DIR.mkdir(exist_ok=True) SEM = asyncio.Semaphore(12) def safe_filename(url: str, index: int) -> str: ext = Path(urlparse(url).path).suffix.lower() if ext not in {".jpg", ".jpeg", ".png", ".gif", ".webp"}: ext = ".jpg" name_hash = hashlib.md5(url.encode("utf-8")).hexdigest()[:16] return f"{index:04d}_{name_hash}{ext}" async def download_one(client: httpx.AsyncClient, url: str, index: int): filename = safe_filename(url, index) filepath = SAVE_DIR / filename if filepath.exists() and filepath.stat().st_size > 0: print(f"skip exist: {filename}") return False async with SEM: for attempt in range(3): try: async with client.stream("GET", url) as resp: if resp.status_code != 200: raise RuntimeError(f"HTTP {resp.status_code}") temp_path = SAVE_DIR / f"{filename}.part" with open(temp_path, "wb") as f: async for chunk in resp.aiter_bytes(): f.write(chunk) temp_path.rename(filepath) return True except Exception as exc: print(f"[attempt {attempt + 1}] failed: {url}, reason: {exc}") if attempt < 2: await asyncio.sleep(2 ** attempt) return False
2.3 请求头、超时与重试机制为什么不能省
很多初学者下载图片时不带请求头，结果拿回来的是一张错误提示图或者403页面。原因是不少图床做了防盗链，会检查Referer是不是来自它自己的页面。所以下载表情包时，请求头至少要带上User-Agent和Referer，伪装成一个真实浏览器。
超时也必须设置。如果不设超时，某个连接卡死了，整个协程会一直挂在那里，事件循环里其他任务也会受影响。我一般用httpx.Timeout(30.0)，连接超时和读取超时都控制在合理范围。配合重试机制，单张图片最多尝试3次，每次失败后等待时间按指数退避：第一次等1秒，第二次等2秒。这样既能扛住偶发网络波动，又不会对服务器造成太大压力。
重试时要注意一个细节：下载到一半失败时，残留的.part临时文件要保留还是删除？我的习惯是不删除，因为下一次重试会用同一个filename.part覆盖写入。但如果失败次数用尽了，脚本结束之后最好清理掉所有.part文件，免得下次运行时误判。
2.4 文件命名、去重与目录管理
表情包文件名五花八门，有些是乱码，有些带中文，有些甚至没有扩展名。直接用原始文件名保存，很容易遇到文件系统不允许的字符，或者重名覆盖。所以这个项目里我用“序号 + URL哈希 + 扩展名”作为文件名。序号方便排序，哈希保证同一个URL不会重复下载，扩展名从URL路径里解析。如果URL里没有合理扩展名，就默认存成jpg。
去重方面，我做了两层判断。第一层在提取URL阶段，用集合去重，同一个URL只保留一次。第二层在下载阶段，检查目标文件是否已存在且大小不为0，如果存在就跳过。这样脚本中断了、图片下载失败了，重新跑一遍也不需要把已完成的图片再下一遍。
目录管理也值得注意。如果一次抓取几百页，所有图片堆在一个文件夹里，后续根本没法管理。我建议按照“站点名/分类/日期”这种层级建目录，或者在文件名前加上页码。比如每抓取一页，先创建page_01这样的子目录，再把本页的图片放进去。虽然多了一点代码量，但后期整理起来会轻松很多。
3. 实操过程与核心环节实现
3.1 环境准备与依赖安装
这是个手把手可以复现的项目，环境不用太复杂。首先确保电脑里已经装了Python 3.9以上版本，然后用虚拟环境隔离依赖。如果你用的是VSCode，建议先在VSCode里配置好Python解释器，再打开终端执行下面的命令。Windows环境注意PowerShell和CMD的激活命令略有不同。
python -m venv .venv # Windows .venv\Scripts\activate # Linux / macOS source .venv/bin/activate pip install playwright httpx beautifulsoup4 lxml playwright install chromiumplaywright install chromium这步一定要执行，它会下载浏览器内核。下载时间取决于网络情况，耐心等一会儿就好。装完之后可以跑一个极简测试：打开百度首页，打印标题，如果能看到标题说明Playwright环境没问题。
import asyncio from playwright.async_api import async_playwright async def test_playwright(): async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page() await page.goto("https://example.com") print(await page.title()) await browser.close() asyncio.run(test_playwright())3.2 用Playwright获取动态渲染后的页面数据
下面这段代码是抓取流程的第一阶段：遍历多个页面，等待图片节点出现，然后把图片地址收集到一个列表里。这里的example.com只是示意，实际使用时替换成你自己的目标站点。
async def fetch_image_urls(start_page: int, end_page: int) -> list[str]: image_urls = [] seen = set() async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page() for page_no in range(start_page, end_page + 1): target_url = f"https://example.com/expressions?page={page_no}" print(f"crawling page {page_no}") await page.goto(target_url, wait_until="domcontentloaded") await page.wait_for_selector("img.expression-image", timeout=15000) items = await page.query_selector_all("img.expression-image") for item in items: src = await item.get_attribute("data-src") or await item.get_attribute("src") if src and src.startswith("http") and src not in seen: seen.add(src) image_urls.append(src) print(f"page {page_no}: collected {len(items)} images, total {len(image_urls)}") await browser.close() return image_urls这里的wait_for_selector是重点。它告诉Playwright：轮询页面，直到能找到img.expression-image这个选择器对应的元素。选择器可以从浏览器的开发者工具里复制，也可以根据页面结构自己写。如果15秒内还没出现，会抛超时异常。超时后别急着加等待时间，先看看是不是选择器写错了。
有些图片是背景图，元素不是img而是div，地址写在style属性里。这时候选择器要改成div.expression-item，再用正则从style里把url(...)提取出来。处理方式略有不同，但整体思路一致：先确认DOM结构，再写提取逻辑。
3.3 解析表情包链接与信息清洗
我自己习惯把页面交给BeautifulSoup再做一轮清洗，因为有时候Playwright拿到的属性值里带有缩进、换行、引号等杂质。比如>from bs4 import BeautifulSoup def clean_urls(raw_urls: list[str]) -> list[str]: cleaned = [] for url in raw_urls: url = url.strip() if url.startswith("//"): url = "https:" + url if url.startswith("http://") or url.startswith("https://"): cleaned.append(url) return list(dict.fromkeys(cleaned))
这里也适合做内容筛选。如果你只想要gif动图，就用suffix in (".gif", ".webp")过滤一遍；如果只想要某一类表情包，可以在URL路径里匹配关键词。这一步早点做，能减少无效下载，节省带宽和时间。
3.4 异步批量下载图片的全过程
拿到URL列表后，异步下载的主流程特别简单：创建AsyncClient，把每个URL包装成协程，再用asyncio.gather一起执行。
async def main(): urls = await fetch_image_urls(1, 10) urls = clean_urls(urls) print(f"total urls to download: {len(urls)}") headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://example.com/", } async with httpx.AsyncClient( headers=headers, timeout=httpx.Timeout(30.0), follow_redirects=True, ) as client: tasks = [ download_one(client, url, index) for index, url in enumerate(urls) ] results = await asyncio.gather(*tasks, return_exceptions=True) downloaded = sum(1 for r in results if r is True) print(f"download finished: {downloaded} new files") if __name__ == "__main__": asyncio.run(main())几点说明。follow_redirects=True必须开，很多图床会把图片地址重定向到CDN，不开的话拿不到最终内容。gather的return_exceptions=True也很关键，它保证单个任务异常不会影响其他任务，否则一个烂链接就能让整个下载任务崩掉。下载逻辑里已经把异常在协程内部消化了，返回True或False，所以外层其实不会拿到太多异常。
关于事件循环的关闭，asyncio.run(main())会自动创建和关闭事件循环，不需要手动调用loop.close()。如果自己用asyncio.new_event_loop()，一定要在程序退出前关闭，否则容易看到Event loop is closed的警告。
3.5 断点续传与多轮脚本运行的实现
这个项目的核心要求是“大规模”，几百上千张图片很难保证一次跑完。断点续传是我认为这个项目里最有价值的细节，比单纯的下载速度更能体现工程的完整性。
实现方案其实很简单：下载前检查目标文件是否已存在并且大小大于0；下载中先写入.part临时文件，全部写完后重命名为正式文件名。这样即使中途断电、断网、脚本被手动终止，已下载完成的文件不会被破坏，没有完成的只是留下一个.part文件，重新运行脚本时会从零开始覆盖写入。
另一个实用做法是把失败的URL写到一个文本文件里，比如failed.txt。下载结束后读一遍这个文件，把失败项再跑一轮。对于动辄几千张图片的任务，这个机制能救命。我第一次跑这个脚本时没有加失败记录，结果最后发现40多张图片静默失败了，排查半天才定位到是Referer不对，只能重新跑一遍，浪费时间。
4. 常见问题与排查技巧实录
4.1 出现SSL证书校验错误怎么办
httpx对SSL证书校验比较严格，有些图床的证书链不完整，会抛出certificate verify failed。这时候很多人的第一反应是设置verify=False，确实能临时解决问题，但我不建议在正式脚本里这么做，因为它会让你所有请求都失去证书校验，存在中间人攻击风险。
更稳妥的办法是先用浏览器打开目标图片地址，看看是不是真的证书有问题。有些时候是系统根证书过期，升级Python或者更新系统证书就能解决。如果只是个别图床的问题，可以单独给这个域名配置一个SSLContext，而不是全局关闭校验。
4.2 图片下载到一半就结束，文件损坏
出现过好几次这样的情况：HTTP状态码是200，文件也下载了，但打开图片提示文件损坏。原因通常是重定向或压缩问题。CDN返回的内容不一定是图片本身，可能是错误页、压缩包或者一个302跳转。所以下载时一定要检查响应的Content-Type，应该是image/jpeg、image/png这类类型，如果返回text/html，说明拿到的根本不是图片。
另一个原因是磁盘空间不足。当并发数量调得很大、图片又比较大时，临时文件会占用大量磁盘空间。我的建议是下载完成后先检查文件大小，如果小于几KB，多半就是错误响应，直接删掉并记录到失败列表。
if filepath.stat().st_size < 1024: filepath.unlink() print(f"unlink too small file: {filename}")4.3 动态内容一直加载不出来
如果wait_for_selector超时，先不要盲目把超时时间从15秒改成60秒，那样只会让脚本卡得更久。建议进入调试模式，把headless=False打开，让浏览器窗口显示出来，亲自观察页面加载过程。你可能会看到两种情况：页面里有个验证码弹窗，需要人工处理；或者图片懒加载机制是“滚动到可视区域才触发”，从来没触发过滚动事件。
对于懒加载页面，可以在Playwright里执行滚动操作，模拟用户往下翻页。简单的做法是用page.mouse.wheel(0, 3000)滚几次，或者执行JS把所有图片的src从>async def download_one(client, url, index): async with SEM: await asyncio.sleep(random.uniform(0.1, 0.6)) # 原有下载逻辑
随机延时的作用是让请求在时间轴上分布得更自然。另外如果被封的对象是整个IP，断网重拨或者等待一段时间恢复是常见办法。这个项目本身以学习和练手为主，不建议也不讨论任何绕过反爬的手段，合规和克制才是长期能稳定运行的前提。
4.5 常见问题速查表
| 症状 | 可能原因 | 解决办法 |
|---|---|---|
| 页面拿不到图片元素 | 选择器写错或懒加载未触发 | 调试模式查看DOM，模拟滚动 |
| 下载返回403 | 缺少Referer或User-Agent | 带上完整浏览器请求头 |
| 图片文件无法打开 | 响应不是图片内容 | 检查Content-Type，设置重试 |
| 连接卡住不结束 | 超时未设置 | 设置httpx.Timeout并重试 |
| 磁盘被占满 | 并发过高导致临时文件积压 | 降低并发，清理.part |
| 重复下载大量图片 | 没有做去重 | URL集合去重，文件hash命名 |
5. 避坑经验与扩展方向
5.1 我在实操中踩过的三个比较典型的坑
第一个坑是复用连接。最初我每下载一张图片就新建一个httpx.Client，下载到200多张时明显感觉速度下降，CPU占用也不正常。后来改成复用AsyncClient，整个脚本只用一次连接池，速度立刻提上来了。连接池能复用TCP连接，省掉了反复握手的时间，对大规模下载提升非常明显。
第二个坑是URL里的问号参数导致扩展名识别失败。有些图片地址类似https://cdn.example.com/photo?id=12345，我用urlparse(url).path取不到.jpg后缀，默认存成了.jpg，但实际内容是PNG或者WebP，结果图片能打开，只是扩展名不匹配。后来用系统库或者依赖响应头的Content-Type来推断扩展名，顺便把format转成统一格式，问题就解决了。
第三个坑是页面编码问题。BeautifulSoup在解析网页时，如果页面声明的是UTF-8但实际内容是GBK，中文标题会乱码。虽然表情包URL大多是英文数字，但万一以后扩展成抓取标题、分类信息，编码问题就绕不过去。稳妥的做法是在page.goto()之后用page.content()获取整个渲染后的HTML，再交给BeautifulSoup，让BeautifulSoup按照页面meta标签自动识别编码。
5.2 这个项目后续还能怎么扩展
表情包批量获取只是异步爬虫的一个切入场景，把这个项目吃透之后，稍加改动就能迁移到很多其他场景。比如把输入从固定URL改成关键词搜索，就能做成“关键词表情包下载器”；把解析规则抽象成配置文件，就能适配不同站点。下载下来的图片还能继续做数据清洗、分类打标，甚至拿去训练一个表情包分类模型，整个链路就更有意思了。
从工程化角度，可以把URL列表存进SQLite，下载状态用数据库字段标记；进度显示可以用tqdm库，跑起来更直观；多分类站点可以考虑用 playwright 的browser.new_context()开多个隔离上下文，互不干扰。异步架构里还能加入任务队列，比如用asyncio.Queue缓存待下载URL，生产者不断往队列里放，消费者按优先级下载。
另外一个扩展点是保存策略。现在图片直接落本地磁盘，图片量大了以后可以改造成：先下载到临时目录，校验完毕后移动到正式目录，定期做增量备份。对于动辄几个GB的表情包库，这个习惯能避免很多重复劳动。
最后分享一个小技巧：写这种批量下载脚本时，我在每个环节都会打印进度日志，并且把日志同步写入一个download.log文件。这样脚本跑到一半出了异常，不用一直盯着终端，回头翻日志就能定位是哪一张图片、哪个请求失败。异步程序的错误不像同步代码那么直观，日志越详细，排查越快。
我个人的体会是，爬虫项目的核心从来不是“能不能爬到”，而是“能不能稳定、高效、可维护地爬到”。异步技术负责高效，动态渲染负责解决稳定性里的页面加载问题，而真正让脚本长时间跑不翻车的，是那些看起来不起眼的信号量、重试、临时文件处理和日志。希望这份记录能帮你少走点弯路，也欢迎你在这个思路上改造出更适合自己使用习惯的版本。
