基于Scrapy的七麦APP数据爬虫工程化实践
2026/9/16 17:06:30 网站建设 项目流程

简介:针对七麦APP数据采集需求的 Python 爬虫项目,基于 Scrapy 框架构建,适合爬虫初学者和移动应用数据分析人员参考。程序以命令行方式运行,覆盖 URL 收集、请求发送、HTML 解析和数据存储等核心环节,同时加入 robots 协议与反爬应对策略,代码结构清晰,便于二次开发。压缩包共 11 个文件,其中包含 9 个 Python 脚本、1 个 Markdown 说明文档和 1 个 Scrapy 配置文件,整体大小约 20KB,轻量易读。已有 331 人学习下载。借助该项目可快速了解 Scrapy 工程目录组织方式,掌握 selector 提取、管道存储、中间件配置等关键技巧,同时通过阅读源码理解爬虫工作流程与规则遵守意识,为后续开发分布式爬虫或数据监测工具打下基础。

1. 七麦APP数据爬虫解压后,是一份能直接跑起来的 Scrapy 工程

七麦APP数据爬虫.zip 解压后是一个完整的 Scrapy 工程,核心目录是 crawl_qimai。它解决的问题很具体:把七麦数据(qimai.cn)上的 App Store 榜单、关键词覆盖和排名变化自动拉下来,落到本地文件或数据库,供运营和数据分析做竞品监控。这个场景不是临时抓一次就完,而是每天都要跑,所以工程里涉及了请求会话保持、下载中间件、Pipeline 存储和增量参数化。适合已经用 requests 写过小爬虫、准备转向工程化项目的读者;也适合需要从零搭建 App 数据采集管线的爬虫工程师。它不是那种“一个脚本爬全网”的玩具,而是围绕七麦这一垂直目标做的可维护代码。

2. Scrapy 项目结构与启动链路:从 crawl_qimai 目录说起

2.1 为什么是 Scrapy,而不是 requests + BeautifulSoup

经常有人问我,爬七麦数据用 requests 加 BeautifulSoup 不就行了吗?如果只抓一次,确实可以;但一旦要按天抓、要断点续爬、要统一处理登录态和存储,requests 脚本会越写越乱。Scrapy 把调度器、去重队列、下载器、中间件、Pipeline、日志统计都拆成了固定组件,你只需要在 spider 里写“要抓什么”,在 pipeline 里写“存到哪里”,其余环节由框架接管。这个 zip 包里的 scrapy.cfg 和 crawl_qimai 目录,就是标准 Scrapy 工程该有的样子。

目录结构如下:

SJT-code/ └── crawl_qimai/ ├── scrapy.cfg ├── crawl_qimai/ │ ├── __init__.py │ ├── items.py │ ├── middlewares.py │ ├── pipelines.py │ ├── settings.py │ └── spiders/ │ ├── __init__.py │ └── qimai.py └── README.md

scrapy.cfg是命令行入口配置,里面通常写着default = crawl_qimai.settings,告诉 Scrapy 去哪个模块加载配置。内层crawl_qimai是真正的项目包,spiders/放爬虫类,middlewares.py放下载中间件,pipelines.py放数据存储逻辑。README 一般记录运行方式和已知坑,拿到压缩包后建议先看它,但工程行为最终由 settings.py 决定。

2.2 从 scrapy.cfg 到 Spider 的执行顺序

运行scrapy crawl qimai时,Scrapy 会先读 scrapy.cfg,找到项目配置;然后加载 settings.py,注册中间件、Pipeline 和爬虫类;通过名称匹配到spiders/qimai.py里的 QimaiSpider,调用它的start_requestsstart_urls生成第一批请求,交给引擎调度。下面是一个最简可运行的爬虫骨架:

# crawl_qimai/spiders/qimai.py import scrapy class QimaiSpider(scrapy.Spider): name = "qimai" allowed_domains = ["qimai.cn"] start_urls = ["https://www.qimai.cn/"] def parse(self, response): self.logger.info("status=%s url=%s", response.status, response.url)

name是爬虫唯一标识,scrapy crawl qimai后面的参数就是它。allowed_domains是域名白名单,不在名单里的 URL 会被直接过滤,但这不是安全机制,只是避免手滑爬到站外。parse是默认回调,response.status能快速判断请求是否被拦。

启动命令:

scrapy crawl qimai -o first_run.json

-o参数会把最终 yield 出的 item 自动序列化到 JSON 文件,适合验证阶段。这里没有写任何存储代码,就能看到 Scrapy 的“导出”能力。实际项目中,-o更多用于临时测试,正式存储交给 Pipeline。

