简介这是一套面向高校师生与企业研发人员的拼多多电商数据采集实践系统聚焦商品信息与用户评论的自动化爬取适用于计算机、人工智能、信息工程等专业的课程设计、毕业课题及科研数据获取场景要求学习者具备Python基础与Selenium操作经验。资源包共31个文件含17个核心Python脚本涵盖Selenium驱动控制、验证码识别、代理管理、数据库操作及多线程爬虫调度等模块、3个配置与说明类txt/md文件、5个备份文件.zbak及2个状态文件.state整体仅44KB轻量紧凑且结构清晰便于快速部署与二次开发。已有93人下载学习配套文档详实源码经教师指导与多轮测试验证答辩评分达95分可直接用于教学演示、学术研究或工程原型搭建并支持基于现有架构扩展SKU分析、情感挖掘等进阶功能。1. 项目概述从需求到实现的爬虫系统构建最近在做一个数据分析项目需要大量拼多多的商品和评论数据。市面上现成的数据服务要么太贵要么数据维度不全于是决定自己动手写一个爬虫系统。这个“基于Python的拼多多商品与评论数据爬取系统”就是这次实践的产物它不是一个简单的脚本而是一个考虑了稳定性、可扩展性和数据质量的完整解决方案。如果你也遇到过需要批量获取电商平台数据但苦于没有稳定、高效的抓取工具或者写出的爬虫动不动就被封IP、数据格式乱七八糟那这套系统或许能给你提供一个清晰的思路和可直接复用的代码框架。这个系统核心解决了几个实际问题一是如何绕过或处理拼多多这类大型电商平台的反爬机制比如滑块验证、请求频率限制二是如何高效、结构化地存储商品列表、商品详情和海量评论数据三是如何设计一个健壮的架构使得爬虫可以7x24小时运行即使遇到异常也能自动恢复。我将会把完整的源码和详细文档分享出来你可以直接部署使用也可以根据你的业务需求进行二次开发。无论是用于市场竞品分析、价格监控、用户评论情感分析还是学术研究这套系统都能提供一个可靠的数据来源。2. 系统核心设计与架构解析2.1 技术栈选型与考量构建一个生产级的爬虫系统技术选型是第一步这直接决定了后续开发的效率和系统的上限。我的核心选择是Python原因很简单生态丰富。requests和aiohttp用于网络请求BeautifulSoup4和parsel或lxml用于HTML解析redis用作请求队列和去重缓存MySQL或PostgreSQL作为结构化数据的主存储MongoDB可选用于存储非结构化的原始页面或复杂JSON数据。为什么不用Scrapy框架Scrapy当然强大但对于拼多多这种接口化程度高、反爬策略复杂的平台有时需要更灵活的请求构造和响应处理逻辑用aiohttp配合asyncio来自主控制并发和协程反而更得心应手。当然系统中也借鉴了Scrapy的“引擎-调度器-下载器-爬虫”的思想。注意直接使用requests进行高频同步请求极易被识别和封禁。本系统的核心是异步请求与智能调度。我们使用aiohttp构建异步HTTP客户端配合连接池和会话复用能极大提升IO效率。同时将待抓取的URL或API参数放入redis队列由调度器控制抓取速率模拟人类操作间隔这是对抗反爬的基础。2.2 系统架构分层设计我把整个系统分为五层这样逻辑清晰也便于维护和扩展调度层这是系统的大脑。它从redis的任务队列中取出任务分配给下载器并控制全局的抓取频率和并发度。这里实现了优先级队列比如商品列表页的优先级高于详情页详情页高于评论页确保核心数据优先获取。下载层基于aiohttp的异步下载器。它接收调度层发来的任务构造HTTP请求包括处理Headers、Cookies、代理等并发起请求。这里集成了代理IP池的自动切换模块和用户代理轮换模块是突破反爬封锁的关键。解析层负责处理下载器返回的响应。对于拼多多大量数据是通过内部API返回的JSON格式。解析层需要从JSON中提取出目标字段并清洗、格式化。对于少量仍需从HTML中提取的数据使用parsel兼容XPath和CSS选择器进行解析。解析后的结构化数据传递给存储层。存储层采用混合存储策略。结构化的商品核心信息商品ID、标题、价格、销量等存入MySQL便于复杂的SQL查询和分析。评论数据量巨大且半结构化存入MySQL的同时也可以将原始JSON评论快照存入MongoDB做备份。redis除了做任务队列还用于存储已抓取URL的指纹布隆过滤器实现实现分布式去重。监控与异常处理层一个常被忽略但至关重要的部分。系统会记录每次请求的响应状态码、耗时监控代理IP的可用率以及解析失败率。当连续出现验证码或滑块验证时系统能自动触发告警如发送邮件或钉钉消息并可能降速或暂停抓取。2.3 反爬策略应对方案拼多多的反爬手段在不断升级我们的系统需要动态应对请求签名与参数加密这是最大的难点。通过浏览器开发者工具的网络面板仔细分析商品列表、详情、评论接口的请求会发现关键的anti_content、page_id等参数是加密的。我们的策略不是去逆向其复杂的JS加密逻辑成本高且易失效而是直接复用浏览器环境。使用pyppeteer或playwright这类无头浏览器工具在关键环节如获取初始Cookies或触发加密参数生成模拟真实浏览器行为获取到有效的加密参数后再交由高效的aiohttp去批量请求API。这是一种“混合动力”模式。滑块验证当触发滑块时纯请求库方案基本失效。我们的方案是集成第三方打码平台如超级鹰、图鉴的API。当下载器收到包含滑块验证的响应时自动截取验证图片调用打码平台识别坐标然后通过无头浏览器模拟滑动操作。这个过程全自动化但会增加耗时因此在调度策略上遇到滑块的请求会被暂时挂起不影响其他任务的进行。频率限制与IP封禁通过redis精确控制每个目标域名下的请求间隔并设置随机延迟如random.uniform(1, 3)秒模拟人工操作。代理IP池是必备的我们需要维护一个高质量、高匿名的HTTP/HTTPS代理IP列表并在下载器中实现自动切换。系统会持续测试代理IP的延迟和可用性剔除失效的IP。3. 核心模块实现与关键代码剖析3.1 异步下载器与代理IP池集成下载器的健壮性直接决定爬虫的生死。下面是一个简化的异步下载器核心类它集成了代理、重试和基础异常处理。import aiohttp import asyncio import random from typing import Optional, Dict, Any import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class AsyncDownloader: def __init__(self, proxy_pool: Optional[list] None, max_retries: int 3): 初始化异步下载器 :param proxy_pool: 代理IP池列表格式如 [http://user:passhost:port, ...] :param max_retries: 最大重试次数 self.proxy_pool proxy_pool or [] self.max_retries max_retries # 使用TCPConnector复用连接设置SSL验证和连接限制 self.connector aiohttp.TCPConnector(sslFalse, limit100, limit_per_host20) self.session: Optional[aiohttp.ClientSession] None async def __aenter__(self): self.session aiohttp.ClientSession(connectorself.connector) return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() async def fetch(self, url: str, method: str GET, headers: Optional[Dict] None, params: Optional[Dict] None, data: Optional[Any] None, **kwargs) - Optional[str]: 执行异步HTTP请求 headers headers or {} # 设置一个常见的浏览器User-Agent头 headers.setdefault(User-Agent, Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36) headers.setdefault(Accept, application/json, text/plain, */*) proxy None if self.proxy_pool: proxy random.choice(self.proxy_pool) logger.debug(fUsing proxy: {proxy}) for attempt in range(self.max_retries): try: async with self.session.request(method, url, headersheaders, paramsparams, datadata, proxyproxy, timeoutaiohttp.ClientTimeout(total30), **kwargs) as response: response.raise_for_status() # 如果状态码不是200抛出异常 content_type response.headers.get(Content-Type, ) if application/json in content_type: return await response.json() else: return await response.text() except aiohttp.ClientError as e: logger.warning(fAttempt {attempt 1} failed for {url}: {e}) if attempt self.max_retries - 1: logger.error(fAll {self.max_retries} attempts failed for {url}) return None await asyncio.sleep(2 ** attempt) # 指数退避重试 except asyncio.TimeoutError: logger.warning(fTimeout on attempt {attempt 1} for {url}) if attempt self.max_retries - 1: return None return None # 使用示例 async def main(): proxy_list [http://proxy1.example.com:8080, http://proxy2.example.com:8080] # 请替换为真实代理 async with AsyncDownloader(proxy_poolproxy_list) as downloader: html await downloader.fetch(https://yangkeduo.com/search_result.html?search_key手机) if html: print(抓取成功获取到页面内容) # 这里可以调用解析函数处理html if __name__ __main__: asyncio.run(main())这段代码构建了一个支持代理、自动重试和连接复用的异步下载器。TCPConnector的limit参数控制全局最大连接数limit_per_host控制对单个域名的并发连接数这是避免对目标服务器造成过大压力的重要设置。3.2 数据解析与字段清洗拼多多的商品详情和评论数据主要通过API返回JSON。解析的关键在于找到正确的接口和理清JSON结构。例如商品详情接口返回的数据包非常庞大我们需要从中精准提取所需字段。import json import re from datetime import datetime from typing import Dict, Any, Optional def parse_goods_detail(api_response: Dict[str, Any]) - Optional[Dict[str, Any]]: 解析商品详情API返回的JSON数据 :param api_response: 接口返回的原始字典数据 :return: 清洗后的商品信息字典 try: # 根据实际接口响应结构定位商品信息这里是一个示例路径 goods_info api_response.get(result, {}).get(goods, {}) if not goods_info: logger.error(商品信息字段缺失) return None # 核心字段提取与清洗 goods_id goods_info.get(goods_id) title goods_info.get(goods_name, ).strip() # 价格可能以分为单位需要转换 price_min goods_info.get(min_normal_price, 0) / 100.0 price_max goods_info.get(max_normal_price, 0) / 100.0 sales goods_info.get(sales, 0) # 销量 # 库存注意字段名可能变化 stock goods_info.get(quantity, goods_info.get(stock, 0)) # 商家信息 mall_info goods_info.get(mall, {}) mall_name mall_info.get(mall_name, ) mall_id mall_info.get(mall_id, 0) # 商品规格SKU信息通常是一个列表 sku_list [] for sku in goods_info.get(skus, []): sku_info { sku_id: sku.get(sku_id), spec: .join([s.get(spec_value) for s in sku.get(specs, [])]), price: sku.get(normal_price, 0) / 100.0, stock: sku.get(quantity, 0) } sku_list.append(sku_info) # 构造最终结构化的商品数据 parsed_data { goods_id: goods_id, title: title, price_range: f{price_min:.2f}-{price_max:.2f}, price_min: price_min, price_max: price_max, sales: sales, stock: stock, mall_name: mall_name, mall_id: mall_id, sku_info: json.dumps(sku_list, ensure_asciiFalse), # 将列表转为JSON字符串存储 crawl_time: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } return parsed_data except KeyError as e: logger.error(f解析商品详情时缺少关键字段: {e}) return None except Exception as e: logger.error(f解析商品详情时发生未知错误: {e}) return None # 评论数据解析示例 def parse_comment_list(comment_response: Dict[str, Any]) - list: 解析评论列表API数据 comments [] comment_items comment_response.get(result, {}).get(comments, []) for item in comment_items: try: comment { comment_id: item.get(comment_id), goods_id: item.get(goods_id), user_nickname: item.get(user_nickname, ).strip(), user_avatar: item.get(user_avatar, ), comment_text: item.get(comment, ).strip(), comment_images: json.dumps(item.get(images, []), ensure_asciiFalse), star: item.get(star, 5), # 评分1-5星 like_count: item.get(like_count, 0), create_time: item.get(create_time, 0), # 可能是时间戳 spec: item.get(spec, ) } comments.append(comment) except Exception as e: logger.warning(f解析单条评论失败: {e}, 跳过该条评论) continue return comments解析函数中加入了大量的异常处理和日志记录这是因为API结构可能微调字段可能缺失健壮的解析逻辑能保证系统在遇到意外数据时不会崩溃而是记录错误并跳过问题数据。3.3 数据存储与数据库设计数据存储的设计要兼顾查询效率和扩展性。以下是MySQL核心表的设计示例商品表 (goods)字段名类型说明idBIGINT UNSIGNED AUTO_INCREMENT自增主键goods_idVARCHAR(50) NOT NULL UNIQUE平台商品ID唯一索引titleVARCHAR(500)商品标题price_minDECIMAL(10,2)最低价price_maxDECIMAL(10,2)最高价salesINT UNSIGNED DEFAULT 0销量stockINT UNSIGNED DEFAULT 0库存mall_nameVARCHAR(100)店铺名mall_idVARCHAR(50)店铺IDsku_infoJSON/TEXTSKU规格信息JSON格式categoryVARCHAR(100)商品分类crawl_timeDATETIME抓取时间update_timeTIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP更新时间评论表 (comments)字段名类型说明idBIGINT UNSIGNED AUTO_INCREMENT自增主键comment_idVARCHAR(50) NOT NULL UNIQUE评论ID唯一索引goods_idVARCHAR(50) NOT NULL关联的商品ID外键user_nicknameVARCHAR(100)用户昵称comment_textTEXT评论内容comment_imagesJSON/TEXT评论图片JSON数组starTINYINT UNSIGNED评分1-5like_countINT UNSIGNED DEFAULT 0点赞数specVARCHAR(200)购买规格create_timeDATETIME评论创建时间crawl_timeDATETIME抓取时间使用goods_id和comment_id作为唯一索引可以防止数据重复插入。评论表通过goods_id与商品表关联便于进行关联查询例如“查询某个商品的所有好评”。sku_info和comment_images字段使用JSON类型MySQL 5.7支持便于存储半结构化数据也方便在应用层直接解析。存储操作的代码需要处理重复插入的问题我们使用ON DUPLICATE KEY UPDATE语句import aiomysql import asyncio async def save_goods_to_mysql(parsed_data: Dict[str, Any], pool): 异步存储商品数据到MySQL使用连接池 if not parsed_data: return False sql INSERT INTO goods (goods_id, title, price_min, price_max, sales, stock, mall_name, mall_id, sku_info, crawl_time) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE title VALUES(title), price_min VALUES(price_min), price_max VALUES(price_max), sales VALUES(sales), stock VALUES(stock), sku_info VALUES(sku_info), update_time CURRENT_TIMESTAMP values ( parsed_data[goods_id], parsed_data[title], parsed_data[price_min], parsed_data[price_max], parsed_data[sales], parsed_data[stock], parsed_data[mall_name], parsed_data[mall_id], parsed_data[sku_info], parsed_data[crawl_time] ) async with pool.acquire() as conn: async with conn.cursor() as cursor: try: await cursor.execute(sql, values) await conn.commit() logger.info(f商品 {parsed_data[goods_id]} 数据保存/更新成功) return True except aiomysql.Error as e: logger.error(f保存商品 {parsed_data[goods_id]} 到MySQL失败: {e}) await conn.rollback() return False4. 完整工作流程与调度策略4.1 任务生成与种子URL管理爬虫的启动始于种子。对于拼多多种子可以是搜索关键词如“手机”、“连衣裙”也可以是分类ID或活动页面URL。系统初始化时将这些种子转化为初始的API请求参数并推入redis的高优先级队列。例如一个搜索关键词“手机”会生成拼多多搜索接口的URL附带必要的加密参数可能需要通过无头浏览器预先获取。任务队列我们使用redis的List或Sorted Set结构。List实现简单的FIFO队列Sorted Set可以实现带优先级的队列。每个任务是一个JSON字符串包含任务类型如search_list,goods_detail,goods_comments、目标URL或API参数、优先级、重试次数等信息。import json import redis.asyncio as redis class TaskScheduler: def __init__(self, redis_urlredis://localhost:6379): self.redis_client redis.from_url(redis_url) self.task_queue_key pdd:crawl:tasks # 使用Sorted Set self.dupefilter_key pdd:crawl:dupefilter # 去重集合 async def add_task(self, task: Dict, priority: int 10): 添加任务到优先级队列 :param task: 任务字典 :param priority: 优先级数字越小优先级越高 task_str json.dumps(task) # 先进行去重判断这里用任务ID如URL的MD5作为去重依据 task_id self._generate_task_id(task) added await self.redis_client.sadd(self.dupefilter_key, task_id) if added: # 如果成功添加到集合说明是新任务 # 使用当前时间戳作为scorepriority影响排序 score priority * 1000000 time.time() # 简单实现优先级时间排序 await self.redis_client.zadd(self.task_queue_key, {task_str: score}) logger.debug(f新任务已加入队列: {task.get(type)}) else: logger.debug(f任务已存在跳过: {task_id}) def _generate_task_id(self, task: Dict) - str: 生成任务唯一标识通常使用URL或关键参数的MD5 import hashlib unique_str task.get(url, ) or json.dumps(task.get(params, {})) return hashlib.md5(unique_str.encode()).hexdigest() async def get_task(self) - Optional[Dict]: 从队列中获取最高优先级的任务 # 使用ZPOPMIN获取score最小的任务优先级最高 result await self.redis_client.zpopmin(self.task_queue_key, count1) if result: task_str, score result[0] return json.loads(task_str) return None4.2 异步主循环与并发控制系统的核心是一个异步事件循环它持续从调度器获取任务交给下载器执行然后解析、存储并可能生成新的子任务如从商品列表页解析出商品详情页任务。import asyncio import signal class CrawlerEngine: def __init__(self, downloader, parser, storage, scheduler, max_concurrent10): self.downloader downloader self.parser parser self.storage storage self.scheduler scheduler self.max_concurrent max_concurrent self.semaphore asyncio.Semaphore(max_concurrent) self.is_running True async def process_task(self, task: Dict): 处理单个任务 async with self.semaphore: # 控制并发数 task_type task.get(type) url task.get(url) params task.get(params) logger.info(f开始处理任务: {task_type}, URL: {url}) # 1. 下载 response_data await self.downloader.fetch(url, paramsparams) if not response_data: logger.warning(f任务下载失败: {task_type}) await self._handle_failed_task(task) # 失败任务处理 return # 2. 解析 parsed_data, new_tasks await self.parser.parse(task_type, response_data, task) if not parsed_data: logger.warning(f任务解析失败: {task_type}) return # 3. 存储 if parsed_data: await self.storage.save(task_type, parsed_data) # 4. 生成新任务如从列表页解析出详情页链接 if new_tasks: for new_task in new_tasks: await self.scheduler.add_task(new_task) # 5. 礼貌延迟避免请求过快 await asyncio.sleep(random.uniform(1.0, 2.5)) async def _handle_failed_task(self, task: Dict): 失败任务重试逻辑 retry_count task.get(retry, 0) if retry_count 3: task[retry] retry_count 1 task[priority] task.get(priority, 10) 5 # 降低优先级稍后重试 await self.scheduler.add_task(task) logger.info(f任务将重试 ({retry_count 1}/3): {task.get(type)}) else: logger.error(f任务重试次数用尽放弃: {task.get(type)}) async def run(self): 引擎主循环 logger.info(爬虫引擎启动...) while self.is_running: task await self.scheduler.get_task() if task: # 为每个任务创建异步任务但不等待实现并发 asyncio.create_task(self.process_task(task)) else: # 队列为空等待一段时间再检查 logger.info(任务队列为空等待5秒...) await asyncio.sleep(5) def stop(self): 优雅停止 self.is_running False logger.info(爬虫引擎正在停止...) # 信号处理实现优雅退出 def signal_handler(): loop asyncio.get_event_loop() crawler.stop() loop.stop() async def main(): # 初始化各个组件 downloader AsyncDownloader(proxy_poolyour_proxy_list) parser DataParser() storage StorageManager() scheduler TaskScheduler() engine CrawlerEngine(downloader, parser, storage, scheduler, max_concurrent5) # 添加初始种子任务 initial_tasks generate_initial_tasks() # 你的函数生成搜索任务等 for task in initial_tasks: await scheduler.add_task(task, priority1) # 种子任务高优先级 # 运行引擎 await engine.run() if __name__ __main__: loop asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, signal_handler) try: loop.run_until_complete(main()) finally: loop.close()这个主循环实现了生产者-消费者模式。调度器是生产者不断提供任务CrawlerEngine是消费者并发地处理任务。asyncio.Semaphore用于控制最大并发数避免同时发起过多请求。失败的任务会根据重试次数重新入队但优先级会降低避免问题任务阻塞队列。4.3 增量爬取与去重策略对于商品数据我们通常需要定期更新价格和销量。对于评论数据则需要增量抓取新的评论。我们的策略是商品列表/搜索页定期全量抓取以发现新上架商品。去重依靠goods_id。商品详情页根据业务需求设置更新频率如每小时一次。通过ON DUPLICATE KEY UPDATE语句更新变化字段价格、销量、库存。评论页增量抓取。记录每个商品最后抓取的评论ID或时间戳。下次抓取时请求该时间点之后的评论。拼多多评论接口通常支持after_id或offset参数。去重不仅在任务调度层面redis集合也在数据存储层面数据库唯一索引双重保障避免数据冗余。5. 部署、监控与常见问题排查5.1 系统部署与环境配置为了让爬虫稳定运行在服务器上建议使用Docker容器化部署。下面是一个简化的Dockerfile和docker-compose.yml示例用于一键部署爬虫及其依赖的Redis和MySQL。DockerfileFROM python:3.9-slim WORKDIR /app # 安装系统依赖如Chromium用于无头浏览器 RUN apt-get update apt-get install -y \ wget \ gnupg \ unzip \ wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - \ echo deb [archamd64] http://dl.google.com/linux/chrome/deb/ stable main /etc/apt/sources.list.d/google.list \ apt-get update apt-get install -y google-chrome-stable \ rm -rf /var/lib/apt/lists/* # 复制依赖文件并安装Python包 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 复制应用代码 COPY . . # 启动命令 CMD [python, main.py]docker-compose.ymlversion: 3.8 services: redis: image: redis:7-alpine container_name: pdd-crawler-redis ports: - 6379:6379 volumes: - redis_data:/data command: redis-server --appendonly yes mysql: image: mysql:8.0 container_name: pdd-crawler-mysql environment: MYSQL_ROOT_PASSWORD: your_strong_password MYSQL_DATABASE: pdd_data MYSQL_USER: crawler MYSQL_PASSWORD: crawler_password ports: - 3306:3306 volumes: - mysql_data:/var/lib/mysql - ./init.sql:/docker-entrypoint-initdb.d/init.sql # 初始化表结构 crawler: build: . container_name: pdd-crawler depends_on: - redis - mysql environment: - REDIS_URLredis://redis:6379/0 - MYSQL_HOSTmysql - MYSQL_USERcrawler - MYSQL_PASSWORDcrawler_password - MYSQL_DATABASEpdd_data volumes: - ./logs:/app/logs # 挂载日志目录 - ./data:/app/data # 挂载数据目录可选 restart: unless-stopped # 异常退出时自动重启 # 可以设置多个crawler实例实现分布式 # deploy: # replicas: 3 volumes: redis_data: mysql_data:使用docker-compose up -d即可启动全套服务。restart: unless-stopped保证了爬虫进程的持续运行。5.2 日志、监控与告警没有监控的爬虫就像在黑夜中航行。我们使用Python标准库logging模块将日志分级输出到文件和控制台并配置日志轮转避免日志文件过大。import logging from logging.handlers import RotatingFileHandler import sys def setup_logger(name, log_filecrawler.log, levellogging.INFO): 设置日志记录器 logger logging.getLogger(name) logger.setLevel(level) # 格式 formatter logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) # 控制台处理器 console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件处理器轮转每个文件10MB最多保留5个 file_handler RotatingFileHandler(log_file, maxBytes10*1024*1024, backupCount5) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger # 在程序入口处调用 logger setup_logger(pdd_crawler)监控指标可以包括请求成功率成功响应数 / 总请求数。平均响应时间判断代理IP或目标网站速度。代理IP池健康度可用代理数 / 总代理数。数据抓取速率单位时间内抓取的商品/评论数。队列深度redis中待处理任务数。可以定期将这些指标打印到日志或推送到Prometheus、Grafana等监控系统。当请求成功率持续低于阈值如90%或触发了滑块验证可以通过smtplib或第三方API如钉钉、企业微信机器人发送告警消息。5.3 常见问题与排查技巧实录在实际运行中你会遇到各种各样的问题。这里记录几个最典型的问题1突然大量返回403 Forbidden或429 Too Many Requests。排查首先检查请求头特别是User-Agent,Referer,Cookie是否完整且有效。检查代理IP是否大量失效。查看当前请求频率是否过高。解决立即暂停爬虫。更换一批新的高质量代理IP。增加请求之间的随机延迟。检查并更新Cookie可能需要重新用无头浏览器登录或获取。如果使用了固定IP可能需要暂停几小时甚至一天。问题2解析函数突然报KeyError找不到某个字段。排查打印出原始的API响应JSON查看结构是否发生变化。可能是拼多多接口更新了。解决更新解析函数使用更健壮的.get()方法并提供默认值。如果是接口路径或参数变化需要重新分析网络请求更新任务生成逻辑。问题3数据库连接数过多或插入速度变慢。排查检查MySQL的SHOW PROCESSLIST;。可能是没有正确使用连接池或者插入语句没有批量操作。解决确保使用aiomysql的连接池。对于评论这类海量数据使用INSERT INTO ... VALUES (...), (...), ...进行批量插入或者使用LOAD DATA INFILE从文件导入能极大提升性能。问题4爬虫运行一段时间后内存占用越来越高。排查可能是异步任务中产生了对象引用循环或者缓存的数据没有及时释放。解决定期检查并清理全局缓存字典。使用asyncio.sleep(0)偶尔让出控制权。对于大规模数据考虑使用磁盘缓存或数据库作为中间存储而非全部放在内存。一个宝贵的实操心得不要试图一次性追求100%的数据抓取率。电商平台的反爬是动态的我们的策略也应该是动态的。设定一个合理的目标如95%的成功率当遇到难以攻克的反爬如复杂的图形验证码时记录下失败的任务跳过它继续抓取其他数据。保证系统整体能持续运行比纠结于某一个难点更重要。可以在系统中设计一个“疑难任务队列”专门存放这些失败的任务后续可以人工介入分析或采用更高级的策略如更真实的浏览器模拟来处理。这套系统经过多次迭代和实战检验已经能够稳定地获取拼多多的商品与评论数据。它不仅仅是一堆代码更是一套应对复杂网络爬取问题的工程化思路。你可以根据附带的完整源码和文档快速搭建起属于自己的数据采集平台并根据实际需求进行调整和优化。记住爬虫技术是工具合理合法地使用它才能为你的项目创造价值。本文还有配套的精品资源点击获取
