1. 爬取目标

数据源：苹果商店对应 APP 的评论区
目标内容：对应 APP 的评论信息

2. 思路

苹果商店各平台看接口调用，找到：

应用检索接口
评论获取接口

撰写爬虫，清洗数据
保存数据

3. 具体实现

ios 设备抓包和 web 抓包都没有找到能获取大量评论的接口，但搜索时，发现有用户在 itunes 的接口上能实现上述功能，所以决定使用 itunes 的接口实现

具体代码：

from concurrent.futures import ThreadPoolExecutor

import requests
import pandas as pd


def search_appid(kw: str):
    # 查询 appid 和 appname
    url = f"http://itunes.apple.com/search?term={kw}&entity=software"
    res = requests.get(url)
    if res.status_code == 200:
        data = res.json()
        results = data['results']
        search_res = []
        for item in results:
            search_res.append({
                "appname": item['trackName'],
                "appid": item['trackId'],
            })
        print(search_res)
        return search_res


def get_comments(appid: str | int):
    # 根据 appid 抓取数据
    client = requests.session()
    f_res = []
    for i in range(1, 11):
        url = f"https://itunes.apple.com/rss/customerreviews/page={i}/id={appid}/sortby=mostrecent/json?l=en&&cc=cn"
        res = client.get(url)
        if res.status_code == 200:
            print(res.json())
            data = res.json()["feed"]["entry"]
            for item in data:
                t = {
                    'name': item['author']['name']['label'],
                    'rate': item['im:rating']['label'],
                    'user_id': item['id']['label'],
                    'content': item['content']['label'],
                }
                f_res.append(t)
    return f_res


def save_comments(data: list, appname: str):
    # 使用 pandas 便捷保存数据
    df = pd.DataFrame(data)
    df.to_csv(f'{appname}.csv', index=False, encoding='utf-8-sig')
    print(f'保存数据成功：{appname}.csv')


def get_comments_and_save(appname_and_appid: dict):
    # 封装工作集
    data = get_comments(appname_and_appid['appid'])
    save_comments(data, appname_and_appid['appname'])


def main():
    # 启动函数，使用多线程，根据关键词爬取所有结果的评论
    task_list = search_appid('分期乐')
    with ThreadPoolExecutor() as executor:
        executor.map(get_comments_and_save, task_list)


if __name__ == '__main__':
    main()