1. 项目概述与核心价值最近在做一个本地生活服务的市场分析项目需要大量、持续的美团外卖商家和商品数据。手动收集显然不现实数据量太大更新也太快。于是我自然而然地想到了用网络爬虫技术来自动化这个采集过程。这听起来像是一个典型的爬虫应用场景但实际操作起来你会发现从美团外卖这类大型、复杂的商业平台获取数据远不是写几行requests代码那么简单。它涉及反爬策略对抗、数据结构解析、采集策略制定等一系列工程化问题。简单来说这个项目的目标就是设计并实现一个稳定、高效、可维护的自动化程序能够按照我们的需求从美团外卖的网页或接口中抓取指定区域、指定品类的商家列表、商家详情如评分、销量、地址、菜单商品信息以及用户评价等核心数据。这些数据对于竞品分析、商圈热度评估、价格监控、趋势预测等商业决策至关重要。无论你是数据分析师、市场研究员还是对爬虫技术感兴趣的开发者理解这套方法都能帮你打开一扇获取高价值商业数据的大门。2. 整体设计与核心思路拆解在动手写代码之前我们必须先想清楚怎么“挖”数据。直接对美团外卖的主站页面进行暴力爬取几乎是行不通的会立刻触发其强大的反爬机制。因此我们的设计思路需要更巧妙、更贴近真实用户行为。2.1 数据源分析与策略选择首先我们要明确数据在哪里。美团外卖的数据呈现主要通过两个渠道一是用户直接访问的网页端二是手机APP。我们的爬虫可以针对这两者设计。网页端爬取这是最直观的方式。通过浏览器访问美团外卖官网查看页面源代码或利用开发者工具F12监控网络请求。这种方式的好处是易于调试和分析但反爬措施如验证码、行为检测、动态加载也最为严格。通常需要配合Selenium、Playwright这类自动化测试工具来模拟浏览器操作成本较高。APP接口爬取这是目前更主流、更稳定的方式。通过抓包工具如Charles、Fiddler、mitmproxy分析手机APP发出的网络请求找到其与服务器通信的API接口。这些接口返回的数据通常是结构化的JSON非常干净易于解析。难点在于接口参数可能加密、需要携带特定的认证令牌Token并且接口地址和参数格式可能频繁变动。我的策略选择对于美团外卖这种级别的应用优先选择分析APP接口。虽然入门门槛稍高但一旦摸清规律采集效率和稳定性远超模拟浏览器。本次分享也将以APP接口爬虫为核心进行展开。2.2 核心流程设计一个完整的采集流程可以抽象为以下几个步骤地理定位与区域划分美团外卖的数据是基于地理位置的。我们首先需要确定目标城市进而将其划分为更小的区域如按商圈、按行政区、或按自定义的网格。通常我们可以利用美团外卖自身的地理搜索接口通过传递城市名、行政区名或直接的地理坐标经纬度来获取一个address_id或area_id作为后续请求的入参。商家列表获取在确定了具体区域后调用商家列表接口。这里需要处理分页并且可以附加筛选条件如“美食”、“销量最高”、“距离最近”等。接口会返回一批商家的基础信息最重要的是每个商家的唯一IDpoi_id或wm_poi_id。商家详情抓取拿到商家ID列表后逐个请求商家详情接口。这个接口信息最全包括商家名称、地址、联系电话、营业时间、综合评分、口味/包装/配送评分、月售数量、起送价、配送费、公告等。菜单商品信息抓取商家详情中可能只包含部分商品完整的菜单通常有独立的接口。需要传入商家ID获取其所有分类如“热销榜”、“折扣商品”、“主食”以及分类下的具体商品包括商品名、价格、原价、图片、描述、月售等。评价数据抓取用户评价是另一个维度的宝贵数据。通过评价接口可以按时间、评分等级筛选抓取评价内容、评分、用户昵称通常脱敏、推荐菜、追加评价等信息。这里同样需要注意分页。整个流程就像一个漏斗从城市到区域到商家列表再到具体的商家、商品和评价。我们需要设计一个任务调度器来有序地管理这个流程。2.3 技术栈选型与考量工欲善其事必先利其器。以下是经过实战检验的技术栈组合编程语言Python 3.8。几乎是爬虫领域的标准选择生态丰富库多易用。HTTP请求库requests。简单易用足以应对大部分接口请求。需要配合session对象维持会话状态。JSON解析内置的json库。处理接口返回的JSON数据。数据存储轻量级/快速原型SQLite。单文件无需安装服务器非常适合初期测试和小规模数据存储。正式项目/大规模数据MySQL 或 PostgreSQL。关系型数据库便于进行复杂的查询和分析。结构化存储Pandas CSV/Excel。适合数据分析师直接进行后续处理。异步框架可选但推荐aiohttpasyncio。当需要采集成千上万家店铺时同步请求会非常慢。使用异步IO可以极大提升采集效率但代码复杂度也会增加。抓包与分析工具Charles/Fiddler配置手机代理抓取APP的HTTPS流量。这是逆向分析接口的起点。mitmproxy一个基于Python的中间人代理工具功能强大甚至可以实时修改请求和响应适合高级用户。反爬应对基础库fake_useragent随机生成User-Agent请求头避免因固定UA被识别。代理IP池这是应对IP封锁的核心。可以购买付费的代理IP服务或者自建代理池。请求时需要轮换使用不同的IP。注意技术选型的核心原则是“够用就好保持灵活”。初期可以从最简单的requestsSQLite开始验证流程可行性。当数据量增大、速度要求变高时再逐步引入异步、分布式队列如Redis Celery等更复杂的架构。3. 核心细节解析与实操要点这一部分我们深入每个环节的“魔鬼细节”。这些细节往往是爬虫能否稳定运行的关键。3.1 逆向分析找到并理解关键接口这是最具挑战性也最核心的一步。你需要像一个侦探一样从杂乱的网络请求中找出规律。环境配置在电脑上安装Charles并配置好SSL证书使其能解密HTTPS流量。将手机和电脑连接到同一Wi-Fi并在手机网络设置中配置手动代理指向电脑的IP和Charles的端口默认8888。操作与抓包打开手机上的美团外卖APP进行关键操作例如定位到某个地址、搜索“奶茶”、进入一家店铺、查看评论。同时在Charles中观察产生的网络请求。识别接口关注请求URL中包含wmapi、takeout、meituan等关键词的GET或POST请求。这些很可能是后端数据接口。重点查看其Query ParametersURL参数和Request Headers请求头。分析参数关键ID如wm_poi_id店铺ID、address_id地址ID、city_id城市ID等。这些是构建请求链的基础。签名与令牌寻找像token、uuid、_token、sign这样的参数。特别是sign它很可能是一个对多个参数进行加密后生成的签名用于验证请求的合法性。这是最大的难点可能需要逆向分析APP代码才能找到生成算法。其他必要参数platform平台如1代表Android、versionAPP版本、timestamp时间戳等。模拟请求在Python中尝试用requests库原样复制你看到的URL、请求头和参数。如果返回200状态码且数据正常恭喜你成功了一半。如果返回403、400或数据为空很可能签名验证失败或缺少某个关键头信息。实操心得不要试图一次性分析所有接口。先从最简单的、不需要复杂签名的接口入手比如根据城市名获取city_id的接口。逐步推进积累信心和经验。将抓包过程中发现的固定请求头如User-Agent,Referer,Content-Type记录下来在代码中统一设置。3.2 请求头管理与会话保持服务器会通过请求头来识别客户端。一个看起来像真实APP的请求头集合至关重要。import requests session requests.Session() # 设置一个看起来像真实APP的请求头 headers { User-Agent: Mozilla/5.0 (Linux; Android 10; SM-G975F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.120 Mobile Safari/537.36, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Referer: https://h5.waimai.meituan.com/, # 模拟从H5页面跳转过来 Origin: https://h5.waimai.meituan.com, # 可能还需要一些APP特有的头如设备ID、渠道号等这需要从抓包中获取 } session.headers.update(headers) # 后续所有请求都使用这个session它会自动管理cookies response session.get(https://api.example.com/shop/list, paramsparams)使用Session对象的好处是它可以自动处理Cookies在一次会话中保持某些登录或认证状态如果接口需要的话。3.3 数据解析与清洗接口返回的数据通常是JSON格式但结构可能嵌套很深且包含我们不需要的字段。import json def parse_shop_list(json_data): 解析商家列表JSON数据 shops [] data json.loads(json_data) # 如果response.json()可以直接用 # 实际路径需要根据抓包结果确定这里仅为示例 for item in data.get(data, {}).get(poilist, []): shop { poi_id: item.get(id), name: item.get(name), month_sale: item.get(month_sale), # 月售 avg_delivery_time: item.get(avg_delivery_time), # 平均送达时间 distance: item.get(distance), # 距离 avg_score: item.get(avg_score), # 评分 latitude: item.get(latitude), longitude: item.get(longitude), address: item.get(address), } # 清洗数据处理可能为None的字段 shop[month_sale] shop[month_sale] or 0 shops.append(shop) return shops注意事项一定要做好异常处理和数据清洗。字段可能缺失None数字可能是字符串形式文本里可能包含多余的空格、换行符或表情符号。使用.get()方法安全地访问字典键值并编写清洗函数统一处理。3.4 存储设计根据数据量和使用场景设计数据库表结构。一个简单的设计可能包含以下几张表城市/区域表存储city_id,city_name,district_name等。商家表核心表存储poi_id主键、名称、地址、坐标、评分、月售、起送价、配送费、联系电话等。商品分类表与商家关联存储分类ID、分类名、所属商家ID。商品表与分类关联存储商品ID、名称、价格、原价、描述、图片链接、月售、所属分类ID。评价表与商家关联存储评价ID、内容、评分、用户昵称、推荐菜、评价时间、所属商家ID。使用SQLAlchemy这样的ORM库可以让你用Python类来操作数据库更加方便。from sqlalchemy import create_engine, Column, Integer, String, Float, Text, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship Base declarative_base() class Shop(Base): __tablename__ shops id Column(Integer, primary_keyTrue) poi_id Column(String(50), uniqueTrue, nullableFalse) name Column(String(200)) address Column(String(500)) latitude Column(Float) longitude Column(Float) avg_score Column(Float) month_sale Column(Integer) # ... 其他字段 # 定义关系 products relationship(Product, back_populatesshop) # 初始化数据库连接 engine create_engine(sqlite:///meituan_data.db) Base.metadata.create_all(engine) Session sessionmaker(bindengine) db_session Session()4. 实操过程与核心环节实现让我们以一个具体的例子串联起从启动到存储的完整流程。假设我们的目标是抓取“北京市海淀区中关村”附近所有“咖啡厅”的商家列表和基础信息。4.1 步骤一获取目标区域的地理编码首先我们需要将“北京市海淀区中关村”转换为美团系统内部可识别的address_id。通过抓包分析我们可能找到一个地理搜索接口。import requests import time import hashlib def get_location_id(keyword, city_name): 根据关键词和城市名获取地址ID示例非真实接口 url https://api.waimai.meituan.com/location/search params { keyword: keyword, city: city_name, platform: 1, timestamp: int(time.time()), # sign: ... # 签名通常需要计算这里省略 } headers { ... } # 你的请求头 resp requests.get(url, paramsparams, headersheaders) if resp.status_code 200: data resp.json() # 假设返回数据结构中有address_id address_id data.get(data, {}).get(address_id) return address_id else: print(f获取地址ID失败: {resp.status_code}) return None address_id get_location_id(中关村, 北京) print(f中关村的address_id是: {address_id})4.2 步骤二获取商家列表拿到address_id后调用商家列表接口。这里需要处理分页。def fetch_shop_list_by_page(address_id, category咖啡, page1, page_size20): 分页获取商家列表 url https://api.waimai.meituan.com/poi/list params { address_id: address_id, category: category, page: page, page_size: page_size, sort_type: 7, # 假设7代表综合排序 platform: 1, timestamp: int(time.time()), # sign: ... } resp requests.get(url, paramsparams, headersheaders) if resp.status_code 200: return resp.json() # 返回整个JSON数据供后续解析 else: print(f获取第{page}页列表失败: {resp.status_code}) return None def get_all_shops(address_id, category, max_pages50): 获取所有页的商家列表 all_shops [] page 1 while page max_pages: print(f正在抓取第{page}页...) data fetch_shop_list_by_page(address_id, category, page) if not data: break shop_list parse_shop_list(data) # 使用之前写的解析函数 if not shop_list: # 如果解析出的列表为空说明没有更多数据了 break all_shops.extend(shop_list) # 判断是否还有下一页根据返回数据中的has_next字段或列表长度判断 if len(shop_list) 20: # 假设每页20条不满20条说明是最后一页 break page 1 time.sleep(1.5) # 非常重要的延迟避免请求过快 return all_shops shops get_all_shops(address_id, 咖啡) print(f共获取到{len(shops)}家咖啡店。)4.3 步骤三获取商家详情并入库遍历商家列表获取每个商家的详细信息并存入数据库。def fetch_shop_detail(poi_id): 根据店铺ID获取详情 url fhttps://api.waimai.meituan.com/poi/detail params {poi_id: poi_id, platform: 1, timestamp: int(time.time())} resp requests.get(url, paramsparams, headersheaders) if resp.status_code 200: return resp.json() return None def save_shop_to_db(shop_detail_dict): 将商家详情字典保存到数据库 # 假设我们已经有了Shop模型和db_session # 先从shop_detail_dict中提取需要的字段 poi_id shop_detail_dict.get(poi_id) # 检查是否已存在 existing db_session.query(Shop).filter_by(poi_idpoi_id).first() if existing: print(f店铺 {poi_id} 已存在更新数据。) # 更新现有记录 existing.name shop_detail_dict.get(name) existing.month_sale shop_detail_dict.get(month_sale) # ... 更新其他字段 else: print(f新增店铺 {poi_id}。) new_shop Shop( poi_idpoi_id, nameshop_detail_dict.get(name), month_saleshop_detail_dict.get(month_sale), # ... 设置其他字段 ) db_session.add(new_shop) db_session.commit() # 主循环 for shop in shops: poi_id shop[poi_id] print(f处理店铺: {shop[name]} (ID: {poi_id})) detail_data fetch_shop_detail(poi_id) if detail_data: detail_dict parse_shop_detail(detail_data) # 需要编写详情解析函数 save_shop_to_db(detail_dict) time.sleep(1.2) # 在详情请求间也加入延迟核心环节要点延迟是美德在请求间加入time.sleep()是应对反爬最基本、最有效的手段之一。根据目标网站的反爬强度设置1到3秒甚至更长的随机延迟。异常处理要周全网络超时、连接错误、JSON解析错误、数据库操作失败……每一个环节都可能出错。使用try...except包裹关键代码并记录日志便于排查。增量更新像商家评分、月售量是经常变动的。我们的爬虫应该支持增量更新即只抓取新增的商家或更新已有商家的变动信息。这可以通过对比数据库中的poi_id和update_time如果有来实现。5. 常见问题与排查技巧实录在实际操作中你会遇到各种各样的问题。下面是我踩过的一些坑和总结的排查思路。5.1 请求返回403/418等错误码问题直接返回403 Forbidden或418 Im a teapot这是一个真实的HTTP状态码常被用来表示请求被反爬系统拒绝。排查检查请求头对比你的请求头和抓包看到的真实请求头是否缺少了Referer、Origin、X-Requested-With等关键字段User-Agent是否太像爬虫检查签名这是最常见的原因。确认sign或其他签名参数的计算方式是否正确。可能需要动态调试或逆向APP。检查IP你的IP是否已经被封禁尝试用手机4G网络访问同一个接口如果正常说明你的服务器IP可能被识别。解决方案是使用代理IP池。检查Cookies/Tokens某些接口需要先完成一个登录或初始化流程获取有效的token或session。确保你的请求携带了正确的认证信息。5.2 返回数据为空或结构不对问题状态码是200但data字段为空或者JSON结构和你之前抓包看到的不一样。排查参数错误仔细检查请求参数特别是city_id、address_id、poi_id等关键ID是否正确。一个错误的ID可能导致查询不到数据。接口已更新商业APP的接口可能会频繁变动。重新抓包确认接口URL和参数是否已经改变。地理限制有些数据可能只在特定地理区域返回。确保你模拟的定位信息如通过参数传递的经纬度是有效的。打印完整响应不要只解析data先把整个响应文本打印出来看看可能错误信息在别的字段里。5.3 爬虫运行一段时间后突然失效问题爬虫跑了几个小时或几天后开始大量返回错误或空数据。排查IP被封最可能的原因。即使加了延迟长期用同一个IP高频率请求也会被识别。必须使用代理IP并设置IP轮换策略。Token过期如果接口依赖token它可能有有效期如2小时。需要实现一个刷新token的机制。行为模式被识别你的请求节奏太规律了如固定每秒1次。引入随机延迟time.sleep(random.uniform(1, 3))并模拟更人性化的操作序列如先翻几页列表再点进详情中间有停顿。服务器限流对方服务器可能对单位时间内的请求次数做了限制。除了降低频率没有太好办法。5.4 数据解析出错问题json.loads()失败或者访问字典键时抛出KeyError。排查响应不是JSON先用resp.text[:500]看看返回的前500个字符是什么可能是HTML错误页面。字段缺失或为null永远使用.get(key, default_value)来安全地访问字典。在解析函数中对每个字段都做好默认值处理。编码问题确保响应内容的编码正确通常是utf-8。resp.encoding utf-8可以强制设置。5.5 性能与效率问题问题抓取几万条数据太慢了。优化异步爬虫将同步的requests改为异步的aiohttp可以同时发起数十上百个请求。这是提升速度最有效的方法。优化数据库写入不要每条数据都commit一次可以积累一定数量如100条后批量提交。分布式爬虫对于超大规模采集可以考虑使用Scrapy-Redis等框架将任务分发到多台机器上执行。我的一个关键心得在开始大规模爬取之前先用小规模、慢速的脚本验证整个数据链路从请求到存储是通的。然后再逐步加入代理、异步、错误重试等增强功能。不要试图一开始就构建一个完美的、高性能的爬虫迭代开发会更有效率。另外务必尊重robots.txt如果有并控制爬取速度避免对目标服务器造成不必要的压力。数据采集是为了分析和创造价值技术的使用应当合理且合规。
