基于 mitmproxy 的大众点评数据采集实战：自动化抓取景点与评论数据

基于 mitmproxy 的大众点评数据采集实战：自动化抓取景点与评论数据

📚 前言

在数据分析和市场研究中，爬取大众点评等平台的数据是常见需求。然而，传统的爬虫方式在面对小程序、APP 等场景时往往力不从心。本文将介绍一种基于 mitmproxy 的数据采集方案，通过 HTTP 代理拦截的方式，实现对大众点评小程序数据的自动化采集。

🎯 技术方案概述

为什么选择 mitmproxy？

核心架构

PC微信(大众点评小程序)
↓ HTTP请求
mitmproxy代理服务 (拦截/解析)
↓ 处理后的数据
MySQL数据库 / CSV文件

🛠️ 环境搭建

1. 安装 Python 环境

项目基于 Python 3.12，推荐使用 uv 进行依赖管理：

# 安装 uv (可选)
pip install uv

# 安装项目依赖
uv pip install -r requirements.txt
# 或
pip install -r requirements.txt

2. 核心依赖说明

mitmproxy==11.0.2 # HTTP代理工具
pyautogui==0.9.54 # GUI自动化（模拟滚动）
pymysql==1.1.1 # MySQL数据库操作
python–dotenv==1.1.1 # 环境变量管理

3. 数据库配置（可选）

创建 .env 文件配置数据库：

DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=your_password
DB_NAME=dianping_spiders
DB_CHARSET=utf8mb4

💻 核心代码实现

1. 主程序架构 (main.py)

主程序负责整体流程控制，包括代理管理、服务启动和滚动控制。

1.1 代理管理器

class ProxyManager:
"""代理管理器，负责设置和恢复系统代理"""

def __init__(self, proxy_host: str = "127.0.0.1", proxy_port: int = 8080):
self.proxy_host = proxy_host
self.proxy_port = proxy_port
self.system = platform.system().lower()
self.original_settings = {}

def set_proxy(self) –> bool:
"""设置系统代理"""
if self.system == "darwin": # macOS
return self._set_macos_proxy()
elif self.system == "windows": # Windows
return self._set_windows_proxy()

关键点：

• 支持 Windows 和 macOS 双平台
• 自动保存原始代理设置，退出时恢复
• Windows 通过注册表设置，macOS 通过命令行工具设置

1.2 mitmweb 服务管理

class MitmWebManager:
"""mitmweb服务管理器"""

def start(self) –> bool:
"""启动mitmweb服务"""
cmd = [
"mitmweb",
"-s", self.script_path, # 加载拦截脚本
"–listen-port", str(self.port), # 代理端口
"–web-port", str(self.port + 1) # Web界面端口
]
self.process = subprocess.Popen(cmd, ...)

特性：

• 自动检测服务状态
• 异常退出时自动重启（最多3次）
• 日志记录和错误追踪

1.3 滚动控制

def scroll(scroll_count: int = 99999,
scroll_pause: float = 1,
speed: int = –200) –> None:
"""模拟下滑操作"""
for i in range(scroll_count):
pyautogui.scroll(speed) # 向下滚动
time.sleep(scroll_pause) # 等待数据加载

交互控制：

• 空格键：暂停/继续
• ESC 键：退出程序

2. 评论数据拦截 (src/pinglun_list.py)

这是项目的核心数据采集模块，实现评论 API 的拦截和解析。

2.1 响应拦截函数

def response(flow: http.HTTPFlow) –> None:
"""处理HTTP响应，拦截大众点评评论数据"""
# 检查是否是目标URL
if TARGET_URL_PATTERN not in flow.request.url:
return

# 验证响应状态
if flow.response.status_code != 200:
return

# 解析JSON响应
response_json = flow.response.json()

# 获取请求参数
shop_uuid = flow.request.query.get('shopId')
offset = flow.request.query.get('offset')

# 保存原始响应和解析数据
_save_response_file(flow, shop_uuid, offset)
parse_json(response_json, shop_uuid, offset)

2.2 数据解析和存储

def parse_json(response_data: Dict[str, Any],
shop_uuid: str,
offset: str) –> None:
"""解析JSON数据并保存到CSV文件"""
review_list = response_data['reviewInfo']['reviewListInfo']['reviewList']

# 为每个shop_uuid创建单独的CSV文件
csv_file_path = CSV_DIR / f"{shop_uuid}.csv"

with open(csv_file_path, 'a+', newline='', encoding='utf-8-sig') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

if not csv_file_path.exists():
writer.writeheader() # 写入表头

# 写入每条评论
for review in review_list:
writer.writerow({
'reviewId': review.get('reviewId'),
'shopUuid': shop_uuid,
'userId': review.get('userId'),
'text': _extract_review_text(review),
'pics': _extract_review_pics(review),
# … 其他字段
})

数据存储策略：

• CSV 文件：每个店铺一个文件，支持追加模式
• 原始 JSON：保存到 log/dianping_responses/ 用于回溯
• UTF-8 with BOM：确保 Excel 正确打开中文

3. 景点数据拦截 (src/jingdian_list.py)

