5. 短视频App数据抓取

5-1 抖音短视频抓取需求分析

1.需求分析

1.1 模拟滑动视频和发布者点击发布者

1.2 通过mitmproxy抓取并解析数据

import uiautomator2 as u2
import time

class Douyin(object):
    def __init__(self, serial="192.168.0.101"):
        # 连接
        self.d = u2.connect(serial)
        # 调用方法
        self.start_app()        # 启动
        self.handle_watcher()  # 监控器
        self.size = self.get_windowsize()
        # 获取初始时间
        self.t0 = time.perf_counter()

    def start_app(self):
        self.d.app_start(package_name="com.ss.android.ugc.aweme")

    def stop_app(self):
        """退出逻辑"""
        self.d.watcher.stop()
        self.d.app_stop("com.ss.android.ugc.aweme")
        self.d.app_clear("com.ss.android.ugc.aweme")



    def stop_time(self):
        if time.perf_counter() - self.t0 > 20:  # s
            return True

    def handle_watcher(self):
        # 通知权限
        self.d.watcher.when('//*[@resource-id="com.ss.android.ugc.aweme:id/a4r"]').click()
        # 发现滑动查看更多
        self.d.watcher.when('//*[@text="滑动查看更多"]').click()
        self.d.watcher.when('//*[@text="快速进入TA的个人中心"]').click()
        self.d.watcher.start(interval=1)

    def get_windowsize(self):
        return self.d.window_size()


    def swipe_douyin(self):
        '''滑动抖音短视频 点击视频发布者头像的操作'''
        # 判断是否正常进入抖音界面
        if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists(timeout=20):
            while True:
                if self.stop_time():
                    self.stop_app()
                    return

                # 查看 是不是正常的发布者（不是广告）
                if self.d(resourceId="com.ss.android.ugc.aweme:id/u0").exists:
                    # 是不是正常的 发布者，点击头像
                    self.d(resourceId="com.ss.android.ugc.aweme:id/tw").click()
                    # 返回
                    self.d(resourceId="com.ss.android.ugc.aweme:id/et").click()

                # 判断是否正常进入抖音界面
                if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists:
                    # 滑动： 由下向上滑动；
                    x1 = int(self.size[0]*0.5)
                    y1 = int(self.size[1]*0.9)
                    y2 = int(self.size[1]*0.15)
                    self.d.swipe(x1, y1, x1, y2)

if __name__ == '__main__':
    d = Douyin()
    d.swipe_douyin()

5-6 通过mitmproxy解析短视频App返回数据-接口分析

先保证WIN和手机IP地址的一致性

确保 手机上安装fiddle证书

先点掉 Capturing, 抓取的数据包就是连到手机上的数据包；

有两个JS文件，先看第一个

特别注意

在新版本的抖音里，我们是找到的是加密的数据返回；低版本在豌豆荚中可以下载，得到没有加密的数据；

# 个人信息页接口
# https://aweme-eagle.snssdk.com/aweme/v1/user/?user_id

# 滑动视频接口
# https://aweme-eagle.snssdk.com/aweme/v1/feed

1. 个人信息页接口

第一个JS文件

# json格式文件

"wx_nickname_replace":"开心锤锤♪","replaced":false},
"ky_only_predict":0,"school_auth":0,"is_star":false,"general_permission":{"following_follower_list_toast":1},
"share_weibo_desc":"在抖音，记录美好生活！","share_desc":"在抖音，记录美好生活！",
"share_title":"快来加入抖音，让你发现最有趣的我！","share_qrcode_url":{"uri":"671f001b2afba0ae78dc","url_list":["https://p9-dy.byteimg.com/obj/671f001b2afba0ae78dc","https://p6-dy-ipv6.byteimg.com/obj/671f001b2afba0ae78dc","https://p3-dy-ipv6.byteimg.com/obj/671f001b2afba0ae78dc"]}},"activity":{"use_music_count":0,"digg_count":0},"follower_status":0,"duet_setting":0,"unique_id":"2B2B2","geofencing":

