Python 爬虫爬取应用商店数据:请求构造与数据解析

简介: Python 爬虫爬取应用商店数据:请求构造与数据解析

应用商店(如 Google Play、App Store、豌豆荚)的应用数据是互联网竞品分析、行业市场研究、ASO 优化的核心数据源。应用商店爬虫的核心爬取难点主要集中在两大模块:一是请求构造,多数平台存在接口参数加密、请求签名校验等反爬限制;二是数据解析,接口普遍存在 JSON 层级嵌套深、不同接口字段命名不规范、数据格式杂乱等问题。本文以应用搜索列表、应用详情页为采集主线,完整讲解抓包分析、合规请求构造、结构化数据解析、代理反爬优化全流程落地方案,并提供可直接复用的 Python 源码。
一、应用商店爬取的技术难点
主流应用商店均具备成熟的反爬风控体系,结合Web端接口特征,整理核心难点如下:
难点
详细说明
接口参数加密
部分应用商店接口增加签名校验、动态加密参数,无合法加密参数直接请求会返回 403 权限拒绝
多层反爬机制
包含请求频率限制、UA 请求头校验、IP 封禁、滑块验证码、会话校验等风控策略
非规范数据格式
接口返回 JSON 通常嵌套 3-5 层结构,不同分类、不同版本接口字段命名不统一,适配难度高
页面动态加载
搜索结果分页异步加载、评论数据懒加载,常规静态请求无法获取完整数据
本文实现方案:以国内主流应用商店(豌豆荚、应用宝等)Web 端公开接口为采集对象,基于 Requests 库直接请求结构化 JSON 接口,跳过 JS 渲染流程,降低爬虫开发难度,提升数据采集稳定性。
二、接口分析:抓包定位数据源
应用商店优先采集 XHR 异步接口数据,相较于 HTML 页面解析,JSON 结构化接口数据更规整、提取成本更低。抓包流程采用浏览器开发者工具完成,操作步骤如下:
打开浏览器开发者工具(F12)→ 切换至 Network 面板 → 筛选 XHR 异步请求 → 在应用商店执行搜索操作,筛选出返回 JSON 格式的核心业务接口。抓包过程重点观察以下要素:
请求 URL:区分搜索接口、详情接口、评论接口的 URL 规则,归纳接口通用模板;
请求方法:判定接口为 GET 静态请求或 POST 加密请求;
请求参数:区分固定静态参数、时间戳、签名等动态可变参数;
请求头:识别 Token、签名、校验字段等必备请求头部参数。
本文列举典型应用商店搜索接口示例:
接口标准返回 JSON 结构如下:
结构化 JSON 接口无需编写复杂网页选择器,可直接通过字典路径精准提取字段,大幅简化数据解析逻辑。
三、完整代码实现
3.1 基础配置与代理通用配置
为规避单IP高频请求导致的封禁问题,本项目接入通用动态代理IP,通过代理池轮换请求出口IP,降低风控拦截概率。删除商业化代理冗余介绍,采用通用代理配置格式,适配各类隧道代理服务。
```import requests
import json
import time
import random
import re
import os
import pandas as pd
from urllib.parse import quote

==================== 基础配置区 ====================

应用商店 API 基础地址(替换为目标应用商店的实际接口)

BASE_URL = "https://appstore.example.com/api"

通用标准化请求头,模拟真实浏览器访问

HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
'Accept': 'application/json, text/plain, /',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Referer': 'https://appstore.example.com/',
}

通用代理IP配置(动态隧道代理,可替换为任意代理服务商参数)

PROXY_CONFIG = {
"host": "代理地址",
"port": "代理端口",
"user": "代理账号",
"pass": "代理密码"
}

组装代理请求格式

proxy_meta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % PROXY_CONFIG
PROXIES = {"http": proxy_meta, "https": proxy_meta}

爬虫业务配置

MAX_PAGE = 5 # 每个关键词爬取的最大页数
PAGE_SIZE = 20 # 每页数据量
SAVE_DIR = "app_data"

================================================


代理技术说明:本项目采用亿牛云隧道动态代理模式,依托代理池实现请求IP自动轮换,适配无状态的Requests接口请求;支持高并发采集,毫秒级切换出口IP,有效规避IP封禁、频率限制等反爬拦截。
3.2 请求构造:搜索应用列表
基于GET请求构造搜索接口请求,加入随机隧道参数、请求休眠策略,优化反爬效果;同时增加异常捕获,适配429限流、403拒绝等异常状态码。
```def search_apps(keyword, max_pages=5, use_proxy=True):
    """搜索应用商店,提取应用列表"""
    all_apps = []

    for page in range(1, max_pages + 1):
        # 构造请求参数
        params = {
            'keyword': keyword,
            'page': page,
            'size': PAGE_SIZE,
        }

        # 随机隧道标识,实现请求IP差异化
        req_headers = HEADERS.copy()
        if use_proxy:
            tunnel = random.randint(1, 10000)
            req_headers['Proxy-Tunnel'] = str(tunnel)

        try:
            resp = requests.get(
                f"{BASE_URL}/search",
                params=params,
                headers=req_headers,
                proxies=PROXIES if use_proxy else None,
                timeout=15
            )

            if resp.status_code == 429:
                print(f"  [{keyword}] 第{page}页请求过快 (429)")
                time.sleep(3)
                continue

            if resp.status_code == 403:
                print(f"  [{keyword}] 第{page}页访问拒绝 (403)")
                continue

            resp.raise_for_status()
            data = resp.json()

            # 解析应用列表
            app_list = data.get('data', {}).get('list', [])

            if not app_list:
                print(f"  [{keyword}] 第{page}页无数据,停止翻页")
                break

            for app in app_list:
                all_apps.append(parse_app_item(app, keyword))

            print(f"  [{keyword}] 第{page}页: {len(app_list)} 个应用")

        except requests.exceptions.JSONDecodeError:
            print(f"  [{keyword}] 第{page}页返回非 JSON")
            # 尝试 HTML 解析作为降级方案
            html_apps = parse_html_results(resp.text, keyword)
            all_apps.extend(html_apps)
        except Exception as e:
            print(f"  [{keyword}] 第{page}页异常: {e}")

        time.sleep(random.uniform(1, 3))

    print(f"[{keyword}] 搜索完成,共 {len(all_apps)} 个应用")
    return all_apps

