尧图网页设计 
1. 项目概述从“能爬”到“爬得好”的蜕变搞Python爬虫的朋友估计都经历过这么几个阶段一开始照着教程用requests加BeautifulSoup吭哧吭哧写几行代码能抓到网页上的文字就兴奋得不行接着发现很多网站有图片就想把图也扒下来结果要么是链接不对要么下了一堆破损文件再往后网站开始反爬了什么验证码、动态加载、请求头校验全来了直接给你返回403或者一堆乱码。这时候你才明白爬虫这活儿远不是发个请求、解析一下HTML那么简单尤其是图片爬虫它更像是一场与目标站点之间关于资源获取规则的“友好协商”。今天聊的“Python图片爬虫心得”就是把我这些年从踩坑到填坑从简单粗暴到精细运营的过程梳理出来。这不仅仅是技术实现更多的是策略选择、效率优化和风险规避的实战经验。无论你是想批量下载壁纸、收集电商商品图还是做数据集构建这些心得都能帮你少走弯路。核心就一句话爬虫的目标是稳定、高效、友好地拿到数据而不是把对方服务器搞垮。2. 核心思路与架构设计不止于requests.get()很多人觉得图片爬虫就是找到img标签的src属性然后循环下载。这个思路在十年前可能还行得通但现在你需要一个更健壮的架构。2.1 核心流程拆解一个完整的图片爬虫应该像一条设计精良的流水线每个环节都有其职责和容错机制任务调度与URL管理从哪里开始爬如何发现新的图片页面如何避免重复爬取和死循环这需要一个URL队列或调度器。页面抓取器负责发送HTTP请求获取网页HTML。这是与网站正面交锋的第一关需要处理网络异常、编码问题、反爬策略。内容解析器从HTML中精准定位图片链接。不仅要找img src...还要考虑懒加载># 示例使用BeautifulSoup和lxml的XPath查找多种图片源 from bs4 import BeautifulSoup import requests from lxml import etree html requests.get(https://example.com).text # 方法1: BeautifulSoup 查找 soup BeautifulSoup(html, lxml) img_tags soup.find_all(img) for img in img_tags: # 优先取>from urllib.parse import urljoin, urlparse base_url https://www.example.com/gallery img_relative /static/img1.jpg img_protocol_relative //cdn.example.com/img2.png img_relative_dot_dot ../assets/img3.jpg # 使用urljoin进行补全它是处理这个问题的标准工具 print(urljoin(base_url, img_relative)) # https://www.example.com/static/img1.jpg print(urljoin(base_url, img_protocol_relative)) # https://cdn.example.com/img2.png (继承base_url的协议) print(urljoin(base_url, img_relative_dot_dot)) # https://www.example.com/assets/img3.jpg # 一个实用的清洗函数 def clean_image_url(raw_url, page_url): if not raw_url: return None # 去除URL中可能影响下载的查询参数如缩略图参数 parsed urlparse(raw_url) # 有些图片链接带尺寸参数如 ?x-oss-processimage/resize,w_200 # 是否需要去除取决于你是否想要原图。这里提供一个去除常见缩略参数的思路 # 更常见的做法是如果发现参数中有‘thumbnail’、‘small’、‘w_’等字样尝试寻找原图链接模式 cleaned_url parsed._replace(query).geturl() if thumbnail in parsed.query else raw_url # 补全为绝对URL absolute_url urljoin(page_url, cleaned_url) return absolute_url心得urljoin是处理相对路径的利器。但要小心有些网站的图片原图和高清图藏在不同的URL模式里可能需要你观察规律比如把thumbnail.jpg换成large.jpg或直接去掉查询字符串。3.3 稳健的下载与存储命名、去重与断点续传下载成百上千张图片时稳定的下载逻辑和清晰的文件管理至关重要。import os import hashlib from pathlib import Path import aiohttp import asyncio async def download_image(session, url, save_dir, filenameNone, max_retries3): 异步下载图片支持重试和自定义文件名 if not filename: # 使用URL的MD5值作为默认文件名避免特殊字符和重复 url_hash hashlib.md5(url.encode()).hexdigest() # 从URL中猜测或通过响应头获取文件扩展名 ext os.path.splitext(urlparse(url).path)[1] ext ext if ext.lower() in [.jpg, .jpeg, .png, .gif, .webp] else .jpg filename f{url_hash}{ext} filepath Path(save_dir) / filename # 如果文件已存在跳过下载简易去重 if filepath.exists(): print(fFile exists, skip: {filename}) return filepath for attempt in range(max_retries): try: async with session.get(url, timeoutaiohttp.ClientTimeout(total30)) as resp: if resp.status 200: # 检查Content-Type确认是图片 content_type resp.headers.get(Content-Type, ) if image not in content_type: print(fURL not an image: {url}, Content-Type: {content_type}) return None # 读取并保存图片数据 image_data await resp.read() # 可以在这里添加简单的图片数据验证例如检查文件头 if image_data[:4] b\xff\xd8\xff\xe0 or image_data[:8] b\x89PNG\r\n\x1a\n: filepath.write_bytes(image_data) print(fDownloaded: {filename}) return filepath else: print(fInvalid image data from: {url}) else: print(fFailed to download {url}, status: {resp.status}) except (aiohttp.ClientError, asyncio.TimeoutError) as e: print(fAttempt {attempt1} failed for {url}: {e}) if attempt max_retries - 1: await asyncio.sleep(2 ** attempt) # 指数退避重试 else: print(fMax retries exceeded for {url}) return None return None # 使用示例 async def main(): async with aiohttp.ClientSession() as session: tasks [] for img_url in image_url_list: task download_image(session, img_url, ./downloaded_images) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue)注意事项命名策略不要直接用URL最后一段作为文件名因为可能包含乱码或重复。使用MD5哈希或SHA1哈希值是常见的去重和规范化命名方法。也可以考虑按“站点/分类/日期”的目录结构存储。文件去重除了在下载前检查文件是否存在更严谨的做法是在内存或数据库中记录已下载图片的哈希值如MD5防止内容相同的图片因URL不同而被重复下载。设置超时和重试网络不稳定是常态。必须为请求设置超时如30秒并实现带有退避策略的重试机制例如第一次等待2秒第二次4秒第三次8秒。验证文件内容下载完成后最好能验证文件是否是有效的图片格式。可以通过检查文件头Magic Number或使用PILPillow库尝试打开图片来验证。控制并发和频率使用信号量asyncio.Semaphore限制同时发起的请求数并在请求间添加随机延迟asyncio.sleep(random.uniform(1, 3))这是对目标网站最基本的尊重也能降低被封IP的风险。4. 高级策略与反反爬虫实战当你的爬虫开始规律性地访问一个网站时很快就会触发它的防御机制。4.1 请求头伪装让自己看起来像“人”最基本的反爬措施是检查请求头User-Agent, Referer, Accept-Encoding等。一个来自Pythonrequests的默认User-Agent很容易被识别和屏蔽。import fake_useragent import aiohttp # 使用 fake-useragent 库随机生成浏览器User-Agent ua fake_useragent.UserAgent() headers { User-Agent: ua.random, # 每次请求使用随机UA Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Referer: https://www.google.com/, # 模拟从搜索引擎跳转而来 Connection: keep-alive, } async with aiohttp.ClientSession(headersheaders) as session: # 使用这个session发起请求心得User-Agent池是必备的。Referer字段也很有用适当设置可以让请求看起来更自然。有些网站还会检查Accept、Accept-Language等字段尽量模拟真实浏览器的请求头。4.2 会话Session与Cookie管理有些网站需要登录后才能查看图片或者通过Cookie来跟踪会话状态。import aiohttp async def login_and_crawl(): # 1. 创建会话 async with aiohttp.ClientSession() as session: # 2. 模拟登录如果需要 login_data {username: your_user, password: your_pass} login_url https://example.com/login async with session.post(login_url, datalogin_data) as resp: if resp.status 200: print(Login successful) # 登录成功后session会自动管理后续请求的cookies else: print(Login failed) return # 3. 使用同一个session访问需要登录的图片页面 gallery_url https://example.com/private/gallery async with session.get(gallery_url) as resp: html await resp.text() # 解析并下载图片...关键点使用aiohttp.ClientSession或requests.Session对象来维持会话。它会在多次请求间自动处理Cookie就像浏览器一样。对于更复杂的登录流程如带有CSRF token的表单你需要先访问登录页解析出token再构造登录请求。4.3 处理动态加载与JavaScript渲染越来越多的网站使用前端框架如React, Vue构建图片数据通过AjaxXHR/Fetch请求获取或者页面内容在JavaScript执行后才渲染。方案一直接分析Ajax接口这是最高效的方法。打开浏览器的开发者工具F12切换到“网络”Network选项卡过滤XHR或Fetch请求刷新页面观察哪些请求返回了图片数据通常是JSON格式。然后你的爬虫直接模拟这些Ajax请求。# 假设你发现获取图片列表的API是GET https://api.example.com/images?page1 import json async def fetch_via_api(session, page): api_url fhttps://api.example.com/images?page{page} headers {X-Requested-With: XMLHttpRequest} # 有时需要这个头 async with session.get(api_url, headersheaders) as resp: if resp.status 200: data await resp.json() # 直接解析JSON image_urls [item[url] for item in data[images]] return image_urls优点速度快数据干净。缺点需要一定的逆向分析能力且API可能变更或有签名验证。方案二使用无头浏览器当网站混淆严重或无法直接找到API时无头浏览器是终极武器。Selenium、Playwright、Pyppeteer可以控制一个真实的浏览器内核如Chrome来加载页面执行JavaScript等所有内容渲染完毕后再提取。# 使用Playwright示例 from playwright.async_api import async_playwright async def crawl_with_playwright(): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) # 无头模式 page await browser.new_page() await page.goto(https://example.com/dynamic-gallery) # 等待图片加载完成例如等待某个图片元素出现 await page.wait_for_selector(img.loaded, timeout10000) # 获取渲染后的页面HTML html await page.content() # 或者直接获取图片元素的属性 img_elements await page.query_selector_all(img) for img in img_elements: src await img.get_attribute(src) data_src await img.get_attribute(data-src) img_url data_src or src print(img_url) await browser.close()优点通杀所有网站所见即所得。缺点资源消耗大内存、CPU速度慢不稳定因素多。心得优先尝试方案一。直接调用API的效率是无头浏览器的几十倍甚至上百倍。只有在前者走不通时才考虑使用无头浏览器并且要做好错误处理和资源管理。4.4 IP代理与速率限制即使你伪装得再好一个IP地址在短时间内发起过多请求也极易被封锁。速率限制Rate Limiting在你的爬虫逻辑里加入延迟。这是最基本的道德和技术要求。import asyncio import random async def polite_crawler(url_list): async with aiohttp.ClientSession() as session: for url in url_list: await download_image(session, url) # 你的下载函数 # 在请求之间添加随机延迟模拟人类操作 await asyncio.sleep(random.uniform(1, 3)) # 延迟1到3秒IP代理池当单IP被禁或需要更高频率抓取时必须使用代理IP。你可以购买付费代理服务或者自建代理池维护成本高。proxy http://user:passproxy_ip:proxy_port # 格式协议://用户名:密码IP:端口 async with aiohttp.ClientSession() as session: async with session.get(url, proxyproxy) as resp: ...使用代理的注意事项质量免费代理大多不稳定、速度慢。生产环境建议使用付费的优质代理服务。验证每次使用代理前最好先访问一个测试网站如http://httpbin.org/ip检查代理是否有效且返回的是代理IP而非本机IP。轮换维护一个代理IP列表每次请求随机选取或按策略轮换并在代理失效时及时剔除。5. 工程化与最佳实践当爬虫项目从脚本升级为需要长期运行的系统时工程化思维就变得非常重要。5.1 使用成熟的爬虫框架Scrapy对于中型以上、结构复杂的爬虫项目强烈建议使用Scrapy。它是一个为爬虫而生的异步框架内置了请求调度、去重、管道Pipeline用于处理下载的数据、中间件Middleware用于处理请求和响应等组件。# 一个简单的Scrapy Spider示例用于爬取图片 import scrapy from scrapy.pipelines.images import ImagesPipeline from itemadapter import ItemAdapter class MyImageSpider(scrapy.Spider): name image_spider start_urls [https://example.com/gallery] def parse(self, response): # 解析图片页面获取图片URL image_urls response.xpath(//img/data-src).getall() for img_url in image_urls: yield {image_urls: [response.urljoin(img_url)]} # 将图片URL交给Pipeline处理 # 翻页逻辑 next_page response.xpath(//a[classnext]/href).get() if next_page: yield response.follow(next_page, self.parse) # 自定义ImagesPipeline可以重写文件存储路径、命名规则等 class CustomImagesPipeline(ImagesPipeline): def file_path(self, request, responseNone, infoNone, *, itemNone): # 自定义文件存储路径和名称 image_guid hashlib.sha1(request.url.encode()).hexdigest() return ffull/{image_guid}.jpgScrapy的优势内置强大组件自动去重DupeFilter、请求优先级调度、并发控制、日志统计等。清晰的架构Spider负责解析Item定义数据结构Pipeline负责后处理如下载图片、存入数据库中间件负责全局钩子。扩展性强可以通过中间件轻松集成代理、更换User-Agent、处理Cookie等。生产效率高用scrapy genspider命令快速生成模板社区插件丰富。心得对于一次性、简单的任务自己写脚本更灵活。但对于需要持续维护、规模较大、有复杂处理流程如下载、清洗、存储的项目Scrapy能节省你大量的开发和时间成本让代码更健壮、更易维护。5.2 错误处理与日志记录健壮的爬虫必须能妥善处理各种异常并留下清晰的日志方便排查问题。import logging import sys # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(crawler.log), # 输出到文件 logging.StreamHandler(sys.stdout) # 输出到控制台 ] ) logger logging.getLogger(__name__) async def robust_download(session, url): try: async with session.get(url, timeout30) as resp: resp.raise_for_status() # 如果状态码不是200抛出HTTPError return await resp.read() except aiohttp.ClientResponseError as e: logger.error(fHTTP error for {url}: {e.status} - {e.message}) if e.status 404: logger.warning(fImage not found: {url}) elif e.status 403: logger.error(fAccess forbidden, possible anti-bot: {url}) # 这里可以触发IP更换或暂停爬取 return None except asyncio.TimeoutError: logger.error(fTimeout for {url}) return None except Exception as e: logger.exception(fUnexpected error for {url}: {e}) # 记录异常堆栈 return None关键点区分错误类型网络超时、HTTP 404/403/500、连接拒绝等需要不同的处理策略重试、跳过、报警。记录足够的信息日志中应包含URL、时间戳、错误类型、响应状态码等方便事后分析。设置警报对于关键任务当连续出现特定错误如大量403时可以通过邮件、钉钉、微信机器人等方式发送警报。5.3 数据存储与去重优化下载的图片如何组织如何避免重复下载存储结构不要把所有图片扔进一个文件夹。建议按“域名/日期/分类”这样的层级存放。例如./data/example.com/2023-10-27/product_thumbnails/。去重策略URL去重在爬取前将已爬取的URL存入一个集合set或布隆过滤器Bloom Filter节省内存。这是最直接的方式但无法应对同一张图片对应不同URL的情况。内容去重下载图片后计算其哈希值如MD5、SHA1、感知哈希pHash将哈希值存入数据库或集合。下次下载前先计算待下载图片的哈希值可能需要先下载一小部分数据进行比对。这种方式更准确但计算开销大。数据库记录对于大规模爬虫使用数据库如SQLite, MySQL, MongoDB记录任务状态URL、状态、下载时间、文件路径、哈希值是标准做法。可以方便地实现断点续爬、状态查询和去重。5.4 法律与伦理边界这是最重要的一条心得。爬虫技术本身中性但如何使用它决定了性质。遵守robots.txt在爬取前访问网站的/robots.txt文件查看哪些目录是允许或禁止爬取的。使用Python的urllib.robotparser可以方便地解析。尊重版权明确你爬取的图片的版权归属。个人学习、研究使用通常问题不大但未经授权用于商业用途、公开传播或训练AI模型可能构成侵权。不要造成伤害严格控制请求频率避免对目标网站的正常运营造成影响DDos攻击效果。你的爬虫应该像一个有礼貌的访客而不是一群横冲直撞的野牛。查看服务条款很多网站的用户协议中明确禁止自动化数据抓取。6. 实战案例构建一个简单的异步图片爬虫让我们把上面的知识点串起来写一个针对静态图片画廊的完整示例。import asyncio import aiohttp from bs4 import BeautifulSoup from urllib.parse import urljoin, urlparse import hashlib import os from pathlib import Path import random import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class SimpleAsyncImageCrawler: def __init__(self, start_url, concurrency5, download_dir./downloads): self.start_url start_url self.seen_urls set() # 已发现的URL集合用于去重 self.download_dir Path(download_dir) self.download_dir.mkdir(parentsTrue, exist_okTrue) self.semaphore asyncio.Semaphore(concurrency) # 控制并发数 self.client_timeout aiohttp.ClientTimeout(total60) async def fetch_html(self, session, url): 获取页面HTML async with self.semaphore: # 控制并发 try: headers {User-Agent: Mozilla/5.0...} # 应使用动态UA async with session.get(url, headersheaders, timeoutself.client_timeout) as resp: resp.raise_for_status() # 注意编码有些网站不是utf-8 html await resp.text(encodingutf-8, errorsignore) await asyncio.sleep(random.uniform(1, 2)) # 礼貌延迟 return html except Exception as e: logger.error(fFailed to fetch {url}: {e}) return None def extract_image_urls(self, html, base_url): 从HTML中提取图片URL if not html: return [] soup BeautifulSoup(html, lxml) img_urls [] for img in soup.find_all(img): # 优先考虑懒加载属性 raw_url img.get(data-src) or img.get(data-original) or img.get(src) if raw_url: # 清洗和补全URL absolute_url urljoin(base_url, raw_url) # 简单的过滤比如只抓取jpg, png, webp if any(absolute_url.lower().endswith(ext) for ext in [.jpg, .jpeg, .png, .webp, .gif]): img_urls.append(absolute_url) return list(set(img_urls)) # 简单去重 async def download_image(self, session, img_url): 下载单张图片 file_hash hashlib.md5(img_url.encode()).hexdigest()[:12] ext os.path.splitext(urlparse(img_url).path)[1] or .jpg filename f{file_hash}{ext} filepath self.download_dir / filename if filepath.exists(): logger.info(fSkip existing: {filename}) return try: async with session.get(img_url, timeoutself.client_timeout) as resp: if resp.status 200: image_data await resp.read() # 简单验证是否为图片 if image_data[:3] b\xff\xd8\xff or image_data[:8] b\x89PNG\r\n\x1a\n: filepath.write_bytes(image_data) logger.info(fDownloaded: {filename} from {img_url}) else: logger.warning(fInvalid image data from {img_url}) else: logger.warning(fFailed to download {img_url}, status: {resp.status}) except Exception as e: logger.error(fError downloading {img_url}: {e}) async def crawl_page(self, session, url): 爬取单个页面提取图片并发现新链接简单示例未深入实现链接发现 logger.info(fCrawling: {url}) html await self.fetch_html(session, url) if not html: return [] # 提取本页图片 image_urls self.extract_image_urls(html, url) download_tasks [self.download_image(session, img_url) for img_url in image_urls] if download_tasks: await asyncio.gather(*download_tasks, return_exceptionsTrue) # 这里可以添加发现并递归爬取新页面的逻辑 # 例如提取所有a href过滤出同域名下的链接加入任务队列 # 注意需要深度控制和去重避免无限循环 new_page_urls [] # ... (链接发现逻辑) return new_page_urls async def run(self): 主运行函数 connector aiohttp.TCPConnector(limit100, sslFalse) # 调整连接器参数 async with aiohttp.ClientSession(connectorconnector) as session: # 初始任务队列 to_crawl [self.start_url] self.seen_urls.add(self.start_url) while to_crawl: current_url to_crawl.pop(0) new_urls await self.crawl_page(session, current_url) # 将新发现的、未爬取过的页面加入队列 for new_url in new_urls: if new_url not in self.seen_urls: self.seen_urls.add(new_url) to_crawl.append(new_url) # 控制整体爬取速度 await asyncio.sleep(random.uniform(2, 4)) logger.info(Crawling finished.) if __name__ __main__: # 使用示例请替换为你要爬取的画廊起始页 crawler SimpleAsyncImageCrawler(start_urlhttps://example.com/gallery, concurrency3) asyncio.run(crawler.run())这个案例集成了异步请求、链接提取、图片下载、基础去重和礼貌延迟。它只是一个起点你可以根据实际网站结构丰富其链接发现逻辑、错误处理、代理集成等功能。爬虫是一个不断学习和适应的过程。网站技术在变反爬策略在升级你的爬虫也需要持续迭代。保持耐心仔细分析尊重规则你就能高效、稳定地获取到所需的网络资源。记住最好的爬虫是那个能长期稳定运行、不被发现、也不打扰别人的爬虫。 Python图片爬虫进阶：从requests到异步架构与反爬策略实战
1. 项目概述从“能爬”到“爬得好”的蜕变搞Python爬虫的朋友估计都经历过这么几个阶段一开始照着教程用requests加BeautifulSoup吭哧吭哧写几行代码能抓到网页上的文字就兴奋得不行接着发现很多网站有图片就想把图也扒下来结果要么是链接不对要么下了一堆破损文件再往后网站开始反爬了什么验证码、动态加载、请求头校验全来了直接给你返回403或者一堆乱码。这时候你才明白爬虫这活儿远不是发个请求、解析一下HTML那么简单尤其是图片爬虫它更像是一场与目标站点之间关于资源获取规则的“友好协商”。今天聊的“Python图片爬虫心得”就是把我这些年从踩坑到填坑从简单粗暴到精细运营的过程梳理出来。这不仅仅是技术实现更多的是策略选择、效率优化和风险规避的实战经验。无论你是想批量下载壁纸、收集电商商品图还是做数据集构建这些心得都能帮你少走弯路。核心就一句话爬虫的目标是稳定、高效、友好地拿到数据而不是把对方服务器搞垮。2. 核心思路与架构设计不止于requests.get()很多人觉得图片爬虫就是找到img标签的src属性然后循环下载。这个思路在十年前可能还行得通但现在你需要一个更健壮的架构。2.1 核心流程拆解一个完整的图片爬虫应该像一条设计精良的流水线每个环节都有其职责和容错机制任务调度与URL管理从哪里开始爬如何发现新的图片页面如何避免重复爬取和死循环这需要一个URL队列或调度器。页面抓取器负责发送HTTP请求获取网页HTML。这是与网站正面交锋的第一关需要处理网络异常、编码问题、反爬策略。内容解析器从HTML中精准定位图片链接。不仅要找img src...还要考虑懒加载># 示例使用BeautifulSoup和lxml的XPath查找多种图片源 from bs4 import BeautifulSoup import requests from lxml import etree html requests.get(https://example.com).text # 方法1: BeautifulSoup 查找 soup BeautifulSoup(html, lxml) img_tags soup.find_all(img) for img in img_tags: # 优先取>from urllib.parse import urljoin, urlparse base_url https://www.example.com/gallery img_relative /static/img1.jpg img_protocol_relative //cdn.example.com/img2.png img_relative_dot_dot ../assets/img3.jpg # 使用urljoin进行补全它是处理这个问题的标准工具 print(urljoin(base_url, img_relative)) # https://www.example.com/static/img1.jpg print(urljoin(base_url, img_protocol_relative)) # https://cdn.example.com/img2.png (继承base_url的协议) print(urljoin(base_url, img_relative_dot_dot)) # https://www.example.com/assets/img3.jpg # 一个实用的清洗函数 def clean_image_url(raw_url, page_url): if not raw_url: return None # 去除URL中可能影响下载的查询参数如缩略图参数 parsed urlparse(raw_url) # 有些图片链接带尺寸参数如 ?x-oss-processimage/resize,w_200 # 是否需要去除取决于你是否想要原图。这里提供一个去除常见缩略参数的思路 # 更常见的做法是如果发现参数中有‘thumbnail’、‘small’、‘w_’等字样尝试寻找原图链接模式 cleaned_url parsed._replace(query).geturl() if thumbnail in parsed.query else raw_url # 补全为绝对URL absolute_url urljoin(page_url, cleaned_url) return absolute_url心得urljoin是处理相对路径的利器。但要小心有些网站的图片原图和高清图藏在不同的URL模式里可能需要你观察规律比如把thumbnail.jpg换成large.jpg或直接去掉查询字符串。3.3 稳健的下载与存储命名、去重与断点续传下载成百上千张图片时稳定的下载逻辑和清晰的文件管理至关重要。import os import hashlib from pathlib import Path import aiohttp import asyncio async def download_image(session, url, save_dir, filenameNone, max_retries3): 异步下载图片支持重试和自定义文件名 if not filename: # 使用URL的MD5值作为默认文件名避免特殊字符和重复 url_hash hashlib.md5(url.encode()).hexdigest() # 从URL中猜测或通过响应头获取文件扩展名 ext os.path.splitext(urlparse(url).path)[1] ext ext if ext.lower() in [.jpg, .jpeg, .png, .gif, .webp] else .jpg filename f{url_hash}{ext} filepath Path(save_dir) / filename # 如果文件已存在跳过下载简易去重 if filepath.exists(): print(fFile exists, skip: {filename}) return filepath for attempt in range(max_retries): try: async with session.get(url, timeoutaiohttp.ClientTimeout(total30)) as resp: if resp.status 200: # 检查Content-Type确认是图片 content_type resp.headers.get(Content-Type, ) if image not in content_type: print(fURL not an image: {url}, Content-Type: {content_type}) return None # 读取并保存图片数据 image_data await resp.read() # 可以在这里添加简单的图片数据验证例如检查文件头 if image_data[:4] b\xff\xd8\xff\xe0 or image_data[:8] b\x89PNG\r\n\x1a\n: filepath.write_bytes(image_data) print(fDownloaded: {filename}) return filepath else: print(fInvalid image data from: {url}) else: print(fFailed to download {url}, status: {resp.status}) except (aiohttp.ClientError, asyncio.TimeoutError) as e: print(fAttempt {attempt1} failed for {url}: {e}) if attempt max_retries - 1: await asyncio.sleep(2 ** attempt) # 指数退避重试 else: print(fMax retries exceeded for {url}) return None return None # 使用示例 async def main(): async with aiohttp.ClientSession() as session: tasks [] for img_url in image_url_list: task download_image(session, img_url, ./downloaded_images) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue)注意事项命名策略不要直接用URL最后一段作为文件名因为可能包含乱码或重复。使用MD5哈希或SHA1哈希值是常见的去重和规范化命名方法。也可以考虑按“站点/分类/日期”的目录结构存储。文件去重除了在下载前检查文件是否存在更严谨的做法是在内存或数据库中记录已下载图片的哈希值如MD5防止内容相同的图片因URL不同而被重复下载。设置超时和重试网络不稳定是常态。必须为请求设置超时如30秒并实现带有退避策略的重试机制例如第一次等待2秒第二次4秒第三次8秒。验证文件内容下载完成后最好能验证文件是否是有效的图片格式。可以通过检查文件头Magic Number或使用PILPillow库尝试打开图片来验证。控制并发和频率使用信号量asyncio.Semaphore限制同时发起的请求数并在请求间添加随机延迟asyncio.sleep(random.uniform(1, 3))这是对目标网站最基本的尊重也能降低被封IP的风险。4. 高级策略与反反爬虫实战当你的爬虫开始规律性地访问一个网站时很快就会触发它的防御机制。4.1 请求头伪装让自己看起来像“人”最基本的反爬措施是检查请求头User-Agent, Referer, Accept-Encoding等。一个来自Pythonrequests的默认User-Agent很容易被识别和屏蔽。import fake_useragent import aiohttp # 使用 fake-useragent 库随机生成浏览器User-Agent ua fake_useragent.UserAgent() headers { User-Agent: ua.random, # 每次请求使用随机UA Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Referer: https://www.google.com/, # 模拟从搜索引擎跳转而来 Connection: keep-alive, } async with aiohttp.ClientSession(headersheaders) as session: # 使用这个session发起请求心得User-Agent池是必备的。Referer字段也很有用适当设置可以让请求看起来更自然。有些网站还会检查Accept、Accept-Language等字段尽量模拟真实浏览器的请求头。4.2 会话Session与Cookie管理有些网站需要登录后才能查看图片或者通过Cookie来跟踪会话状态。import aiohttp async def login_and_crawl(): # 1. 创建会话 async with aiohttp.ClientSession() as session: # 2. 模拟登录如果需要 login_data {username: your_user, password: your_pass} login_url https://example.com/login async with session.post(login_url, datalogin_data) as resp: if resp.status 200: print(Login successful) # 登录成功后session会自动管理后续请求的cookies else: print(Login failed) return # 3. 使用同一个session访问需要登录的图片页面 gallery_url https://example.com/private/gallery async with session.get(gallery_url) as resp: html await resp.text() # 解析并下载图片...关键点使用aiohttp.ClientSession或requests.Session对象来维持会话。它会在多次请求间自动处理Cookie就像浏览器一样。对于更复杂的登录流程如带有CSRF token的表单你需要先访问登录页解析出token再构造登录请求。4.3 处理动态加载与JavaScript渲染越来越多的网站使用前端框架如React, Vue构建图片数据通过AjaxXHR/Fetch请求获取或者页面内容在JavaScript执行后才渲染。方案一直接分析Ajax接口这是最高效的方法。打开浏览器的开发者工具F12切换到“网络”Network选项卡过滤XHR或Fetch请求刷新页面观察哪些请求返回了图片数据通常是JSON格式。然后你的爬虫直接模拟这些Ajax请求。# 假设你发现获取图片列表的API是GET https://api.example.com/images?page1 import json async def fetch_via_api(session, page): api_url fhttps://api.example.com/images?page{page} headers {X-Requested-With: XMLHttpRequest} # 有时需要这个头 async with session.get(api_url, headersheaders) as resp: if resp.status 200: data await resp.json() # 直接解析JSON image_urls [item[url] for item in data[images]] return image_urls优点速度快数据干净。缺点需要一定的逆向分析能力且API可能变更或有签名验证。方案二使用无头浏览器当网站混淆严重或无法直接找到API时无头浏览器是终极武器。Selenium、Playwright、Pyppeteer可以控制一个真实的浏览器内核如Chrome来加载页面执行JavaScript等所有内容渲染完毕后再提取。# 使用Playwright示例 from playwright.async_api import async_playwright async def crawl_with_playwright(): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) # 无头模式 page await browser.new_page() await page.goto(https://example.com/dynamic-gallery) # 等待图片加载完成例如等待某个图片元素出现 await page.wait_for_selector(img.loaded, timeout10000) # 获取渲染后的页面HTML html await page.content() # 或者直接获取图片元素的属性 img_elements await page.query_selector_all(img) for img in img_elements: src await img.get_attribute(src) data_src await img.get_attribute(data-src) img_url data_src or src print(img_url) await browser.close()优点通杀所有网站所见即所得。缺点资源消耗大内存、CPU速度慢不稳定因素多。心得优先尝试方案一。直接调用API的效率是无头浏览器的几十倍甚至上百倍。只有在前者走不通时才考虑使用无头浏览器并且要做好错误处理和资源管理。4.4 IP代理与速率限制即使你伪装得再好一个IP地址在短时间内发起过多请求也极易被封锁。速率限制Rate Limiting在你的爬虫逻辑里加入延迟。这是最基本的道德和技术要求。import asyncio import random async def polite_crawler(url_list): async with aiohttp.ClientSession() as session: for url in url_list: await download_image(session, url) # 你的下载函数 # 在请求之间添加随机延迟模拟人类操作 await asyncio.sleep(random.uniform(1, 3)) # 延迟1到3秒IP代理池当单IP被禁或需要更高频率抓取时必须使用代理IP。你可以购买付费代理服务或者自建代理池维护成本高。proxy http://user:passproxy_ip:proxy_port # 格式协议://用户名:密码IP:端口 async with aiohttp.ClientSession() as session: async with session.get(url, proxyproxy) as resp: ...使用代理的注意事项质量免费代理大多不稳定、速度慢。生产环境建议使用付费的优质代理服务。验证每次使用代理前最好先访问一个测试网站如http://httpbin.org/ip检查代理是否有效且返回的是代理IP而非本机IP。轮换维护一个代理IP列表每次请求随机选取或按策略轮换并在代理失效时及时剔除。5. 工程化与最佳实践当爬虫项目从脚本升级为需要长期运行的系统时工程化思维就变得非常重要。5.1 使用成熟的爬虫框架Scrapy对于中型以上、结构复杂的爬虫项目强烈建议使用Scrapy。它是一个为爬虫而生的异步框架内置了请求调度、去重、管道Pipeline用于处理下载的数据、中间件Middleware用于处理请求和响应等组件。# 一个简单的Scrapy Spider示例用于爬取图片 import scrapy from scrapy.pipelines.images import ImagesPipeline from itemadapter import ItemAdapter class MyImageSpider(scrapy.Spider): name image_spider start_urls [https://example.com/gallery] def parse(self, response): # 解析图片页面获取图片URL image_urls response.xpath(//img/data-src).getall() for img_url in image_urls: yield {image_urls: [response.urljoin(img_url)]} # 将图片URL交给Pipeline处理 # 翻页逻辑 next_page response.xpath(//a[classnext]/href).get() if next_page: yield response.follow(next_page, self.parse) # 自定义ImagesPipeline可以重写文件存储路径、命名规则等 class CustomImagesPipeline(ImagesPipeline): def file_path(self, request, responseNone, infoNone, *, itemNone): # 自定义文件存储路径和名称 image_guid hashlib.sha1(request.url.encode()).hexdigest() return ffull/{image_guid}.jpgScrapy的优势内置强大组件自动去重DupeFilter、请求优先级调度、并发控制、日志统计等。清晰的架构Spider负责解析Item定义数据结构Pipeline负责后处理如下载图片、存入数据库中间件负责全局钩子。扩展性强可以通过中间件轻松集成代理、更换User-Agent、处理Cookie等。生产效率高用scrapy genspider命令快速生成模板社区插件丰富。心得对于一次性、简单的任务自己写脚本更灵活。但对于需要持续维护、规模较大、有复杂处理流程如下载、清洗、存储的项目Scrapy能节省你大量的开发和时间成本让代码更健壮、更易维护。5.2 错误处理与日志记录健壮的爬虫必须能妥善处理各种异常并留下清晰的日志方便排查问题。import logging import sys # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(crawler.log), # 输出到文件 logging.StreamHandler(sys.stdout) # 输出到控制台 ] ) logger logging.getLogger(__name__) async def robust_download(session, url): try: async with session.get(url, timeout30) as resp: resp.raise_for_status() # 如果状态码不是200抛出HTTPError return await resp.read() except aiohttp.ClientResponseError as e: logger.error(fHTTP error for {url}: {e.status} - {e.message}) if e.status 404: logger.warning(fImage not found: {url}) elif e.status 403: logger.error(fAccess forbidden, possible anti-bot: {url}) # 这里可以触发IP更换或暂停爬取 return None except asyncio.TimeoutError: logger.error(fTimeout for {url}) return None except Exception as e: logger.exception(fUnexpected error for {url}: {e}) # 记录异常堆栈 return None关键点区分错误类型网络超时、HTTP 404/403/500、连接拒绝等需要不同的处理策略重试、跳过、报警。记录足够的信息日志中应包含URL、时间戳、错误类型、响应状态码等方便事后分析。设置警报对于关键任务当连续出现特定错误如大量403时可以通过邮件、钉钉、微信机器人等方式发送警报。5.3 数据存储与去重优化下载的图片如何组织如何避免重复下载存储结构不要把所有图片扔进一个文件夹。建议按“域名/日期/分类”这样的层级存放。例如./data/example.com/2023-10-27/product_thumbnails/。去重策略URL去重在爬取前将已爬取的URL存入一个集合set或布隆过滤器Bloom Filter节省内存。这是最直接的方式但无法应对同一张图片对应不同URL的情况。内容去重下载图片后计算其哈希值如MD5、SHA1、感知哈希pHash将哈希值存入数据库或集合。下次下载前先计算待下载图片的哈希值可能需要先下载一小部分数据进行比对。这种方式更准确但计算开销大。数据库记录对于大规模爬虫使用数据库如SQLite, MySQL, MongoDB记录任务状态URL、状态、下载时间、文件路径、哈希值是标准做法。可以方便地实现断点续爬、状态查询和去重。5.4 法律与伦理边界这是最重要的一条心得。爬虫技术本身中性但如何使用它决定了性质。遵守robots.txt在爬取前访问网站的/robots.txt文件查看哪些目录是允许或禁止爬取的。使用Python的urllib.robotparser可以方便地解析。尊重版权明确你爬取的图片的版权归属。个人学习、研究使用通常问题不大但未经授权用于商业用途、公开传播或训练AI模型可能构成侵权。不要造成伤害严格控制请求频率避免对目标网站的正常运营造成影响DDos攻击效果。你的爬虫应该像一个有礼貌的访客而不是一群横冲直撞的野牛。查看服务条款很多网站的用户协议中明确禁止自动化数据抓取。6. 实战案例构建一个简单的异步图片爬虫让我们把上面的知识点串起来写一个针对静态图片画廊的完整示例。import asyncio import aiohttp from bs4 import BeautifulSoup from urllib.parse import urljoin, urlparse import hashlib import os from pathlib import Path import random import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class SimpleAsyncImageCrawler: def __init__(self, start_url, concurrency5, download_dir./downloads): self.start_url start_url self.seen_urls set() # 已发现的URL集合用于去重 self.download_dir Path(download_dir) self.download_dir.mkdir(parentsTrue, exist_okTrue) self.semaphore asyncio.Semaphore(concurrency) # 控制并发数 self.client_timeout aiohttp.ClientTimeout(total60) async def fetch_html(self, session, url): 获取页面HTML async with self.semaphore: # 控制并发 try: headers {User-Agent: Mozilla/5.0...} # 应使用动态UA async with session.get(url, headersheaders, timeoutself.client_timeout) as resp: resp.raise_for_status() # 注意编码有些网站不是utf-8 html await resp.text(encodingutf-8, errorsignore) await asyncio.sleep(random.uniform(1, 2)) # 礼貌延迟 return html except Exception as e: logger.error(fFailed to fetch {url}: {e}) return None def extract_image_urls(self, html, base_url): 从HTML中提取图片URL if not html: return [] soup BeautifulSoup(html, lxml) img_urls [] for img in soup.find_all(img): # 优先考虑懒加载属性 raw_url img.get(data-src) or img.get(data-original) or img.get(src) if raw_url: # 清洗和补全URL absolute_url urljoin(base_url, raw_url) # 简单的过滤比如只抓取jpg, png, webp if any(absolute_url.lower().endswith(ext) for ext in [.jpg, .jpeg, .png, .webp, .gif]): img_urls.append(absolute_url) return list(set(img_urls)) # 简单去重 async def download_image(self, session, img_url): 下载单张图片 file_hash hashlib.md5(img_url.encode()).hexdigest()[:12] ext os.path.splitext(urlparse(img_url).path)[1] or .jpg filename f{file_hash}{ext} filepath self.download_dir / filename if filepath.exists(): logger.info(fSkip existing: {filename}) return try: async with session.get(img_url, timeoutself.client_timeout) as resp: if resp.status 200: image_data await resp.read() # 简单验证是否为图片 if image_data[:3] b\xff\xd8\xff or image_data[:8] b\x89PNG\r\n\x1a\n: filepath.write_bytes(image_data) logger.info(fDownloaded: {filename} from {img_url}) else: logger.warning(fInvalid image data from {img_url}) else: logger.warning(fFailed to download {img_url}, status: {resp.status}) except Exception as e: logger.error(fError downloading {img_url}: {e}) async def crawl_page(self, session, url): 爬取单个页面提取图片并发现新链接简单示例未深入实现链接发现 logger.info(fCrawling: {url}) html await self.fetch_html(session, url) if not html: return [] # 提取本页图片 image_urls self.extract_image_urls(html, url) download_tasks [self.download_image(session, img_url) for img_url in image_urls] if download_tasks: await asyncio.gather(*download_tasks, return_exceptionsTrue) # 这里可以添加发现并递归爬取新页面的逻辑 # 例如提取所有a href过滤出同域名下的链接加入任务队列 # 注意需要深度控制和去重避免无限循环 new_page_urls [] # ... (链接发现逻辑) return new_page_urls async def run(self): 主运行函数 connector aiohttp.TCPConnector(limit100, sslFalse) # 调整连接器参数 async with aiohttp.ClientSession(connectorconnector) as session: # 初始任务队列 to_crawl [self.start_url] self.seen_urls.add(self.start_url) while to_crawl: current_url to_crawl.pop(0) new_urls await self.crawl_page(session, current_url) # 将新发现的、未爬取过的页面加入队列 for new_url in new_urls: if new_url not in self.seen_urls: self.seen_urls.add(new_url) to_crawl.append(new_url) # 控制整体爬取速度 await asyncio.sleep(random.uniform(2, 4)) logger.info(Crawling finished.) if __name__ __main__: # 使用示例请替换为你要爬取的画廊起始页 crawler SimpleAsyncImageCrawler(start_urlhttps://example.com/gallery, concurrency3) asyncio.run(crawler.run())这个案例集成了异步请求、链接提取、图片下载、基础去重和礼貌延迟。它只是一个起点你可以根据实际网站结构丰富其链接发现逻辑、错误处理、代理集成等功能。爬虫是一个不断学习和适应的过程。网站技术在变反爬策略在升级你的爬虫也需要持续迭代。保持耐心仔细分析尊重规则你就能高效、稳定地获取到所需的网络资源。记住最好的爬虫是那个能长期稳定运行、不被发现、也不打扰别人的爬虫。