2. 滑动视频接口

点击 滑动按钮,重新抓取数据；

5-7 通过mitmproxy解析短视频App返回数据-编写mitmdump解析文件

确保安装好了手机mitmproxy 以及证书（IP变化就要重新安装）

# decode_douyin.py
import json

def response(flow):
    """解析10版本抖音app返回数据"""
    # 视频
    if 'https://aweme-eagle.snssdk.com/aweme/v1/feed' in flow.request.url:
        # 使用json来loads response.text
        video_response = json.loads(flow.response.text)
        video_list = video_response.get("aweme_list", [])
        for item in video_list:
            print(item.get("desc"), "")   # 视频标题

    # 发布者页面
    if 'https://aweme-eagle.snssdk.com/aweme/v1/user/?user_id' in flow.request.url:
        person_response = json.loads(flow.response.text)
        person_info = person_response.get("user", "")
        if person_info:
            info = {
                'nickname': person_info.get("nickname", ""),   # 名称
                'total_favorited': person_info.get("total_favorited", 0),  # 点赞数
                'following_count': person_info.get("following_count", 0),  # 关注着数量
                'douyin_id': person_info.get("unique_id", ""),               # 抖音ID
                'folllower_count': person_info.get("follower_count", 0)    # 粉丝数量
            }
            print('--------------------------------------------------------------------')
            print(info)

启动 mitmdump

(venv) user1@imooc:~/u2_project/douyin$ mitmdump -s decode_douyin.py -p 8889
Loading script decode_douyin.py
Proxy server listening at http://*:8889

# run handle_douyin.py
import uiautomator2 as u2
import time

class Douyin(object):
    def __init__(self, serial="192.168.0.101"):
        # 连接
        self.d = u2.connect(serial)
        # 调用方法
        self.start_app()        # 启动
        self.handle_watcher()  # 监控器
        self.size = self.get_windowsize()
        # 获取初始时间
        self.t0 = time.perf_counter()

    def start_app(self):
        self.d.app_start(package_name="com.ss.android.ugc.aweme")

    def stop_app(self):
        """退出逻辑"""
        self.d.watcher.stop()
        self.d.app_stop("com.ss.android.ugc.aweme")
        self.d.app_clear("com.ss.android.ugc.aweme")

    def stop_time(self):
        if time.perf_counter() - self.t0 > 20:  # s
            return True

    def handle_watcher(self):
        # 通知权限
        self.d.watcher.when('//*[@resource-id="com.ss.android.ugc.aweme:id/a4r"]').click()
        # 发现滑动查看更多
        self.d.watcher.when('//*[@text="滑动查看更多"]').click()
        self.d.watcher.when('//*[@text="快速进入TA的个人中心"]').click()
        self.d.watcher.start(interval=1)

    def get_windowsize(self):
        return self.d.window_size()


    def swipe_douyin(self):
        '''滑动抖音短视频 点击视频发布者头像的操作'''
        # 判断是否正常进入抖音界面
        if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists(timeout=20):
            while True:
                if self.stop_time():
                    self.stop_app()
                    return

                # 查看 是不是正常的发布者（不是广告）
                if self.d(resourceId="com.ss.android.ugc.aweme:id/u0").exists:
                    # 是不是正常的 发布者，点击头像
                    self.d(resourceId="com.ss.android.ugc.aweme:id/tw").click()
                    # 返回
                    self.d(resourceId="com.ss.android.ugc.aweme:id/et").click()

                # 判断是否正常进入抖音界面
                if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists:
                    # 滑动： 由下向上滑动；
                    x1 = int(self.size[0]*0.5)
                    y1 = int(self.size[1]*0.9)
                    y2 = int(self.size[1]*0.15)
                    self.d.swipe(x1, y1, x1, y2)

if __name__ == '__main__':
    d = Douyin()
    d.swipe_douyin()


--------------------------------------------------------------------
{'nickname': '七猫探长', 'total_favorited': 4012853, 'following_count': 89, 'douyin_id': 'qimao2021', 'folllower_count': 550240}