请求构造核心要点:
搜索接口采用 GET 请求,核心入参包含关键词、页码、分页数量;
生成随机隧道参数,配合代理池实现每次请求出口IP不同;
增设异常降级策略,接口返回非JSON格式时,自动切换HTML解析方案。
3.3 数据解析:JSON 结构化提取
针对应用商店字段杂乱、格式不统一的问题,封装通用解析函数,做字段兼容、数据标准化清洗,统一输出规范格式数据。
```def parse_app_item(app_data, keyword):
"""解析单个应用数据,统一字段格式"""
return {
'keyword': keyword,
'app_id': app_data.get('appId', '') or app_data.get('package_name', ''),
'app_name': app_data.get('appName', '') or app_data.get('title', ''),
'developer': app_data.get('developer', '') or app_data.get('author', ''),
'category': app_data.get('category', ''),
'version': app_data.get('version', ''),
'file_size': normalize_size(app_data.get('fileSize', '')),
'score': safe_float(app_data.get('score', 0)),
'download_count': normalize_download_count(app_data.get('downloadCount', '0')),
'description': clean_text(app_data.get('description', '')),
'icon_url': app_data.get('iconUrl', '') or app_data.get('icon', ''),
'download_url': app_data.get('downloadUrl', '') or app_data.get('apk_url', ''),
'update_time': app_data.get('updateTime', '') or app_data.get('update_date', ''),
}

--- 通用数据清洗工具函数 ---

def normalize_size(size_str):
"""统一文件大小格式:'258MB' → 258"""
if not size_str:
return 0
match = re.search(r'([\d.]+)', str(size_str))
return float(match.group(1)) if match else 0

def normalize_download_count(count_str):
"""统一下载量:'100亿次+' → 10000000000"""
if not count_str:
return 0
count_str = str(count_str).replace(',', '').replace(' ', '')
if '亿' in count_str:
num = safe_float(count_str.replace('亿', '').replace('次', '').replace('+', ''))
return int(num 100000000)
elif '万' in count_str:
num = safe_float(count_str.replace('万', '').replace('次', '').replace('+', ''))
return int(num
10000)
else:
return int(safe_float(count_str))

def safe_float(val, default=0.0):
"""安全转换为浮点数"""
try:
return float(val)
except (ValueError, TypeError):
return default

def clean_text(text):
"""清洗文本:去除 HTML 标签、多余空白"""
if not text:
return ''
text = re.sub(r'<[^>]+>', '', str(text))
text = re.sub(r'\s+', ' ', text).strip()
return text


数据解析优化方案:
数据问题
处理方式
字段名不统一
多字段兼容匹配,适配不同接口命名规则
下载量格式混乱
将万、亿等中文单位统一换算为纯数字
描述含HTML标签
正则表达式清洗冗余标签、空白字符
空值异常
设置默认值,避免程序报错中断
3.4 HTML 降级解析
部分接口风控拦截后会返回静态HTML页面,此时采用BeautifulSoup做降级解析,通过通用CSS选择器提取页面应用数据,保障爬虫稳定性。
```def parse_html_results(html, keyword):
    """HTML 降级解析:从页面中提取应用列表"""
    from bs4 import BeautifulSoup

    soup = BeautifulSoup(html, 'html.parser')
    results = []

    # 通用应用卡片容器选择器,适配多数应用商店页面结构
    cards = soup.select('.app-card, .app-item, [class*="appItem"]')

    for card in cards:
        try:
            app = {
                'keyword': keyword,
                'app_id': '',
                'app_name': '',
                'developer': '',
                'category': '',
                'version': '',
                'file_size': 0,
                'score': 0,
                'download_count': 0,
                'description': '',
                'icon_url': '',
                'download_url': '',
                'update_time': '',
            }

            # 提取应用名称
            name_elem = card.select_one('.app-name, .title, h2, h3')
            if name_elem:
                app['app_name'] = name_elem.get_text(strip=True)

            # 提取应用ID
            link = card.select_one('a[href*="detail"], a[href*="app"]')
            if link:
                href = link.get('href', '')
                app_id_match = re.search(r'([a-z]+\.[a-z]+\.[\w.]+)', href)
                if app_id_match:
                    app['app_id'] = app_id_match.group(1)
                app['download_url'] = href

            # 提取图标、评分、下载量
            icon = card.select_one('img')
            if icon:
                app['icon_url'] = icon.get('src', '')

            score_elem = card.select_one('.score, .rating, [class*="star"]')
            if score_elem:
                app['score'] = safe_float(score_elem.get_text(strip=True))

            download_elem = card.select_one('.download-count, .downloads')
            if download_elem:
                app['download_count'] = normalize_download_count(
                    download_elem.get_text(strip=True)
                )

            if app['app_name']:
                results.append(app)

        except Exception:
            continue

    return results

