简介针对七麦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.mdscrapy.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_requests或start_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默认是 FalseScrapy 为了性能不维护 cookie 会话。但七麦的榜单接口依赖登录 Cookie所以要么在请求里显式传cookies要么打开这个选项。打开后也要小心同一个 Cookie 并发请求太多容易触发风控。写一个最小验证scrapy list scrapy crawl qimai --nolog -o /tmp/qimai_home.jsonscrapy 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?brandiphonecountrycngenre36date2025-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;q0.9, } def start_requests(self): for url in self.start_urls: yield scrapy.Request( url, cookiesCOOKIES, headersHEADERS, callbackself.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_DELAYDOWNLOAD_DELAY 2.0 RANDOMIZE_DOWNLOAD_DELAY TrueDOWNLOAD_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 失效重新登录七麦更新 Cookie403被反爬拦截调低并发、调大延迟、换一个 UA429请求过于频繁停止一段时间再跑或换网络出口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 失效。如果status403说明反爬直接拒绝了请求此时要重点检查 UA 和请求频率。如果status200但 JSON 里的业务code不是 200说明签名或参数有问题。继续调试可以进入交互环境scrapy shell https://api.qimai.cn/rank/index?brandiphonecountrycngenre36在 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_idrec.get(app_id), app_namerec.get(app_name), companyrec.get(company), bundle_idrec.get(bundle_id), rankrec.get(rank), genrerec.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_id、rank做int()转换。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, encodingutf-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 itemopen_spider在爬虫启动时执行close_spider在结束时执行。这样保证 CSV 文件只打开一次不会每条数据都重新开文件。dict(item)可以把 Item 转成普通字典csv.DictWriter按fieldnames顺序写入。MySQL 版 pipeline 稍微复杂一点但核心就两条插入和提交。class MysqlPipeline: def open_spider(self, spider): self.conn pymysql.connect( host127.0.0.1, userroot, password123456, databaseqimai, charsetutf8mb4, ) 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, dateNone, genreNone, *args, **kwargs): super().__init__(*args, **kwargs) self.date date or latest self.genre genre or 36启动时用-a传参scrapy crawl qimai -a date2025-06-05 -a genre6014 -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 21注意 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)能在第一时间区分是账号失效还是频率受限。本文还有配套的精品资源点击获取