2.3 settings.py 中的关键参数与空跑验证

settings.py 是整个爬虫的“总闸”,下面几个参数在七麦场景下尤其要调好:

参数典型值作用
ROBOTSTXT_OBEYFalse是否遵守 robots.txt,七麦对爬虫不友好,一般关闭,但必须自己控频
CONCURRENT_REQUESTS4同一时间最大请求数,调小能降低被封概率
DOWNLOAD_DELAY2.0每个请求之间最少等待的秒数,值越大越安全
RANDOMIZE_DOWNLOAD_DELAYTrue在延迟基础上乘以 0.5~1.5 的随机因子,模拟人工浏览
COOKIES_ENABLEDTrue保持会话状态,访问需要登录态的接口时得打开
DEFAULT_REQUEST_HEADERSdict给所有请求带默认请求头,实际请求头可在中间件覆盖

注意COOKIES_ENABLED默认是 False,Scrapy 为了性能不维护 cookie 会话。但七麦的榜单接口依赖登录 Cookie,所以要么在请求里显式传cookies,要么打开这个选项。打开后也要小心同一个 Cookie 并发请求太多,容易触发风控。

写一个最小验证:

scrapy list scrapy crawl qimai --nolog -o /tmp/qimai_home.json

scrapy list会列出所有可用爬虫名称,看到qimai说明项目加载成功。--nolog关闭日志输出,-o /tmp/qimai_home.json将输出写到临时文件。如果文件内容为空,先别急着改解析逻辑,用scrapy shell单独看一下首页响应,判断是否被反爬。

提示:解压后先安装依赖再运行,常见依赖是 scrapy 和 fake-useragent。pip install scrapy fake-useragent即可,MySQL 支持需要额外装 pymysql。

3. 登录态、请求头与下载中间件:把七麦的反爬挡在门外

3.1 七麦数据接口的特点:能拿到 JSON 就别去解析 HTML

七麦网站页面是服务端渲染加异步接口混合的形式,直接解析 HTML 会遇到乱码、动态 token 和大量无关 DOM 结构。更稳定的做法是打开浏览器开发者工具,筛选XHR请求,找到榜单数据接口。七麦接口通常返回 JSON,请求 URL 类似:

https://api.qimai.cn/rank/index?brand=iphone&country=cn&genre=36&date=2025-06-04

但接口要求必须带登录后的 Cookie 和请求头,否则会返回 302 或业务错误码。常见做法是:在浏览器里登录七麦,然后从 Network 面板复制 Cookie、User-Agent、Referer,写进 spider 的启动请求里。

# qimai.py 部分代码 COOKIES = { "qimai_session": "这里填浏览器复制出的会话值", "gr_user_id": "用于辅助识别身份", } HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://www.qimai.cn/rank", "Accept-Language": "zh-CN,zh;q=0.9", } def start_requests(self): for url in self.start_urls: yield scrapy.Request( url, cookies=COOKIES, headers=HEADERS, callback=self.parse, )

cookies参数接收字典,Scrapy 会拼成 Cookie 头;如果 Cookie 里有特殊字符,字典写法更安全。Referer在七麦场景下很关键,部分接口会校验来源页面,如果 Referer 不对,即使 Cookie 正确也可能拿不到数据。User-Agent要完整带上版本号,不能只写一个Mozilla/5.0,否则某些接口的签名校验会拒绝响应。

3.2 自定义下载中间件:随机 User-Agent 与请求间隔

同一个 UA 长时间高频访问,很快会被识别。我一般会维护一个 UA 列表,在下载中间件里随机选择。

# middlewares.py import random import time class RandomUserAgentMiddleware: def __init__(self, ua_list): self.ua_list = ua_list @classmethod def from_crawler(cls, crawler): return cls(crawler.settings.getlist("USER_AGENT_LIST")) def process_request(self, request, spider): request.headers["User-Agent"] = random.choice(self.ua_list)

这个中间件的好处是每个请求都会换 UA,不需要在 spider 里逐个设置。然后在 settings.py 里注册:

DOWNLOADER_MIDDLEWARES = { "crawl_qimai.middlewares.RandomUserAgentMiddleware": 300, } USER_AGENT_LIST = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 13_1) AppleWebKit/605.1.15", ]

数字 300 是执行顺序,越小越靠近引擎。

除了随机 UA,还要控制请求节奏。Scrapy 的DOWNLOAD_DELAY可以配合RANDOMIZE_DOWNLOAD_DELAY

DOWNLOAD_DELAY = 2.0 RANDOMIZE_DOWNLOAD_DELAY = True