3.5 应用详情页爬取
基于应用唯一ID,请求详情接口,采集应用介绍、权限、更新日志等深度数据,完善数据维度。
3.6 批量采集与持久化存储
实现多关键词批量采集,完成数据去重、格式清洗,最终导出Excel、JSON两种通用格式文件,便于数据分析与二次使用。
```def crawl_keywords(keywords, max_pages=5, use_proxy=True):
"""批量采集多个关键词的应用数据"""
all_apps = []

for keyword in keywords:
    print(f"\n{'='*40}")
    print(f"关键词: {keyword}")
    apps = search_apps(keyword, max_pages, use_proxy)
    all_apps.extend(apps)

# 根据app_id完成数据去重
df = pd.DataFrame(all_apps)
if not df.empty and 'app_id' in df.columns:
    before = len(df)
    df = df.drop_duplicates(subset=['app_id'], keep='first')
    print(f"\n去重: {before} → {len(df)}")

# 创建保存目录
os.makedirs(SAVE_DIR, exist_ok=True)

# 导出Excel、JSON文件
excel_path = os.path.join(SAVE_DIR, "app_store_data.xlsx")
df.to_excel(excel_path, index=False)
print(f"Excel 已保存: {excel_path}")

json_path = os.path.join(SAVE_DIR, "app_store_data.json")
df.to_json(json_path, orient='records', force_ascii=False, indent=2)
print(f"JSON 已保存: {json_path}")

return df

def main():

# 自定义采集关键词
keywords = ["社交", "短视频", "电商", "游戏"]

df = crawl_keywords(
    keywords=keywords,
    max_pages=MAX_PAGE,
    use_proxy=True
)

# 基础数据统计
print(f"\n{'='*40}")
print(f"采集完成")
print(f"应用总数: {len(df)}")
if 'score' in df.columns:
    print(f"平均评分: {df['score'].mean():.1f}")
if 'keyword' in df.columns:
    print(f"各关键词数量:")
    print(df['keyword'].value_counts().to_string())

if name == 'main':
main()
```

