十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

图解原理:搞定保存网页图片的5个死坑,代码跑通不报错

图解原理:搞定保存网页图片的5个死坑,代码跑通不报错 图解原理:搞定保存网页图片的5个死坑,代码跑通不报错 复制来的爬虫代码,一运行就报 403 Forbidden,或者图片全是乱码,改了半天参数还是不行?别急着删库跑路,这大概率不是代码逻辑写错了,而是你根本没搞懂浏览器渲染与底层请求的区别。今天不讲虚的,直接上图解原理,把保存网页图片这事儿里那些让你头秃的坑,一个个挖出来填平。 坑点一:静态HTML里根本找不到图片地址 很多新手的第一反应是 requests.get(url) 拿到 HTML,然后用正则或者 BeautifulSoup 去提取 img 标签的 src。结果发现,要么提取不到,要么提取出来全是 data:image/... 这种 Base64 编码,或者相对路径 /img/logo.png。 现象:控制台报错 FileNotFoundError 或者下载下来的文件打不开,后缀名是 .html 但内容全是二进制乱码。 根本原因:现代前端框架(React, Vue)或动态加载机制。图片地址往往不在初始 HTML 里,而是在 JS 执行后动态插入 DOM 的。或者是使用了 WebP 格式,传统解析器不识别。还有一种情况,服务器开启了防盗链(Referer Check),你直接用 requests 裸奔请求,服务器一看没有合法的来源标识,直接拒载。 正确写法对比: ❌ 错误写法:纯 HTTP 请求 + 正则 import requests import reurl = 'https://example.com/goods/123' html = requests.get(url).text # 简单粗暴的正则,遇到动态加载直接失效 imgs = re.findall(r'img.*?src=(.*?).*?', html) print(imgs) # 输出可能是空列表,或者一堆相对路径✅ 正确写法:Selenium/Playwright 模拟浏览器渲染 from selenium import webdriver from selenium.webdriver.chrome.options import Optionsoptions = Options() options.add_argument('--headless') # 无头模式,不弹窗 driver = webdriver.Chrome(options=options) driver.get('https://example.com/goods/123')# 等待图片加载完成,避免拿到占位图 driver.implicitly_wait(5)# 获取所有 img 标签 imgs = driver.find_elements_by_tag_name('img') for img in imgs:src = img.get_attribute('src')if src and src.startswith('http'):print(src)driver.quit()图解原理: 想象一下,requests 就像是你直接给网站服务器打了个电话问:“那个图片在哪?” 服务器回答:“没登录不给看。” 而 Selenium 就像是雇了一个真人演员,真的去网站点了点鼠标,屏幕渲染出来了,你再去截图或者读取内存里的数据。对于动态内容,后者才是正解。 坑点二:Referer 防盗链导致 403 错误 即使你拿到了图片的真实 URL,直接 requests.get(img_url) 依然可能失败。尤其是爬取一些大型电商、新闻网站时,你会发现明明 URL 是对的,却返回 403 Forbidden。 现象:状态码 403,响应头里可能有 X-Frame-Options 或者自定义的错误信息。在 Stack Overflow 上,这类问题常年占据 Python 爬虫板块的高票区,核心原因几乎都指向防盗链。 根本原因:服务器校验 Referer 头。浏览器在加载图片时,会自动带上“我是从哪个页面跳过来的”这个信息。如果服务器发现 Referer 不是它指定的域名,就认为你在恶意盗用资源,直接切断连接。 正确写法对比: ❌ 错误写法:忽略请求头 import requestsimg_url = 'https://cdn.example.com/pic/001.jpg' # 裸奔请求,没有携带任何身份信息 response = requests.get(img_url) if response.status_code != 200:print(fError: {response.status_code}) # 结果:Error: 403✅ 正确写法:伪造合法的 Referer 和 User-Agent import requestsimg_url = 'https://cdn.example.com/pic/001.jpg' headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36','Referer': 'https://www.example.com/goods/123' # 关键:指定来源页面 }response = requests.get(img_url, headers=headers) if response.status_code == 200:with open('001.jpg', 'wb') as f:f.write(response.content)print(保存成功) else:print(fError: {response.status_code})避坑建议: 不要硬编码一个固定的 User-Agent。有些高级的风控系统会检测 UA 的一致性(比如 UA 说是 Chrome 110,但请求头里的 Accept-Language 却是英文,而 IP 在国内,这种不一致容易触发风控)。尽量保持请求头的真实性和一致性。 坑点三:Base64 图片保存为 .jpg 打不开 有些网页(尤其是移动端适配较好的页面或小程序)为了减少 HTTP 请求次数,会把小图标或装饰图直接以 Base64 字符串的形式写在 src 属性里。 现象:提取到的 src 是以 data:image/png;base64,iVBORw0KGgo... 开头的长字符串。你直接把它当文件路径去读,或者直接把这一长串存进 .jpg 文件,用图片查看器打开,显示“文件已损坏”。 根本原因:Base64 是一种编码方式,不是二进制数据。你需要先解码,再写入文件。而且,MIME 类型(image/png, image/jpeg, image/webp)决定了文件的后缀名。如果你把 PNG 数据存成 .jpg,虽然有些查看器能猜出来,但在服务器端或严格校验的程序里会报错。 正确写法对比: ❌ 错误写法:直接写入 base64_str = data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAA...with open('img.jpg', 'w') as f:f.write(base64_str) # 结果:文件是文本格式,图片查看器无法识别✅ 正确写法:解码 + 动态后缀 import base64 import rebase64_str = data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAA...# 1. 提取 MIME 类型和纯 Base64 数据 match = re.match(r'data:(.*?);base64,(.*)', base64_str) if match:mime_type = match.group(1) # e.g., image/pngimg_data = match.group(2) # e.g., iVBORw0KGgo...# 2. 确定后缀名ext_map = {'image/png': '.png','image/jpeg': '.jpg','image/webp': '.webp'}ext = ext_map.get(mime_type, '.img')# 3. 解码并写入二进制try:decoded_data = base64.b64decode(img_data)filename = f'img_{abs(hash(img_data))}{ext}'with open(filename, 'wb') as f:f.write(decoded_data)print(fSaved: {filename})except Exception as e:print(fDecode Error: {e})图解原理: Base64 就像是用文字描述了一幅画。你不能把描述文字直接贴在墙上当画看(打不开),你得先根据描述把画还原出来(解码),然后再挂上去(写入二进制文件)。 坑点四:并发下载导致磁盘 I/O 阻塞或 IP 封禁 当你要保存成千上万张图片时,串行下载(一张接一张)效率极低。于是大家开始用 threading 或 asyncio 搞并发。结果要么电脑风扇狂转 CPU 占用 100%,要么刚跑一半,IP 就被服务器拉黑了,剩下的全变 403。 现象:程序运行速度极慢,或者中途大量报错 ConnectionResetError 或 429 Too Many Requests。 根本原因:线程竞争:Python 的 GIL(全局解释器锁)导致多线程在 CPU 密集型任务(如 Base64 解码)上并没有真正的并行加速,反而因为上下文切换开销变大。 缺乏限流:并发数设置过高(比如一开就是 100 个线程),瞬间对服务器产生巨大压力,触发服务器的速率限制(Rate Limiting)。 I/O 阻塞:在多线程环境下,如果没有做好文件句柄管理,容易出现写入冲突或资源未释放。正确写法对比: ❌ 错误写法:无限制并发 + 同步 I/O import threading import requestsdef download(url):r = requests.get(url)with open(url.split('/')[-1], 'wb') as f:f.write(r.content)urls = [fhttps://example.com/img/{i}.jpg for i in range(1000)] threads = [] for url in urls:t = threading.Thread(target=download, args=(url,))t.start()threads.append(t) # 瞬间启动1000个线程,服务器直接宕机或封IP for t in threads:t.join()✅ 正确写法:使用线程池 + 信号量限流 + 异步 I/O import asyncio import aiohttp import aiofiles import hashlib# 设置最大并发数,比如 20 SEM = asyncio.Semaphore(20)async def fetch_image(session, url, file_path):async with SEM: # 信号量控制并发try:async with session.get(url) as response:if response.status == 200:data = await response.read()async with aiofiles.open(file_path, 'wb') as f:await f.write(data)print(fSaved: {file_path})else:print(fFailed: {url} - {response.status})except Exception as e:print(fError: {url} - {str(e)})async def main():# 使用 aiohttp 的默认连接池,限制总连接数connector = aiohttp.TCPConnector(limit=50)timeout = aiohttp.ClientTimeout(total=30)async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:tasks = []urls = [fhttps://example.com/img/{i}.jpg for i in range(1000)]for url in urls:# 生成唯一文件名,避免覆盖file_name = hashlib.md5(url.encode()).hexdigest() + '.jpg'tasks.append(fetch_image(session, url, file_name))await asyncio.gather(*tasks)if __name__ == '__main__':asyncio.run(main())避坑建议:限流是关键:永远不要假设服务器能扛住你的并发。使用 Semaphore 或连接池限制来平滑请求速率。 异步优于多线程:对于 I/O 密集型任务(网络请求、文件读写),asyncio 配合 aiohttp 和 aiofiles 效率远高于多线程,且代码结构更清晰。 重试机制:加上 tenacity 库或简单的 try-except 重试逻辑,应对偶发的网络波动。坑点五:文件命名冲突与断点续传缺失 爬取过程中,如果网络抖动导致程序中断,或者同一张图片出现在不同页面(URL 相同但文件名相同),再次运行时要么覆盖旧文件,要么报错。 现象:程序中断后重新运行,之前的进度全部丢失,或者文件数量对不上。 根本原因:缺乏状态管理。没有记录哪些图片已经下载成功,哪些失败了。 正确做法:使用 SQLite 或 JSON 记录状态 import json import osclass ImageDownloader:def __init__(self, state_file='download_state.json'):self.state_file = state_fileself.state = self.load_state()def load_state(self):if os.path.exists(self.state_file):with open(self.state_file, 'r') as f:return json.load(f)return {}def save_state(self):with open(self.state_file, 'w') as f:json.dump(self.state, f, indent=2)def is_downloaded(self, url):return url in self.statedef mark_downloaded(self, url):self.state[url] = True# 每下载10张保存一次状态,防止丢失if len(self.state) % 10 == 0:self.save_state()# 使用示例 downloader = ImageDownloader() if not downloader.is_downloaded(img_url):# ... 执行下载逻辑 ...downloader.mark_downloaded(img_url)图解原理: 这就好比你搬砖,搬一块记一笔账。如果半路停电了,你重新来的时候,看一眼账本,就知道哪几块砖已经搬进屋了,只需要搬剩下的。如果没有账本,你就得从头开始搬,或者把已经搬好的砖再搬出来检查一下,效率极低。 总结与互动 保存网页图片看似简单,实则坑多。从静态解析到动态渲染,从防盗链绕过到并发控制,每一个环节都可能让你卡住。记住,图解原理不仅仅是看代码,更要理解代码背后的网络协议和计算机运行机制。 在 Stack Overflow 上,很多关于爬虫的问题,最终答案往往不是“换个库”,而是“检查你的请求头”或“等待页面加载完成”。 这个知识点你面试被问过吗? 比如让你设计一个高并发的图片下载器,你会怎么考虑去重、断点续传和 IP 池的问题?留言说说你的思路,看看有多少老铁踩过同样的坑。
返回列表