DOWNLOAD_DELAY是基础延迟,开启随机化后,实际延迟在 1 秒到 3 秒之间波动。这个节奏比固定延迟更接近人工浏览。如果想要更大的随机范围,可以在中间件里自己time.sleep,但要注意这会阻塞下载线程,并发较高时反而容易积压请求。下面的中间件适合低频采集:

class ThrottleMiddleware: def process_request(self, request, spider): time.sleep(random.uniform(1.5, 3.5))

不过这属于“大炮打蚊子”,一般靠DOWNLOAD_DELAY就够用。

从实际反爬响应来看,下面这些状态码最常见:

状态码常见原因处理建议
200正常返回正常解析
302需要登录或 Cookie 失效重新登录七麦,更新 Cookie
403被反爬拦截调低并发、调大延迟、换一个 UA
429请求过于频繁停止一段时间再跑,或换网络出口

3.3 抓包失败的现场排查

很多人在爬七麦时遇到 app 抓包失败,但造成失败的往往不是抓包工具,而是请求没带上正确的会话信息。当你发现输出文件为空,第一步不是改解析逻辑,而是打印原始响应。

def parse(self, response): self.logger.warning( "status=%s content_type=%s body_head=%s", response.status, response.headers.get("Content-Type"), response.text[:200], )

如果响应头里是text/html且 body 是一段登录跳转代码,说明 Cookie 失效。如果status=403,说明反爬直接拒绝了请求,此时要重点检查 UA 和请求频率。如果status=200但 JSON 里的业务code不是 200,说明签名或参数有问题。

继续调试可以进入交互环境:

scrapy shell "https://api.qimai.cn/rank/index?brand=iphone&country=cn&genre=36"

在 shell 里可以直接执行response.json()查看数据,反复测试解析逻辑,不需要每次改完代码都重跑整个爬虫。

提示:如果接口返回的 JSON 里msg提示“签名错误”,多半是时间戳或analysis参数过期。先把浏览器里的完整接口 URL 复制下来,不要手动改日期,确认能拿到数据后再做参数化。

4. 数据解析与 Pipeline 落库:把榜单与关键词结构化

4.1 榜单响应里的数据层级

七麦的榜单接口经过签名校验后,返回的 JSON 结构大致如下:

{ "code": 200, "data": { "list": [ { "app_id": 414478124, "app_name": "微信", "company": "Tencent", "bundle_id": "com.tencent.xin", "rank": 1, "genre": "社交" } ] } }

具体字段名可能随接口变化,但思路一致:拿到response.json()之后,先用list(response.json().keys())确认顶层键,再按层取。不要直接用response.body正则匹配,JSON 解析更快也更稳定。

# qimai.py 的 parse 方法 def parse(self, response): payload = response.json() if payload.get("code") != 200: self.logger.warning("qimai api error: %s", payload.get("msg")) return records = payload["data"]["list"] for rec in records: item = QimaiItem( app_id=rec.get("app_id"), app_name=rec.get("app_name"), company=rec.get("company"), bundle_id=rec.get("bundle_id"), rank=rec.get("rank"), genre=rec.get("genre"), ) yield item

这里先校验code,再取data.list。很多爬虫只检查 HTTP 状态码,忽略了业务状态码;七麦接口即使 HTTP 200,也可能因为签名过期返回code: 4001之类的业务错误。yield item会把 Item 交给 Pipeline,继续往后走。

4.2 用 Item 固化字段,避免漏字段

随着抓取范围扩大,字段会越来越多。如果直接用字典,字段名在 spider、pipeline、报表之间靠字符串拼接,容易漏。定义 Item 相当于给数据加了一层“结构约定”。

# items.py import scrapy class QimaiItem(scrapy.Item): app_id = scrapy.Field() app_name = scrapy.Field() company = scrapy.Field() bundle_id = scrapy.Field() rank = scrapy.Field() genre = scrapy.Field()

scrapy.Field()本身不限制类型,但如果之后要写入 MySQL,建议在 pipeline 里对app_idrankint()转换。Item的另一个好处是,保存到 CSV 时csv.DictWriter可以直接按字段名顺序输出,不会出现字段错位。

字段映射可以整理成下面的表格,方便对照接口文档:

目标字段来源 key类型说明
app_idapp_idintApp Store 应用唯一标识,适合做主键
app_nameapp_namestr应用显示名称
companycompanystr开发者主体
bundle_idbundle_idstr包名,iOS 上通常和证书关联
rankrankint排名数值
genregenrestr榜单分类

4.3 Pipeline 写 CSV 和 MySQL:先留底,再入库