实战2：小红书

1.抓取需求分析

抓取 推荐也文章基本信息（文章标题、类型、基本描述、用户信息等）

文章详细内容

文章评论

2. 滑动视频接口

import uiautomator2 as u2
import time, datetime
import adbutils, multiprocessing
from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(hosts="192.168.0.101:9200")
class Xiaohongshu(object):
    def __init__(self, serial="192.168.0.103:5555"):
        '''
        手机IP:192.168.0.103
        Linux系统IP ： 192.168.0.101
        desc: V1:刷新新的内容， V2：下一版本功能：爬取文章详细内容和评论数
        :param serial:
        '''
        self.d = u2.connect(serial)
        self.size = self.get_window_size()
        self.start_app()


    def start_app(self, name='com.xingin.xhs'):
        self.d.app_start(package_name=name)

    def get_window_size(self):
        return self.d.window_size()

    def swipe_screen(self):
        x1 = int(self.size[0] * 0.5)
        y1 = int(self.size[1] * 0.9)
        y2 = int(self.size[1] * 0.25)
        self.d.swipe(x1, y2, x1, y1)


if __name__ == '__main__':
    d = Xiaohongshu()
    time.sleep(3)
    flag = True
    while flag:
        d.swipe_screen()
        # time.sleep(1)
    print(es.search(index="xiaohongshu_2021-03-17"))

3.mitmproxy解析短视频App返回数据

import json
import time
import datetime

from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(hosts="192.168.0.101:9200")

def handle_date(key):
    """处理时间方法"""
    date_touple = time.localtime()
    date = None
    if key == "hms":
        date = time.strftime("%Y-%m-%d %H:%M:%S", date_touple)
        date = date.replace(' ', 'T')
    elif key == "bir":
        date = time.strftime("%Y-%m-%d", date_touple)
    return date


def response(flow):
    '''
    解析 小红书 最新版本 app 的 目录页数据；
    每次滑动拿到 10条数据；
    '''

    if 'https://edith.xiaohongshu.com/api/sns/v6/homefeed' in flow.request.url:
        article_response = json.loads(flow.response.text)

        article_list = article_response.get('data', [])
        data_list = []
        for item in article_list:
            info ={
                "_index": "xiaohongshu_%s" % handle_date(key="bir"),
                "_source": {
                    "@timestamp": handle_date(key="hms"),
                    "birthday": handle_date(key="bir"),
                    "title": item.get("display_title", ""),
                    "name": item.get("user", "").get("nickname", ""),
                    "userid": item.get("user", "").get("userid", ""),
                    "type": item.get("type", ""),
                    "display_image":item.get("images_list", "").get("url_size_large", "")
                    "desc": item.get("desc", ""),
                    "likes": item.get("likes", "")
                }
            }
            print(info)
            data_list.append(info)
        # 通过helpers插入数据
        helpers.bulk(es, data_list)

5. 查看10分钟爬取效果

3284/10m

get xiaohongshu_2021-03-17/_search
#---------------------------------------------------------
{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 3284,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "xiaohongshu_2021-03-17",
        "_type" : "_doc",
        "_id" : "4d3sPXgBy2dlyd-YuFIb",
        "_score" : 1.0,
        "_source" : {
          "@timestamp" : "2021-03-17T10:03:47",
          "birthday" : "2021-03-17",
          "title" : "考研复试简历模板大放送，请查收㊙️",
          "name" : "研学长",
          "userid" : "5d8f85a600000000010036ac",
          "type" : "normal",
          "desc" : """🌈哈喽，各位21/22考研的小伙伴们
❤️在考研复试中，我们为了让导师更加快速全面的了解我们，通常都需要准备一份精美的简历进行自我介绍，一份好的模板通常能让导师对你有更好的印象与了解！也是你个人形象的""",
          "likes" : 5813
        }
      },

注意
1.解决kibana 超过10000条数据无法查询的问题

get xiaohongshu_2021-03-17/_settings
{
  "max_result_window" : 200000000
}