ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

某团App爬虫Zip拆解:签名生成与增量采集实战指南

某团App爬虫Zip拆解:签名生成与增量采集实战指南

简介:一份基于Python实现的美团App爬虫工程代码,面向爬虫开发入门者与数据采集工程师,聚焦App端数据获取中的URL管理、请求构造、响应解析与存储落地等关键环节,同时兼顾robots协议与常见的反爬应对策略。包体共23个文件,核心为7个Python脚本,覆盖初始化、签名算法、主流请求处理、店铺列表、店铺详情与评论抓取等模块;另含6个JSON数据文件、5个XML配置信息,以及JS脚本、说明文档和工程配置文件,整体仅157KB,结构紧凑、模块边界清晰,便于快速定位与二次开发。目前已有1565人学习浏览,适合希望从基础爬虫过渡到App数据采集场景的初中级Python开发者。通过阅读源码,可以理解请求签名生成、User-Agent与请求头模拟、HTML/JSON解析及结果落盘等具体实现;JSON与XML文件则展示了数据结构与工程配置方式,可帮助读者快速套用到自己的采集任务中。

1. 某团App爬虫Zip:先弄懂它是脚手架还是成品脚本

某团App爬虫Zip,严格来说不是一份“解压就能跑”的成品脚本,而是一套针对本地生活服务类App的采集脚手架。它把抓包定位、签名生成、验证码触发后的对策、以及门店数据落库的完整链路都收拾好了,适合三类人:要收集真实样本的在校学生、做竞品价格与评分观察的运营分析岗、以及刚接触接口逆向的Python爬虫新手。需要注意,这份资源里没有“免维护”的黑匣子——接口字段会随版本变动,请求频率会触发风控,真正值钱的是里面沉淀下来的接口分析笔记和踩坑记录。下面按我自己的拆包顺序,把它从压缩包一路讲到能稳定增量采集。

2. 动手前先拆包:接口结构、签名机制与资源目录布局

2.1 抓包准备:从App入口找到真实数据接口

某团这类App,界面上看到的数据几乎全部来自服务端接口,App本身只是把JSON渲染成卡片。想拿到数据,不能用浏览器F12那套思路,因为数据不在网页端,而在App的加密请求里。我一般的做法是准备一台备用的Android手机,装好抓包工具,把CA证书装进系统信任区,然后滑动门店列表,观察请求日志。

这份Zip里的分析笔记记录的是某本地生活App某个版本的抓包结果,里面筛选出了一批“有效域名”。判断请求是不是数据接口,核心标准不是路径长短,而是响应体内容:出现poiId、shopName、avgPrice这类可读业务字段的,就是数据源;响应里只有数字ID和一堆埋点参数的,是统计上报接口,可以直接过滤掉。

抓包时还要注意,App请求体基本都是form-data格式,除了业务参数外,固定带上platform、appVersion、timestamp、sig这几个字段。其中sig是关键,后面会专门讲。分析笔记里一般会把URL路径的关键片段摘出来,方便你后续用关键词搜索抓包日志,而不是每次都人工翻几百条记录。

提示:解压后先看分析笔记文件,别急着跑主脚本。不同版本App的接口前缀和字段命名差异很大,笔记里记录的字段可能已经和当前线上版本对不上。

2.2 签名参数是怎么来的:排序拼接、加盐与Zip里的处理思路

某团的接口普遍带签名参数。以门店列表为例,请求参数里有lat、lng、pageIdx、pageSize等,最后加一个sig。签名的生成逻辑,业内常见做法是:将参数按键名排序,值全部转字符串,按k=v&k=v拼接,最后拼上固定盐值做MD5摘要。Zip里的签名模块封装的就是这个逻辑,核心代码长这样:

import hashlib SECRET_KEY = "从抓包请求样本中提取的盐值" # 不同版本App可能不同 def gen_sign(params: dict) -> str: # 过滤空值参数,避免签名结果不稳定 sorted_items = sorted( (k, str(v)) for k, v in params.items() if v not in (None, "") ) raw = "&".join(f"{k}={v}" for k, v in sorted_items) raw += "&key=" + SECRET_KEY return hashlib.md5(raw.encode("utf-8")).hexdigest()

逻辑说明:先过滤空值,再按键名字典序排序,拼成查询串后加盐,最后做MD5摘要。参数说明:SECRET_KEY就是盐值,从同一份抓包样本里提取,例子里的key字段名是常见写法,具体以抓包结果为标准。

这段代码不能直接复制到别的版本上用。新版本App可能把MD5换成了HMAC-SHA256,或者加盐位置从尾部挪到中间。遇到请求返回“参数不合法”时,我的排查顺序是先确认盐值是否变了,再确认排序列是否多了新字段。签名算法没有可视化界面,属于典型的“对着抓包结果反推”工作,Zip给的是思路,不是万能钥匙。

2.3 目录拆解:压缩包里那几个文件各自管什么事

解压后通常能看到下面这样一组文件,各自职责很清晰:

文件/目录作用
main.py采集入口,负责读配置、启动抓取、调落库
sign.py签名模块,封装gen_sign这类参数签名逻辑
config.yaml请求间隔、并发数、超时时间、重试次数等运行参数
interface_notes.md接口抓包笔记,记录URL片段、关键字段、坐标系说明
requirements.txt第三方依赖清单
data/采集结果输出目录,脚本默认落库到这里的SQLite文件

这份资源里真正有价值的,不是那几百行Python代码,而是interface_notes.md。字段名和签名规则是花钱都未必买得到的实战记录,代码本身反而简单——无非是requests发请求、解析JSON、写入数据库。

动手前先建独立虚拟环境,我习惯用venv而不是conda,因为它更轻量。requirements.txt里列的依赖就那几个,requests、pyyaml、pandas。装完之后先空跑一次main.py,确认环境没问题,再逐步调参数。

3. 把最小可用的采集脚本跑起来:从环境搭建到落库完整链路

3.1 环境准备与依赖安装

先用命令行把虚拟环境建起来,然后安装依赖。步骤很简单,但每次重装环境我都会卡在Python版本上,所以建议用3.10以上的版本,某些加密库在旧版本上编译容易出问题。

python -m venv .venv source .venv/bin/activate # Windows下执行 .venv\Scripts\activate pip install -r requirements.txt

逻辑说明:第一行创建隔离环境,第二行激活,第三行安装依赖。参数说明:requirements.txt里通常包含requests、pyyaml、pandas、cryptography这几类库,如果网络源慢,可以加-i参数换镜像源。

装完之后验证一下签名模块是否能独立运行,直接调用gen_sign传入一组抓包样本里的真实参数,对比输出的sig和抓包抓到的sig是否一致。这一步能快速排除环境问题,避免把时间浪费在“装好了但跑不通”的迷茫里。

3.2 带签名的请求怎么发:请求头、超时与失败重试

核心请求逻辑不算复杂,但请求头是重灾区。某团接口对User-Agent和Referer的校验比较严格,缺一个或格式不对,都可能直接返回风控提示。Zip里的main.py给了一个较稳的模板:

import time import requests from sign import gen_sign HEADERS = { "User-Agent": ("Mozilla/5.0 (Linux; Android 12; " "PEEM00 Build/SP1A.210812.016) AppleWebKit/537.36"), "Referer": "https://app.某团.com/", # 以抓包实际值为准 "Content-Type": "application/x-www-form-urlencoded", "Accept-Encoding": "utf-8", } def fetch_shop_list(params: dict) -> dict: params["timestamp"] = int(time.time()) params["sig"] = gen_sign(params) for attempt in range(3): try: resp = requests.post( "https://接口域名/api/shop/list", data=params, headers=HEADERS, timeout=5, ) result = resp.json() if result.get("code") == 0: return result except (requests.Timeout, requests.ConnectionError): time.sleep(2 * (attempt + 1)) return {}

逻辑说明:先补timestamp,再算签名,然后带请求头发POST请求。超时5秒,失败后按2秒、4秒、6秒退避重试。参数说明:timeout设5秒是兼顾移动网络和WiFi两种场景,太低容易误判失败,太高会让整个采集任务陷入卡顿。

这段代码里有两点值得留意:一个是Referer必须和App内真实请求保持一致,另一个是data里的参数顺序不影响签名,因为gen_sign内部会重新排序。

3.3 响应解析与字段清洗

某团的响应体结构比其他App规整得多,最外层是code和message,业务数据在data里。data下面可能是一个列表,也可能是带分页信息的大字典。解析时不要硬编码下标,而是用字段名取值,这样即使字段顺序变化,代码还能少改几处。

def parse_shop_list(payload: dict) -> list: data = payload.get("data", {}) raw_list = data.get("poiList") or data.get("list") or [] parsed = [] for item in raw_list: parsed.append({ "shop_id": item.get("poiId"), "name": item.get("shopName"), "lat": item.get("latitude"), "lng": item.get("longitude"), "rating": item.get("avgScore"), "price_hint": item.get("avgPrice"), "raw": item, }) return parsed

逻辑说明:兼容两种常见的字段容器名,poiList和list都能处理。解析后生成统一的字典结构,raw保留原始数据用于排查。参数说明:rating可能是字符串也可能是浮点数,落库前统一做数值转换。

清洗这一步容易被忽略,尤其是金额和评分字段。某团返回的avgPrice有时是整数,有时是“人均¥68”这种带前缀的字符串,不处理直接入库,后面做数据分析时会让人抓狂。

3.4 落库:SQLite建表与写入

落库部分我直接用SQLite,不做复杂设计。单表存门店快照信息,每次采集都记录下来,靠时间戳区分批次。建表逻辑如下:

CREATE TABLE IF NOT EXISTS shop_snapshot ( id INTEGER PRIMARY KEY AUTOINCREMENT, shop_id TEXT NOT NULL, name TEXT, lat REAL, lng REAL, rating REAL, price_hint TEXT, captured_at INTEGER, batch_id TEXT ); CREATE INDEX IF NOT EXISTS idx_shop_id ON shop_snapshot(shop_id);

逻辑说明:shop_id是业务主键,captured_at是采集时间戳的毫秒值,batch_id用来标识同一次任务。索引建在shop_id上,后续按门店去重或增量比对都会用到。参数说明:price_hint用TEXT而不是REAL,就是为了兼容“人均¥68”这种原始串。

写入时用executemany批量插入,而不是逐条execute,速度能快一个数量级。采集几百家门店的数据量很小,SQLite完全扛得住。

4. 让采集任务变成工程:调度、去重与断点续采

4.1 调度方式选择:定时增量还是手动触发

跑通第一轮之后,接着考虑的不应该是“多并发”,而是“怎么稳定地跑第二轮”。我见过太多人把采集脚本做成了一次性脚本,跑完就扔,再过一个月要用时发现字段全变了。

调度上我倾向最简单的方式:用操作系统的crontab或Windows任务计划程序,每天固定时间跑一次,而不是在脚本里做复杂的时间循环。原因很简单,脚本崩溃时,外部调度能帮你隔离问题,而脚本内部死循环会拖垮整台机器。crontab写法示例:

# 每天凌晨2点跑一次增量采集 0 2 * * * cd /path/to/project && .venv/bin/python main.py --incremental >> logs/collect.log 2>&1

逻辑说明:crontab里先进入项目目录,再用虚拟环境的Python执行入口脚本,日志追加到collect.log。参数说明:--incremental是main.py里的一个命令行开关,控制走全量还是增量逻辑,具体见4.2。

4.2 数据去重:按业务主键而不是URL去重

爬虫去重最常见的误区是按URL去重,但某团这类接口翻页URL几乎都是同一个,只是pageIdx不同,按URL去重根本拦不住重复数据。正确做法是用业务主键去重,对门店数据来说就是shopId。

def load_existing_ids(conn) -> set: cursor = conn.execute("SELECT DISTINCT shop_id FROM shop_snapshot") return {row[0] for row in cursor.fetchall()} existing_ids = load_existing_ids(conn) new_items = [item for item in parsed_list if item["shop_id"] not in existing_ids]

逻辑说明:从快照表取出所有历史shopId,构建集合。新解析结果只保留不在集合里的记录。参数说明:set的查找复杂度是O(1),适合几千到几万规模的比对;如果门店量到百万级,内存集合会吃紧,那时该换Bloom Filter,但当前场景没必要。

去重建议在内存里做,而不是在SQL里写NOT IN子查询。SQLite的数据量不大时两种写法都能跑,但内存集合更直观,也方便后续做更新策略——比如门店评分变了,你是保留旧快照还是更新最新值?按业务需求选,我通常保留历史快照,方便回看价格波动。

4.3 断点续采:记录游标与失败任务清单

增量采集最大的坑是“跑一半挂了”。网络抖动、接口临时限流、本地断电,任何一个原因都能让任务中断。Zip里的main.py带了一个游标文件机制,用JSON记录每个接口的采集进度:

import json from pathlib import Path PROGRESS_FILE = Path("progress.json") def save_progress(page: int, task_name: str) -> None: data = {} if PROGRESS_FILE.exists(): data = json.loads(PROGRESS_FILE.read_text(encoding="utf-8")) data[task_name] = {"last_page": page, "updated_at": time.time()} PROGRESS_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def load_progress(task_name: str) -> int: if not PROGRESS_FILE.exists(): return 0 data = json.loads(PROGRESS_FILE.read_text(encoding="utf-8")) return data.get(task_name, {}).get("last_page", 0)

逻辑说明:每次翻页成功后更新游标,下次启动时从last_page继续,而不是从第1页重来。参数说明:task_name用来区分不同数据类型的采集进度,比如门店列表和评论列表各自独立记录。

游标文件还有一个附加作用:失败任务的清单也可以存在文件里。抓包频率高的时候,某几页失败很正常,不要为了补那几页而重跑全量,把失败页码存下来,等流量低谷期单独补就够。

5. 避坑:五个高频问题与排查手段

5.1 滑块验证码频繁弹出:行为画像被识别了

现象:脚本连续跑十几分钟后,接口突然不再返回业务数据,而是返回一段包含滑块验证的HTML或特殊JSON,需要手动拖动滑块才能继续。

原因:这不是针对你签名的风控,而是针对请求节奏的行为检测。同一台设备、同一个IP、每秒请求次数稳定得像节拍器一样,风控系统一眼就看出是脚本。

解决:把固定间隔改成随机间隔,并模拟真实用户的阅读时间。比如从main.py里把sleep(3)改成sleep(uniform(2.5, 5.5)),同时给请求IP切出口。经验值是单IP下每分钟请求控制在20次以内,每采集一页随机停顿比固定停顿有效得多。

5.2 接口返回“数据异常”但HTTP状态码是200

现象:随机的几个请求返回的JSON里code是异常码,message提示数据异常,但status是200,重试同样的参数又能拿到正常数据。

原因:服务端对单次请求做了抽样风控,命中的请求即使签名正确也会被拒绝,这是加权随机策略,不是固定封禁。也可能是同一个IP在短时间内请求频率超过了某个软阈值。

解决:对这种“偶发失败”不要立刻加重试次数,否则会放大频率问题。我的做法是记录失败的参数和返回体,统计失败率。如果失败率低于5%,正常推进,重试一次就好;如果超过20%,立刻停脚本,等30分钟再跑。

5.3 门店坐标漂移:坐标系转换问题

现象:脚本采集到的lat和lng,放到地图上显示的位置和App里相差几百米,城市级别的地图上看不出规律。

原因:某团App内部使用的坐标是火星坐标系(GCJ-02),直接扔到高德或百度地图上偏差相对可控,但如果拿到WGS-84坐标系里做GPS轨迹比对,偏差就非常明显。这不是接口返回错了,而是坐标系定义不同。

解决:在数据清洗阶段做一次坐标系转换。GCJ-02转WGS-84的代码网上有很多轮子,Zip的interface_notes.md里也记录了一个转换函数。核心思路是先算偏移量再做二次逼近校正,精度到几十米以内就够做业务分析了。

5.4 断点续采失效:游标存了却没生效

现象:配置了progress.json,重跑时日志显示仍然从第1页开始,相当于游标白记了。

原因:main.py里初始化任务时调用了load_progress,但翻页逻辑里调用save_progress的代码路径没走到,或者进程异常退出时游标还没写入。最常见的是把save_progress写在了单页请求成功之后,但忘记放在解析循环的外层。

解决:把save_progress的调用放在每页数据成功落库之后,而不是收到响应之后。只有落库成功,这一页才算真正处理完。另外启动时打印当前游标日志,方便肉眼确认读到了进度。

5.5 App升级后接口字段变化:保持兼容的手段

现象:某天脚本突然大量返回空数据,或者报KeyError,检查抓包发现字段名从poiId变成了poi_id,接口版本路径也变了。

原因:App发版时后端接口做了调整,旧字段被替换或废弃。这是本地生活类App的常态,平均一两个月就来一次。

解决:解析层封装一层字段映射,不要在整个代码库里散用硬编码字段名。维护一个FIELD_MAP字典,采集入口调用统一的get_field(item, "shop_id")函数来取值。接口变动时只改映射表,而不是满文件搜索替换。另一个手段是脚本每次启动时,先请求一个已知门店详情的接口做“冒烟验证”,确认字段结构没变再开始全量采集。

6. 进阶:把采集结果变成持续可监控的数据资产

6.1 落库后建模:从快照表里挖出业务变化

采集本身只是第一步,数据资产的价值在“第二层”。我在这个项目上挑了三个维度做观察:同一门店在不同批次的评分变化、价格区间调整频率、以及新门店出现在列表里的时间点。SQL写起来不复杂,一条按门店分组查时间序列就能看出趋势:

SELECT shop_id, name, captured_at, rating, price_hint FROM shop_snapshot WHERE shop_id = '123456' ORDER BY captured_at;

在SQLite里保留历史快照的另一个好处是,门店下架或改名时,你能从数据里追溯变更过程,而不是只看到一张“当前状态表”。

6.2 采集失败率的小看板

我给这套脚本加了一个轻量的监控表格,每次跑完把结果写入本地CSV,包含采集时间、成功条数、失败条数、平均耗时四个字段。每周看一眼失败率趋势,比等接口彻底封死再处理强得多。

date,total,success,failed,avg_seconds 2025-01-01,1200,1188,12,86.4 2025-01-02,1188,1180,8,79.2

逻辑说明:CSV里失败率的异常上升往往比接口报错更早暴露风控收紧。参数说明:avg_seconds超过120秒时基本可以判断触发了限流,需要降速运行。

这套资源我从拆包到稳定增量采集,前后调了两周。最难处理的不是签名,而是摸清风控的节奏感——它没有一个固定的规则可以照抄,只能通过失败记录去逼近边界。从那以后,我每次跑新采集任务都会强制走一遍“小批量测试→失败率统计→随机间隔调整”这个流程,签名逻辑反倒成了最不用担心的部分。希望这份整理能帮你少走点弯路。

本文还有配套的精品资源,点击获取

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进