Pipeline 的职责很纯粹:接收 Item、清洗、写入目标。下面这个 pipeline 同时写了 CSV 和 MySQL,方便先留底再入库。

# pipelines.py import csv import pymysql class CsvPipeline: def open_spider(self, spider): self.file = open("qimai_data.csv", "w", newline="", encoding="utf-8") self.writer = csv.DictWriter( self.file, fieldnames=["rank", "app_id", "app_name", "company", "bundle_id", "genre"], ) self.writer.writeheader() def close_spider(self, spider): self.file.close() def process_item(self, item, spider): self.writer.writerow(dict(item)) return item

open_spider在爬虫启动时执行,close_spider在结束时执行。这样保证 CSV 文件只打开一次,不会每条数据都重新开文件。dict(item)可以把 Item 转成普通字典,csv.DictWriterfieldnames顺序写入。

MySQL 版 pipeline 稍微复杂一点,但核心就两条:插入和提交。

class MysqlPipeline: def open_spider(self, spider): self.conn = pymysql.connect( host="127.0.0.1", user="root", password="123456", database="qimai", charset="utf8mb4", ) self.cursor = self.conn.cursor() def process_item(self, item, spider): sql = ''' INSERT INTO app_rank(app_id, app_name, company, bundle_id, rank, genre) VALUES (%s, %s, %s, %s, %s, %s) ''' self.cursor.execute(sql, ( item["app_id"], item["app_name"], item["company"], item["bundle_id"], item["rank"], item["genre"], )) self.conn.commit() return item def close_spider(self, spider): self.conn.close()

commit放在每条数据之后,数据量小没问题,但数据量到几十万条时会影响速度。常见的做法是累积self.counter,每 500 条 commit 一次。另外注意process_item必须return item,否则多个 pipeline 串联时,后面的 pipeline 拿不到数据。

在 settings.py 里启用:

ITEM_PIPELINES = { "crawl_qimai.pipelines.CsvPipeline": 300, "crawl_qimai.pipelines.MysqlPipeline": 350, }

数字小的先执行。这里 CSV 先写,MySQL 后写,坏数据也能在 CSV 里找到原始记录。

4.4 去重逻辑与重复请求的取舍

Scrapy 默认使用 URL 去重,同一 URL 不会重复请求。但榜单数据每天都在变,如果你把日期拼在 URL 里,URL 去重是没问题的;如果只爬一个固定 URL,就会错过第二天的数据。

关闭 URL 去重可以在 settings.py 里设置:

DUPEFILTER_CLASS = "scrapy.dupefilters.BaseDupeFilter"

这样所有重复 URL 都会被重新请求。但代价是如果某个 URL 请求失败,重试机制也会反复请求同一个地址。更稳妥的做法是在 pipeline 里基于app_id + 日期做业务去重,URL 去重保持开启,这样既不浪费请求,也不会漏数据。

5. 增量更新与定时部署:按业务节奏跑起来

5.1 用启动参数控制日期范围

把日期、榜单分类做成启动参数,是爬虫工程化的第一步。

def __init__(self, date=None, genre=None, *args, **kwargs): super().__init__(*args, **kwargs) self.date = date or "latest" self.genre = genre or "36"

启动时用-a传参:

scrapy crawl qimai -a date=2025-06-05 -a genre=6014 -o output/rank_2025-06-05.csv

-a参数会传入 spider 的__init__,在命令行覆盖默认值。这样可以做到同一天只抓一次,第二天用新日期继续跑。

5.2 定时运行和结果校验

到了生产环境,通常用 crontab 每天定时运行:

0 6 * * * cd /data/SJT-code/crawl_qimai && scrapy crawl qimai -a date=$(date +\%F) -o /data/output/rank_$(date +\%F).csv >> /data/logs/qimai.log 2>&1

注意 cron 里的%要转义成\%,否则会被当作换行符。这句话的意思是:每天早上六点进入项目目录,用当天日期抓取七麦榜单,输出到带日期的 CSV 文件,并把日志追加到 qimai.log。如果担心日志过大,再加一句清理命令:

find /data/output -name '*.csv' -mtime +90 -delete

这条命令会删除 90 天前的旧 CSV 文件,避免磁盘写满。运行结束后,用wc -l检查当天文件行数:

wc -l /data/output/rank_2025-06-05.csv

如果行数和昨天相差悬殊,优先检查七麦接口的 Cookie 是否过期、请求频率是否触发风控。在 spider 里打印payload.get("code")payload.get("msg"),能在第一时间区分是账号失效还是频率受限。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询