景点数据的拦截和存储逻辑与评论类似，但保存到 MySQL 数据库。

3.1 数据库插入逻辑

sql = """
INSERT INTO attractions (
shopUuid, categoryName, name, defaultPic, priceText,
recommendReason, regionName, reviewCount, starScore, …
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, …)
ON DUPLICATE KEY UPDATE
categoryName = VALUES(categoryName),
name = VALUES(name),
…
updated_at = CURRENT_TIMESTAMP
"""

关键特性：

• 使用 ON DUPLICATE KEY UPDATE 避免重复数据
• 自动更新时间戳
• 参数化查询防止 SQL 注入

4. 数据库封装 (src/mysql.py)

采用单例模式管理数据库连接：

class MySQLDatabase:
_instance = None

def __new__(cls):
if cls._instance is None:
cls._instance = super(MySQLDatabase, cls).__new__(cls)
cls._instance.__connect()
return cls._instance

def __connect(self):
"""连接到MySQL数据库"""
db_config = DatabaseConfig.get_mysql_config()
self.connection = pymysql.connect(**db_config)
self.cursor = self.connection.cursor(cursor=cursors.DictCursor)

🚀 使用流程

步骤 1：启动程序

python main.py

程序会自动：

步骤 2：打开目标页面

在 PC 微信中打开大众点评小程序，进入：

• 景点列表页：采集景点信息
• 评论列表页：采集评论数据

步骤 3：自动采集

程序开始自动滚动页面，触发数据加载：

• 每次滚动后等待 3-5 秒（随机）
• mitmproxy 自动拦截 API 请求
• 数据自动保存到本地

步骤 4：查看数据

• 评论数据：log/dianping_csv/{shop_uuid}.csv
• 景点数据：MySQL 数据库 attractions 表
• 原始响应：log/dianping_responses/{shop_uuid}/

🔍 技术要点详解

1. mitmproxy 脚本机制

mitmproxy 提供了多个钩子函数，本项目主要使用：

• request(flow): 请求拦截（可用于修改请求）
• response(flow): 响应拦截（用于提取数据）

from mitmproxy import http

def response(flow: http.HTTPFlow) –> None:
# flow.request: 请求对象
# flow.response: 响应对象
response_json = flow.response.json()
# 处理数据…

2. 代理证书安装

首次使用 mitmproxy 需要安装证书：

3. 滚动参数优化

滚动参数影响采集效率和稳定性：

# 快速采集（可能漏数据）
scroll_pause = 1, speed = –2000

# 稳定采集（推荐）
scroll_pause = random.randint(3, 5), speed = random.randint(–2000, –1000)

# 慢速采集（最稳定）
scroll_pause = 5, speed = –500

4. 错误处理和重试

# mitmweb 服务监控
def _monitor_process(self) –> None:
while self.should_monitor:
if self.process.poll() is not None:
# 服务退出，自动重启
if self.restart_count < self.max_restarts:
self._restart_service()

📊 数据采集结果

评论数据示例（CSV）

景点数据示例（数据库）

⚠️ 注意事项和最佳实践

1. 合法使用

• 本工具仅供学习研究使用
• 遵守网站服务条款和相关法律法规
• 控制采集频率，避免对服务器造成压力

2. 数据采集建议

• 分批采集：不要一次性采集大量数据
• 随机延迟：使用随机延迟模拟人类行为
• 错误重试：实现合理的重试机制
• 数据去重：避免重复采集相同数据

3. 性能优化

• 使用数据库连接池
• 批量插入数据而非逐条插入
• 异步处理非关键路径

4. 故障排查

问题 1：无法拦截数据

• 检查代理是否设置成功
• 确认 mitmproxy 证书已安装
• 验证目标 URL 模式是否正确

问题 2：数据解析错误

• 检查 JSON 结构是否变化
• 查看原始响应文件进行调试
• 添加异常处理和日志

🎓 扩展应用

1. 扩展到其他平台

相同的技术方案可以应用于：

• 美团、饿了么等本地生活平台
• 淘宝、京东等电商平台
• 抖音、快手等短视频平台

2. 数据处理和分析

采集到的数据可以用于：

• 情感分析：分析评论的情感倾向
• 用户画像：基于评论行为分析用户特征
• 竞品分析：对比不同商家的评分和评论
• 市场调研：分析用户偏好和消费趋势

3. 实时监控

可以扩展为实时监控系统：

• 监控特定店铺的新评论
• 价格变动提醒
• 评分变化追踪

📚 总结

本文介绍了基于 mitmproxy 的数据采集方案，通过以下几个关键点实现了高效的数据采集：

这种方案的优势在于：

• ✅ 绕过传统爬虫的限制
• ✅ 获取真实 API 数据，结构清晰
• ✅ 自动化程度高，减少人工干预
• ✅ 易于扩展和维护

希望本文对需要采集小程序/APP 数据的开发者有所帮助。如有问题，欢迎交流讨论！

技术交流：欢迎在评论区讨论相关技术问题

免责声明：本工具仅用于技术学习和研究，请遵守相关法律法规。