一、写在前面:为什么选择花瓣网?
花瓣网(Huaban.com)是国内极具影响力的创意图片收藏与分享平台,每天都有数百万设计师、插画师、创意工作者在这里采集灵感。一个画板(Board)通常包含数十到数千张高质量图片,如果能批量爬取这些图片用于学术研究、个人学习或设计素材整理,将极大提升效率。
然而,随着反爬技术的不断演进,现代网站已不再像几年前那样“裸奔”。花瓣网采用了接口动态加载、参数加密、请求头校验、频率限制等多重防护。本文将从零开始,手把手带你构建一个企业级的爬虫程序,全面覆盖:
-
异步加载分析与XHR抓包
-
动态滚动模拟与延时控制
-
请求签名参数破解思路(合法声明)
-
请求头伪装与IP轮换策略
-
异常捕获与断点续爬
-
图片高效下载与文件命名规范
全文代码均基于 Python 3.10+,使用 requests、selenium(可选)与 httpx 等多方案,并给出适合生产环境的最终版代码。
二、技术分析:花瓣网的防爬机制解剖
2.1 数据加载方式:无限滚动 + Ajax
当你在浏览器中打开一个画板页面(例如 https://huaban.com/boards/xxx),滚动到底部时,新图片会自动出现。这显然是Ajax 异步加载,而非服务端渲染(SSR)。这意味着我们不能直接用 BeautifulSoup 解析初始HTML,需要直接请求后台接口。
打开浏览器开发者工具(F12)→ 网络(Network)→ XHR/Fetch 筛选,滚动页面,你会看到类似这样的请求:
text
GET https://huaban.com/boards/48902858/?max=720694006&limit=20&wfl=1
参数说明:
-
max:上一次返回的最后一张图片的ID(游标),用于分页。
-
limit:每页加载数量,通常20-40。
-
wfl:固定值1,疑似风控标识。
返回的数据是JSON格式,其中 board.pins 数组包含了当前页的图片信息,每个pin包含图片的 key、url 等。图片真实地址形如:
text
2026-06-1104kzmlc2mbt.jpg
2.2 反爬手段分层
| 初级 | User-Agent 校验 | 伪造常见浏览器UA |
| 中级 | Referer & Origin 校验 | 正确设置请求头 |
| 高级 | 请求频率限制(IP限流) | 添加延时、代理IP池 |
| 进阶 | 签名参数(_bsx, _bex等) | 逆向JS(仅供学习)或使用Selenium |
| 高级 | 浏览器指纹(TLS指纹) | 使用 curl_cffi 或 pyhttpx |
重要说明:本文提供的代码严格遵守法律法规,仅用于个人学习与研究。爬取时请控制频率,勿对目标服务器造成压力,并尊重版权,不得商用。
三、方案选型:直接API请求 vs Selenium
3.1 直接API请求(推荐)
优点:速度快、资源占用低、易于大规模爬取。 缺点:需要解析签名算法(花瓣网近年加入了 _bsx、_bex 等动态参数,且会变种)。
实际上,通过抓包发现,目前花瓣网的公开API接口(无登录状态)并不强制校验复杂的签名,只校验 User-Agent 和 Referer。但为了防止随时升级,代码中我们预留了签名处理钩子。
3.2 Selenium自动化(备用方案)
通过模拟浏览器真实操作,完美绕过前端JS加密,但速度较慢,内存消耗大。适合对稳定性要求极高、且不追求速度的场景。
3.3 本教程采用:混合模式
主方案为 requests + 参数伪造 + 重试机制;当遇到封禁或验证码时,可降级至 selenium 或更换代理。
四、环境准备与依赖安装
4.1 创建虚拟环境(强烈建议)
bash
python -m venv huaban_env
source huaban_env/bin/activate # Linux/Mac
# 或 .\\huaban_env\\Scripts\\activate (Windows)
4.2 安装核心依赖
bash
pip install requests beautifulsoup4 lxml retry fake-useragent curl_cffi
pip install selenium webdriver-manager # 备用方案
pip install tqdm loguru # 进度条和日志
所有依赖版本建议锁定:
text
requests==2.31.0
fake-useragent==1.4.0
curl_cffi==0.5.10
selenium==4.15.0
tqdm==4.66.1
loguru==0.7.2
五、手把手实现:从抓包到代码落地
5.1 获取画板ID
任意画板URL示例:https://huaban.com/boards/48902858/ 其中 48902858 就是 board_id。也可以从分享链接中提取。
5.2 构造首次请求
我们需要模拟浏览器发送请求头,并带上必要参数。
python
import requests
from fake_useragent import UserAgent
def get_headers():
ua = UserAgent()
return {
'User-Agent': ua.random,
'Referer': 'https://huaban.com/',
'Origin': 'https://huaban.com',
'Accept': 'application/json, text/plain, */*',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Connection': 'keep-alive',
}
5.3 解析API返回结构
请求接口:
python
url = f"https://huaban.com/boards/{board_id}/?max={max_param}&limit={limit}&wfl=1"
返回JSON关键结构:
json
{
"board": { … },
"pins": [
{
"pin_id": 12345678,
"file": {
"key": "xxx.jpg",
"type": "image/jpeg"
},
"user": { … }
}
]
}
图片真实地址拼接:
text
https://img.hb.aicdn.top/ + file.key
注意:部分图片可能使用 https://hbimg.huabanimg.com/ 域名,两种均可。
5.4 模拟滚动分页
核心逻辑:维护一个 max 游标,每次请求后从返回的 pins 中取最后一个 pin_id 作为下一次的 max,直到返回的 pins 数量小于 limit。
python
def fetch_board_pins(board_id, limit=20, max_pins=None):
pins = []
current_max = max_pins if max_pins else None
while True:
params = {}
if current_max:
params['max'] = current_max
params['limit'] = limit
params['wfl'] = 1
resp = requests.get(
f'https://huaban.com/boards/{board_id}/',
headers=get_headers(),
params=params,
timeout=10
)
if resp.status_code != 200:
print(f"请求失败,状态码:{resp.status_code}")
break
data = resp.json()
new_pins = data.get('pins', [])
if not new_pins:
break
pins.extend(new_pins)
# 更新游标
current_max = new_pins[-1]['pin_id']
if len(new_pins) < limit:
break
# 控制请求频率
time.sleep(1)
return pins
5.5 图片下载与存储
我们需要一个健壮的下载函数,支持重试、校验文件完整性、避免重复下载。
python
import os
import hashlib
import time
from pathlib import Path
def download_image(img_url, save_path, max_retries=3):
"""下载单张图片,支持重试"""
for attempt in range(max_retries):
try:
headers = get_headers()
# 图片下载通常需要referer为huaban.com
headers['Referer'] = 'https://huaban.com/'
resp = requests.get(img_url, headers=headers, timeout=15, stream=True)
if resp.status_code == 200:
content_type = resp.headers.get('content-type', '')
if 'image' not in content_type:
# 可能是反爬返回的html
raise Exception("非图片内容")
with open(save_path, 'wb') as f:
for chunk in resp.iter_content(chunk_size=8192):
f.write(chunk)
# 校验文件大小不为0
if os.path.getsize(save_path) > 0:
return True
else:
print(f"下载失败,状态码:{resp.status_code}")
except Exception as e:
print(f"下载异常(尝试 {attempt+1}/{max_retries}): {e}")
time.sleep(2)
return False
5.6 文件名生成策略
避免文件名冲突:使用 pin_id + 文件哈希 或直接 pin_id.jpg。推荐前者以保证唯一性。
python
def get_unique_filename(pin_id, img_key):
ext = img_key.split('.')[-1].lower()
if ext not in ['jpg', 'jpeg', 'png', 'gif', 'webp']:
ext = 'jpg'
return f"{pin_id}.{ext}"
六、进阶防反爬策略(精华部分)
6.1 动态延时与随机休眠
避免请求过快引起IP临时封锁。我们采用指数退避 + 随机抖动:
python
import random
import time
def random_sleep(base=1, jitter=0.5):
time.sleep(base + random.uniform(0, jitter))
每请求一页,random_sleep(0.5, 0.3),每下载一张图片,随机休眠 0.2~0.8 秒。
6.2 使用代理IP池
当爬取大规模画板(超过1000张图片)时,建议购买代理服务或使用免费代理池(不稳定)。示例:
python
proxies = {
'http': 'http://127.0.0.1:10809',
'https': 'http://127.0.0.1:10809',
}
requests.get(url, proxies=proxies, …)
生产中可用 requests 配合 itertools.cycle 轮换代理。
6.3 解决TLS指纹封锁(最新技术)
许多网站已开始使用 JA3指纹 识别非浏览器请求。此时传统 requests 库会被识别。解决方案:
-
使用 curl_cffi 库模拟 Chrome/Edge 浏览器指纹
-
安装:pip install curl_cffi
-
示例:
python
from curl_cffi import requests as cffi_req
resp = cffi_req.get(url, impersonate="chrome110", headers=get_headers())
impersonate 参数可选 chrome110、edge101、safari15_5 等。这是目前最强的反反爬方案。
6.4 会话保持与Cookie
花瓣网不需要登录也可访问公开画板,但带上会话可以降低风控风险。
python
session = requests.Session()
session.headers.update(get_headers())
# 首次访问首页获取cookie
session.get('https://huaban.com/', timeout=5)
# 后续使用session发起请求
七、完整代码实现(可直接运行)
以下代码整合了上述所有要点,采用 curl_cffi 增强稳定性,支持断点续爬,并输出精美日志。
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
花瓣网画板图片爬虫 – 完整版
支持:模拟滚动、动态延时、指纹伪装、断点续传
作者:技术博客实战
日期:2025-03
"""
import os
import json
import time
import random
from pathlib import Path
from typing import List, Dict, Optional
from urllib.parse import urljoin
from tqdm import tqdm
from loguru import logger
from curl_cffi import requests as cffi_req
from fake_useragent import UserAgent
# 配置日志
logger.add("huaban_crawler.log", rotation="10 MB", level="INFO")
class HuabanBoardCrawler:
"""花瓣网画板爬虫核心类"""
def __init__(self, board_id: str, save_dir: str = "./downloads",
max_images: Optional[int] = None, use_proxy: bool = False):
self.board_id = board_id
self.save_dir = Path(save_dir) / board_id
self.save_dir.mkdir(parents=True, exist_ok=True)
self.max_images = max_images
self.use_proxy = use_proxy
# 断点续爬记录文件
self.record_file = self.save_dir / "downloaded.txt"
self.downloaded_pins = self._load_downloaded_records()
# 初始化会话
self.session = self._create_session()
def _create_session(self):
"""创建带有指纹模拟的会话"""
session = cffi_req.Session()
# 模拟 Chrome 110 指纹
session.impersonate = "chrome110"
# 更新headers
ua = UserAgent()
session.headers.update({
'User-Agent': ua.random,
'Referer': 'https://huaban.com/',
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Cache-Control': 'no-cache',
})
# 先访问首页获取必要的cookie
try:
session.get('https://huaban.com/', timeout=5)
except Exception as e:
logger.warning(f"获取首页cookie失败: {e}")
return session
def _load_downloaded_records(self) -> set:
"""加载已下载的图片ID集合,用于断点续爬"""
if self.record_file.exists():
with open(self.record_file, 'r') as f:
return set(line.strip() for line in f)
return set()
def _save_downloaded_record(self, pin_id: str):
"""记录已下载的图片ID"""
with open(self.record_file, 'a') as f:
f.write(f"{pin_id}\\n")
self.downloaded_pins.add(pin_id)
def fetch_pins(self, limit: int = 20) -> List[Dict]:
"""获取画板下所有图片信息(分页滚动)"""
all_pins = []
current_max = None
page = 0
while True:
page += 1
params = {'limit': limit, 'wfl': 1}
if current_max:
params['max'] = current_max
url = f"https://huaban.com/boards/{self.board_id}/"
try:
logger.info(f"正在请求第{page}页,max={current_max}")
resp = self.session.get(url, params=params, timeout=10)
if resp.status_code != 200:
logger.error(f"请求失败,状态码: {resp.status_code}")
break
data = resp.json()
pins = data.get('pins', [])
if not pins:
logger.info("未获取到新的图片,分页结束")
break
# 过滤已下载的图片
new_pins = [pin for pin in pins if str(pin['pin_id']) not in self.downloaded_pins]
all_pins.extend(new_pins)
logger.info(f"第{page}页获取到{len(pins)}张,新增{len(new_pins)}张,累计{len(all_pins)}张")
# 更新游标
current_max = pins[-1]['pin_id']
# 达到最大数量限制则退出
if self.max_images and len(all_pins) >= self.max_images:
logger.info(f"已达到最大下载数量限制: {self.max_images}")
break
# 如果返回数量小于limit,说明是最后一页
if len(pins) < limit:
break
# 随机延时,防止请求过快
time.sleep(random.uniform(0.5, 1.2))
except Exception as e:
logger.error(f"请求异常: {e}")
break
return all_pins[:self.max_images] if self.max_images else all_pins
def download_single_image(self, pin: Dict, retry: int = 3) -> bool:
"""下载单张图片"""
pin_id = pin['pin_id']
file_info = pin.get('file', {})
img_key = file_info.get('key', '')
if not img_key:
logger.warning(f"pin_id {pin_id} 缺少图片key")
return False
# 构建图片URL(两种可能域名)
img_url = f"https://img.hb.aicdn.top/{img_key}"
# 备用: f"https://hbimg.huabanimg.com/{img_key}"
ext = img_key.split('.')[-1].split('?')[0].lower()
if ext not in ['jpg', 'jpeg', 'png', 'gif', 'webp']:
ext = 'jpg'
filename = f"{pin_id}.{ext}"
save_path = self.save_dir / filename
if save_path.exists():
logger.debug(f"文件已存在,跳过: {filename}")
return True
# 下载重试机制
for attempt in range(retry):
try:
headers = {
'Referer': 'https://huaban.com/',
'User-Agent': UserAgent().random,
}
img_resp = self.session.get(img_url, headers=headers, timeout=15)
if img_resp.status_code == 200:
# 校验是否为图片
content_type = img_resp.headers.get('content-type', '')
if 'image' in content_type or img_resp.content[:4] in [b'\\xff\\xd8\\xff\\xe0', b'\\x89PNG']:
with open(save_path, 'wb') as f:
f.write(img_resp.content)
# 保存成功后记录
self._save_downloaded_record(str(pin_id))
logger.success(f"下载成功: {filename}")
return True
else:
logger.warning(f"非图片内容,尝试{attempt+1}/{retry}")
else:
logger.warning(f"下载失败 {img_url},状态码: {img_resp.status_code}")
except Exception as e:
logger.error(f"下载异常: {e}, 尝试{attempt+1}/{retry}")
time.sleep(random.uniform(1, 2))
logger.error(f"下载失败(重试{retry}次): {filename}")
return False
def run(self):
"""主运行入口"""
logger.info(f"开始爬取画板 {self.board_id},保存目录: {self.save_dir}")
# 1. 获取所有图片信息
pins = self.fetch_pins()
logger.info(f"共获取到 {len(pins)} 张待下载图片")
if not pins:
logger.warning("没有获取到任何图片,请检查画板ID是否正确或网络情况")
return
# 2. 批量下载(带进度条)
success_count = 0
with tqdm(total=len(pins), desc="下载进度", unit="张") as pbar:
for pin in pins:
if self.download_single_image(pin):
success_count += 1
pbar.update(1)
# 每下载一张随机休眠,降低封IP风险
time.sleep(random.uniform(0.2, 0.6))
logger.info(f"爬取完成!成功下载 {success_count}/{len(pins)} 张图片,保存于 {self.save_dir}")
def main():
"""命令行入口"""
# 配置参数(修改这里)
BOARD_ID = "48902858" # 目标画板ID
SAVE_DIR = "./huaban_images"
MAX_IMAGES = 500 # 最大下载数量,None表示全部
USE_PROXY = False # 是否使用代理(需自行配置)
crawler = HuabanBoardCrawler(
board_id=BOARD_ID,
save_dir=SAVE_DIR,
max_images=MAX_IMAGES,
use_proxy=USE_PROXY
)
crawler.run()
if __name__ == "__main__":
main()



