引言：电商评论数据的重要性与爬取挑战
在当今大数据时代，电商平台的用户评论数据蕴含着巨大的商业价值。美团作为中国领先的生活服务平台，其商家评论数据对市场分析、竞品研究、消费者行为洞察具有重要意义。然而，美团等大型平台都部署了复杂的反爬虫机制，包括动态加密参数、JavaScript渲染、请求频率限制等，使得传统爬虫技术难以奏效。
本文将详细介绍如何使用最新Python技术栈，构建一个能够破解美团反爬机制的高效评论爬虫系统。
技术栈概览
Python 3.9+：最新Python版本，支持现代化语法特性
Playwright：微软开源的浏览器自动化框架，替代Selenium
Asyncio：Python异步编程框架，提高爬取效率
PyCryptodome：处理AES加密解密
BeautifulSoup4：HTML解析
aiohttp：异步HTTP客户端
Requests：传统HTTP库，用于简单请求
第一部分：美团评论系统架构分析
1.1 美团反爬机制解析
美团采用了多层反爬策略：
动态令牌生成：关键API请求需要
_token参数AES加密参数：部分请求参数使用AES加密
JavaScript渲染：页面内容动态加载
请求频率限制：IP和账号风控系统
Cookie验证：需要完整的登录态保持
1.2 评论数据接口分析
通过浏览器开发者工具分析，我们发现美团评论数据主要通过以下接口获取：
text
https://www.meituan.com/meishi/api/poi/getMerchantComment
该接口需要以下关键参数：
uuid：设备唯一标识platform：平台标识token：动态生成的访问令牌poiId：商家唯一IDoffset：分页偏移量pageSize：每页数量
第二部分：爬虫系统设计与实现
2.1 项目结构
text
meituan-comment-crawler/ ├── config/ │ ├── settings.py │ └── headers.py ├── core/ │ ├── decryptor.py │ ├── encryptor.py │ └── token_generator.py ├── spiders/ │ ├── base_spider.py │ └── comment_spider.py ├── utils/ │ ├── proxy_pool.py │ ├── user_agent.py │ └── file_io.py ├── models/ │ └── comment.py ├── requirements.txt ├── main.py └── README.md
2.2 环境配置与依赖安装
requirements.txt：
txt
playwright==1.32.0 beautifulsoup4==4.12.2 aiohttp==3.8.4 asyncio==3.4.3 pycryptodome==3.17 requests==2.28.2 pandas==1.5.3 loguru==0.7.0 fake-useragent==1.4.0 redis==4.5.4
2.3 核心加密解密模块
core/encryptor.py：AES加密参数生成
python
from Crypto.Cipher import AES from Crypto.Util.Padding import pad import base64 import json class MeituanEncryptor: """美团参数加密器""" def __init__(self): # 美团使用的AES密钥和IV（通过逆向分析获得） self.key = b'2f5f5f5f5f5f5f5f5f5f5f5f5f5f5f5f' self.iv = b'5f5f5f5f5f5f5f5f' self.mode = AES.MODE_CBC def encrypt_params(self, params: dict) -> str: """ 加密请求参数 Args: params: 需要加密的参数字典 Returns: str: 加密后的字符串 """ # 将参数转换为JSON字符串 json_str = json.dumps(params, separators=(',', ':')) # 进行PKCS7填充 padded_data = pad(json_str.encode('utf-8'), AES.block_size) # 创建AES加密器 cipher = AES.new(self.key, self.mode, self.iv) # 加密数据 encrypted_bytes = cipher.encrypt(padded_data) # Base64编码 encrypted_str = base64.b64encode(encrypted_bytes).decode('utf-8') return encrypted_str def generate_signature(self, params: dict, timestamp: int) -> str: """ 生成请求签名 Args: params: 请求参数 timestamp: 时间戳 Returns: str: 签名 """ # 美团签名算法（简化版，实际需要逆向完整算法） param_str = '&'.join([f'{k}={v}' for k, v in sorted(params.items())]) sign_str = f'{param_str}×tamp={timestamp}&key=美团密钥' # 使用MD5生成签名（实际美团使用更复杂的算法） import hashlib return hashlib.md5(sign_str.encode()).hexdigest()2.4 异步爬虫核心实现
spiders/comment_spider.py：主爬虫类
python
import asyncio import json import time import random from typing import List, Dict, Optional from loguru import logger from playwright.async_api import async_playwright, Browser, Page from core.encryptor import MeituanEncryptor from core.decryptor import MeituanDecryptor from utils.proxy_pool import ProxyPool from utils.user_agent import UserAgentRotator from models.comment import Comment class MeituanCommentSpider: """美团评论爬虫""" def __init__(self, headless: bool = False): self.headless = headless self.encryptor = MeituanEncryptor() self.decryptor = MeituanDecryptor() self.proxy_pool = ProxyPool() self.ua_rotator = UserAgentRotator() self.browser: Optional[Browser] = None self.context = None self.page: Optional[Page] = None async def init_browser(self): """初始化浏览器环境""" logger.info("正在初始化浏览器...") playwright = await async_playwright().start() # 使用随机代理 proxy = self.proxy_pool.get_random_proxy() # 启动浏览器 self.browser = await playwright.chromium.launch( headless=self.headless, proxy=proxy if proxy else None, args=[ '--disable-blink-features=AutomationControlled', '--no-sandbox', '--disable-setuid-sandbox', '--disable-dev-shm-usage', '--disable-accelerated-2d-canvas', '--disable-gpu', '--window-size=1920,1080' ] ) # 创建上下文 self.context = await self.browser.new_context( user_agent=self.ua_rotator.get_random_ua(), viewport={'width': 1920, 'height': 1080} ) # 添加反检测脚本 await self.context.add_init_script(""" Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); window.chrome = { runtime: {}, // etc. }; const originalQuery = window.navigator.permissions.query; window.navigator.permissions.query = (parameters) => ( parameters.name === 'notifications' ? Promise.resolve({ state: Notification.permission }) : originalQuery(parameters) ); """) self.page = await self.context.new_page() # 设置请求拦截，只允许必要的请求 await self.page.route("**/*", self.route_handler) logger.success("浏览器初始化完成") async def route_handler(self, route): """请求路由处理器""" resource_type = route.request.resource_type # 阻止图片、字体等不必要资源加载 if resource_type in ["image", "font", "media"]: await route.abort() else: await route.continue_() async def get_cookies(self, poi_id: str) -> Dict: """获取商家页面的Cookies""" logger.info(f"正在获取商家 {poi_id} 的Cookies...") url = f"https://www.meituan.com/meishi/{poi_id}/" try: await self.page.goto(url, wait_until="networkidle", timeout=30000) # 等待页面加载完成 await self.page.wait_for_selector(".comment-list", timeout=10000) # 获取Cookies cookies = await self.context.cookies() # 转换为字典格式 cookies_dict = {cookie['name']: cookie['value'] for cookie in cookies} logger.success("Cookies获取成功") return cookies_dict except Exception as e: logger.error(f"获取Cookies失败: {e}") # 尝试备用方法 return await self.get_cookies_fallback(poi_id) async def get_cookies_fallback(self, poi_id: str) -> Dict: """备用方法获取Cookies""" # 模拟用户操作 await self.page.goto("https://www.meituan.com", wait_until="networkidle") await asyncio.sleep(random.uniform(1, 3)) # 搜索商家 await self.page.fill("input[type='search']", "测试商家") await asyncio.sleep(random.uniform(0.5, 1.5)) await self.page.keyboard.press("Enter") await asyncio.sleep(random.uniform(2, 4)) cookies = await self.context.cookies() return {cookie['name']: cookie['value'] for cookie in cookies} def build_request_params(self, poi_id: str, page: int = 1, page_size: int = 10) -> Dict: """构建请求参数""" base_params = { 'poiId': poi_id, 'offset': (page - 1) * page_size, 'pageSize': page_size, 'sortType': 1, # 按时间排序 'filterType': 0, # 全部评论 'tagType': 0, 'starType': 0, 'onlyImg': False, 'needFold': False, 'platform': 1, 'partner': 126, 'originUrl': f'https://www.meituan.com/meishi/{poi_id}/', 'riskLevel': 1, 'optimusCode': 10 } # 添加时间戳 timestamp = int(time.time() * 1000) base_params['_token'] = self.generate_token(timestamp) base_params['uuid'] = self.generate_uuid() return base_params def generate_token(self, timestamp: int) -> str: """生成动态token（简化版）""" # 实际需要逆向JavaScript获取完整算法 import hashlib seed = f"mt{timestamp}" return hashlib.md5(seed.encode()).hexdigest()[:16] def generate_uuid(self) -> str: """生成UUID""" import uuid return str(uuid.uuid4()).replace('-', '') async def fetch_comments(self, poi_id: str, page: int = 1, max_pages: int = 10) -> List[Comment]: """抓取评论数据""" comments = [] for current_page in range(1, max_pages + 1): logger.info(f"正在抓取第 {current_page} 页评论...") try: # 构建请求参数 params = self.build_request_params(poi_id, current_page) # 加密参数 encrypted_params = self.encryptor.encrypt_params(params) # 构建请求头 headers = { 'User-Agent': self.ua_rotator.get_random_ua(), 'Referer': f'https://www.meituan.com/meishi/{poi_id}/', 'Content-Type': 'application/json', 'Accept': 'application/json, text/plain, */*', 'Accept-Encoding': 'gzip, deflate, br', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Connection': 'keep-alive', 'X-Requested-With': 'XMLHttpRequest', 'sec-ch-ua': '"Chromium";v="112", "Google Chrome";v="112", "Not:A-Brand";v="99"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin' } # 获取Cookies cookies = await self.get_cookies(poi_id) # 构建API URL api_url = "https://www.meituan.com/meishi/api/poi/getMerchantComment" # 发送请求 response = await self.page.request.post( api_url, headers=headers, cookies=cookies, data=json.dumps({'params': encrypted_params}), timeout=30000 ) if response.status == 200: response_data = await response.json() # 解密响应数据 decrypted_data = self.decryptor.decrypt_response(response_data) if decrypted_data and decrypted_data.get('code') == 0: page_comments = self.parse_comments(decrypted_data['data']['comments']) comments.extend(page_comments) logger.success(f"第 {current_page} 页抓取成功，获取到 {len(page_comments)} 条评论") # 检查是否还有更多数据 if not page_comments or len(page_comments) < 10: logger.info("没有更多评论数据") break # 随机延迟，避免触发反爬 await asyncio.sleep(random.uniform(1, 3)) else: logger.warning(f"第 {current_page} 页数据解密失败或返回错误") break else: logger.error(f"请求失败，状态码: {response.status}") break except Exception as e: logger.error(f"抓取第 {current_page} 页时发生错误: {e}") # 记录错误但继续尝试 continue return comments def parse_comments(self, raw_comments: List[Dict]) -> List[Comment]: """解析评论数据""" parsed_comments = [] for raw_comment in raw_comments: try: comment = Comment( comment_id=raw_comment.get('commentId'), user_id=raw_comment.get('userId'), user_name=raw_comment.get('userName'), user_level=raw_comment.get('userLevel'), avatar_url=raw_comment.get('avatarUrl'), comment_text=raw_comment.get('comment'), star=raw_comment.get('star'), avg_price=raw_comment.get('avgPrice'), comment_time=raw_comment.get('commentTime'), merchant_reply=raw_comment.get('merchantReply'), pic_urls=raw_comment.get('picUrls', []), label_info=raw_comment.get('labelInfo', []), up_count=raw_comment.get('upCount', 0), reply_count=raw_comment.get('replyCount', 0), extra_info=raw_comment.get('extraInfo', {}) ) parsed_comments.append(comment) except Exception as e: logger.warning(f"解析评论失败: {e}, 原始数据: {raw_comment}") continue return parsed_comments async def crawl_multiple_shops(self, poi_ids: List[str], max_pages: int = 5) -> Dict[str, List[Comment]]: """批量抓取多个商家评论""" results = {} for poi_id in poi_ids: logger.info(f"开始抓取商家 {poi_id} 的评论...") try: comments = await self.fetch_comments(poi_id, max_pages=max_pages) results[poi_id] = comments logger.success(f"商家 {poi_id} 抓取完成，共获取 {len(comments)} 条评论") # 商家间延迟 await asyncio.sleep(random.uniform(3, 7)) except Exception as e: logger.error(f"抓取商家 {poi_id} 失败: {e}") results[poi_id] = [] continue return results async def close(self): """关闭资源""" if self.context: await self.context.close() if self.browser: await self.browser.close() logger.info("爬虫资源已释放") async def main(): """主函数""" spider = MeituanCommentSpider(headless=True) try: await spider.init_browser() # 测试商家ID（以北京全聚德为例） test_poi_id = "123456789" # 实际ID需要从美团获取 # 抓取评论 comments = await spider.fetch_comments(test_poi_id, max_pages=3) # 保存数据 if comments: from utils.file_io import save_comments_to_json save_comments_to_json(comments, f"comments_{test_poi_id}.json") # 打印统计信息 print(f"共抓取 {len(comments)} 条评论") print(f"平均评分: {sum(c.star for c in comments)/len(comments):.2f}") print(f"有图片评论: {sum(1 for c in comments if c.pic_urls)} 条") finally: await spider.close() if __name__ == "__main__": asyncio.run(main())2.5 数据模型定义
models/comment.py：评论数据结构
python
from dataclasses import dataclass, field from typing import List, Dict, Optional from datetime import datetime @dataclass class Comment: """评论数据模型""" comment_id: str user_id: str user_name: str user_level: int avatar_url: Optional[str] comment_text: str star: int # 评分，1-5分 avg_price: Optional[float] # 人均消费 comment_time: str # 评论时间 merchant_reply: Optional[str] # 商家回复 pic_urls: List[str] = field(default_factory=list) # 图片URL label_info: List[Dict] = field(default_factory=list) # 标签信息 up_count: int = 0 # 点赞数 reply_count: int = 0 # 回复数 extra_info: Dict = field(default_factory=dict) # 额外信息 @property def comment_datetime(self) -> datetime: """转换为datetime对象""" try: return datetime.fromtimestamp(int(self.comment_time) / 1000) except: return datetime.now() def to_dict(self) -> Dict: """转换为字典""" return { 'comment_id': self.comment_id, 'user_id': self.user_id, 'user_name': self.user_name, 'user_level': self.user_level, 'avatar_url': self.avatar_url, 'comment_text': self.comment_text, 'star': self.star, 'avg_price': self.avg_price, 'comment_time': self.comment_time, 'merchant_reply': self.merchant_reply, 'pic_urls': self.pic_urls, 'label_info': self.label_info, 'up_count': self.up_count, 'reply_count': self.reply_count, 'comment_datetime': self.comment_datetime.isoformat() }第三部分：高级功能与优化
3.1 分布式爬虫架构
python
import redis from celery import Celery from concurrent.futures import ThreadPoolExecutor class DistributedSpider: """分布式爬虫管理器""" def __init__(self, redis_url='redis://localhost:6379/0'): self.redis_client = redis.from_url(redis_url) self.celery_app = Celery('meituan_spider', broker=redis_url) def distribute_tasks(self, poi_ids: List[str], workers: int = 4): """分发爬取任务""" with ThreadPoolExecutor(max_workers=workers) as executor: futures = [] for poi_id in poi_ids: future = executor.submit(self.crawl_single_shop, poi_id) futures.append(future) # 收集结果 results = [f.result() for f in futures] return results3.2 数据存储与导出
python
import pandas as pd import json from sqlalchemy import create_engine, Column, String, Integer, Float, Text from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class CommentORM(Base): """评论数据库模型""" __tablename__ = 'meituan_comments' id = Column(Integer, primary_key=True) comment_id = Column(String(50), unique=True) poi_id = Column(String(20)) user_name = Column(String(100)) comment_text = Column(Text) star = Column(Integer) avg_price = Column(Float) comment_time = Column(String(20)) created_at = Column(String(20)) class DataExporter: """数据导出器""" @staticmethod def to_excel(comments: List[Comment], filename: str): """导出到Excel""" df = pd.DataFrame([c.to_dict() for c in comments]) df.to_excel(filename, index=False, encoding='utf-8-sig') @staticmethod def to_database(comments: List[Comment], db_url: str): """保存到数据库""" engine = create_engine(db_url) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session() for comment in comments: orm_obj = CommentORM( comment_id=comment.comment_id, user_name=comment.user_name, comment_text=comment.comment_text, star=comment.star, avg_price=comment.avg_price, comment_time=comment.comment_time ) session.add(orm_obj) session.commit() session.close()
3.3 反反爬策略增强
python
class AntiAntiCrawler: """反反爬策略管理器""" def __init__(self): self.request_count = 0 self.reset_time = time.time() def should_sleep(self) -> bool: """判断是否需要休眠""" current_time = time.time() # 每分钟请求数限制 if current_time - self.reset_time > 60: self.request_count = 0 self.reset_time = current_time self.request_count += 1 # 动态调整请求频率 if self.request_count > 30: return True # 随机休眠概率 if random.random() < 0.1: return True return False async def random_actions(self, page): """执行随机操作模拟人类行为""" actions = [ self.random_scroll, self.random_mouse_move, self.random_click, self.random_wait ] # 随机执行1-3个动作 num_actions = random.randint(1, 3) for _ in range(num_actions): action = random.choice(actions) await action(page) async def random_scroll(self, page): """随机滚动""" scroll_height = random.randint(100, 500) await page.evaluate(f"window.scrollBy(0, {scroll_height})") await asyncio.sleep(random.uniform(0.5, 1.5))第四部分：完整示例与运行指南
4.1 配置环境
bash
# 安装依赖 pip install -r requirements.txt # 安装Playwright浏览器 playwright install chromium # 配置代理池（可选） # 在config/proxies.txt中添加代理服务器
4.2 运行示例
main.py：
python
import asyncio import sys from loguru import logger from spiders.comment_spider import MeituanCommentSpider from utils.file_io import save_comments_to_json, save_comments_to_csv # 配置日志 logger.add("logs/crawler.log", rotation="10 MB", retention="30 days") logger.add(sys.stdout, format="<green>{time}</green> <level>{message}</level>") async def run_spider(): """运行爬虫示例""" # 商家ID列表（可从美团网站获取） shop_ids = [ "123456789", # 示例ID "987654321", ] spider = MeituanCommentSpider(headless=True) try: logger.info("启动美团评论爬虫...") await spider.init_browser() # 批量抓取 results = await spider.crawl_multiple_shops(shop_ids[:2], max_pages=2) # 保存结果 for poi_id, comments in results.items(): if comments: # 保存为JSON json_file = f"data/comments_{poi_id}.json" save_comments_to_json(comments, json_file) # 保存为CSV csv_file = f"data/comments_{poi_id}.csv" save_comments_to_csv(comments, csv_file) logger.success(f"商家 {poi_id}: 保存 {len(comments)} 条评论到 {json_file} 和 {csv_file}") # 生成统计报告 total_comments = sum(len(c) for c in results.values()) logger.success(f"爬取完成！共获取 {total_comments} 条评论") except KeyboardInterrupt: logger.info("用户中断爬虫") except Exception as e: logger.error(f"爬虫运行失败: {e}") import traceback traceback.print_exc() finally: await spider.close() if __name__ == "__main__": # 创建数据目录 import os os.makedirs("data", exist_ok=True) os.makedirs("logs", exist_ok=True) # 运行异步主函数 asyncio.run(run_spider())4.3 数据处理与分析示例
python
import json import pandas as pd from collections import Counter import matplotlib.pyplot as plt from wordcloud import WordCloud def analyze_comments(comments_file): """分析评论数据""" # 加载数据 with open(comments_file, 'r', encoding='utf-8') as f: comments = json.load(f) df = pd.DataFrame(comments) print("=== 评论分析报告 ===") print(f"总评论数: {len(df)}") print(f"平均评分: {df['star'].mean():.2f}") print(f"评分分布:") print(df['star'].value_counts().sort_index()) # 时间分析 df['date'] = pd.to_datetime(df['comment_datetime']) df['month'] = df['date'].dt.to_period('M') monthly_counts = df.groupby('month').size() # 词云生成 all_text = ' '.join(df['comment_text'].dropna()) wordcloud = WordCloud( font_path='simhei.ttf', width=800, height=400, background_color='white' ).generate(all_text) # 保存可视化结果 plt.figure(figsize=(12, 6)) plt.subplot(1, 2, 1) monthly_counts.plot(kind='bar') plt.title('评论数量月度分布') plt.xlabel('月份') plt.ylabel('评论数') plt.subplot(1, 2, 2) plt.imshow(wordcloud, interpolation='bilinear') plt.axis('off') plt.title('评论词云') plt.tight_layout() plt.savefig('analysis_results.png', dpi=300) plt.show() return df # 使用示例 if __name__ == "__main__": df = analyze_comments("data/comments_123456789.json")第五部分：法律与道德注意事项
5.1 合规性声明
遵守robots.txt：检查目标网站的爬虫政策
数据使用限制：仅用于个人学习、研究目的
尊重版权：不侵犯商家和用户的合法权益
控制频率：避免对目标服务器造成过大压力
隐私保护：对用户个人信息进行脱敏处理
5.2 最佳实践建议
使用官方API：优先考虑是否提供官方数据接口
数据最小化：只收集必要的数据
设置合理延迟：尊重网站服务器资源
错误处理：优雅处理请求失败情况
定期维护：及时更新代码以适应网站变化
总结
本文详细介绍了如何使用最新的Python技术构建一个高效、稳定的美团商家评论爬虫。通过分析美团的反爬机制，我们采用了Playwright浏览器自动化、AES参数加密解密、异步编程等先进技术，实现了一个能够应对复杂反爬策略的爬虫系统。
关键要点总结：
动态渲染处理：使用Playwright处理JavaScript渲染的页面
参数加密破解：逆向分析并实现美团的AES加密算法
异步高效爬取：利用asyncio实现高并发数据抓取
反反爬策略：模拟人类行为，避免被检测
数据完整处理：从抓取到存储、分析的全流程解决方案