四、常见异常错误处理方案
错误码/异常
报错原因
解决方案
403
接口签名缺失、权限校验失败
重新抓包补充加密请求头、签名参数
407
代理身份认证失败
核对代理账号、密码、端口配置
429
请求频率过高触发限流
增加随机休眠、降低单线程采集频率
JSON解析失败
风控拦截返回HTML页面
自动降级为网页静态解析
字段缺失
不同接口字段命名差异化
多候选字段兼容匹配,设置空值默认值
五、进阶优化方向与适用场景
5.1 技术优化方向
接口逆向加固:海外应用商店存在高强度签名加密,可通过Frida、Xposed客户端Hook技术或APK逆向工程,解析加密签名算法;
选择器维护:HTML降级解析选择器易随页面改版失效,可配置多组备用选择器,按优先级自适应匹配;
增量采集:本地存储已采集应用ID,二次运行时跳过重复数据,仅采集新增应用,提升采集效率。
5.2 业务适用场景
竞品分析:批量采集同类产品评分、下载量、更新节奏,分析竞品运营策略;
行业调研:按应用分类采集行业数据,统计应用分布、热度、用户评分;
ASO优化:监控关键词搜索排名,优化应用商店上架权重。
总结
应用商店数据爬虫的核心技术逻辑为:抓包逆向定位接口、标准化构造请求、结构化清洗解析、风控策略优化。相较于传统网页爬虫,JSON接口采集方式代码简洁、稳定性更强。针对平台反爬机制,可通过动态代理池、随机请求间隔、请求头伪装等方式降低拦截概率。本文提供的通用代码无需复杂改造,替换目标接口地址与代理配置即可快速落地,适用于中小型数据采集、行业分析、技术学习等场景。同时采集过程需遵守网站robots协议,合法合规采集公开业务数据。

相关文章
|
9月前
|
数据采集 存储 Web App开发
处理Cookie和Session:让Python爬虫保持连贯的"身份"
处理Cookie和Session:让Python爬虫保持连贯的"身份"
|
4月前
|
数据采集 JSON API
Python 进阶爬虫:解析知识星球 API
Python 进阶爬虫:解析知识星球 API
|
5月前
|
数据采集 JSON Java
Java 异步爬虫高效获取小红书短视频内容
Java 异步爬虫高效获取小红书短视频内容
|
2月前
|
数据采集 存储 Web App开发
基于 Selenium 的美团外卖动态数据爬虫实现方案
基于 Selenium 的美团外卖动态数据爬虫实现方案
|
2月前
|
数据采集 存储 数据安全/隐私保护
Python 爬取图片攻略:告别水印,批量保存高清图片
Python 爬取图片攻略:告别水印,批量保存高清图片
|
1月前
|
数据采集 JSON 数据挖掘
抖音搜索页数据批量爬取,多关键词同步采集实现
抖音搜索页数据批量爬取,多关键词同步采集实现
|
2月前
|
数据采集 存储 数据可视化
Python 爬虫:拍卖网站列表页与详情页数据联动爬取
Python 爬虫:拍卖网站列表页与详情页数据联动爬取
|
3月前
|
数据采集 搜索推荐 应用服务中间件
如何判断网站流量飙升是搜索引擎爬虫导致的?
如何判断网站流量飙升是搜索引擎爬虫导致的?
|
3月前
|
数据采集 存储 前端开发
Python 爬虫实战:批量抓取应用商店分类应用
Python 爬虫实战:批量抓取应用商店分类应用
|
3月前
|
数据采集 Web App开发 数据安全/隐私保护
对比分析:Python爬虫模拟登录的3种主流实现方式
对比分析:Python爬虫模拟登录的3种主流实现方式