想你所想,为你所爱

之前出于兴趣和学习的目的尝试了一下抖音APP的数据抓包，实现了我要的效果之后把总结发到了博客。嘎嘎没想到有这么多同学感兴趣，也难怪谁让抖音现在这么火呢。上次发了段小代码，说什么给吴亦凡的粉丝颜值打分？现在看看当时的代码真是失了智。虽然之前的代码比较蠢，但是研究了那么久我还是有点心得体会的。这里分享一下希望大家研究抖音爬虫的时候少踩坑。

其实高手看这个真的是一点难度都没有，最苦逼的还是我们这些小白，要一点一点摸索。其实我倒是喜欢这种慢慢摸索的学习过程，这样的学习方式对个人能力提升真的很大，我深有体会。但是自学的前提是网上有足够的资料可以查阅，不然确实无从下手。下面我将从搭建环境开始讲下我学习过程中遇到的坑，希望可以帮到大家。

环境搭建

由于抖音数据包的URL有很多加密字段，我们不可能通过构造URL的方式实现数据爬取。用到的方法是通过截取抖音的数据包实现的简单爬虫，效率不是很高但也不失为一种解决问题的方法。通过抓包APP实现的爬虫需要控制手机行为使得APP发送相应的数据包，然后通过中间人代理获取相应的数据。简单来说Appium和Mitmproxy实现的就是requests模块实现的功能。

Appium的安装

Appium的安装依赖三个环境：Node.js、Java、Android-SDK。环境的安装之前有做介绍可以翻翻前面的文章再看一下。这里有几点注意事项要跟大家说一下；

①.adb连接不上设备的话一定要检测是否有同类型的软件冲突，比如说手机助手之类的东西。如果有冲突的话用Android-SDK里的adb.exe替换掉冲突路径内的adb.exe。可以通过shell命令查看adb的端口占用，具体方法自行搜索引擎。

②.Appium安装可以通过npm直接安装，这种形式的安装没有GUI界面，程序启动需要在Shell控制台用指令启动。但是我们一般安装的都是Appium-Desktop，Desktop不支持控制台启动，这个可以通过【原创】appium-desktop版本配置命令行运行服务（windows）配置控制台运行服务。推荐使用第二种方式配置，这样升级Appium之后Shell版本可以同步更新。

Appium-doctor可以检测Appium服务是否安装成功，但是需要另外安装，通过npm install appium-doctor安装。如果命令无效检查环境变量的配置。有些warning是可以无视的，比如提示没有安装ffmpeg都可以忽略，不影响后期的操作。附上几条常用的adb指令。

获取包名appPackage和appActivity：adb shell dumpsys window w | findstr \/ | findstr name=

获取所有设备: adb devices
获取手机系统信息:adb shell "cat /system/build.prop | grep "product""
获取手机系统版本:adb shell getprop ro.build.version.release
获取手机系统api版本:adb shell getprop ro.build.version.sdk
获取手机设备型号:adb -d shell getprop ro.product.model
获取手机厂商名称:adb -d shell getprop ro.product.brand

多设备的话在在adb后添加参数 -s devicesname

Mimtproxy的安装

①.安装Mitmproxy建议直接安装到系统的环境内，因为Mitmproxy作为抓包工具的话个人感觉比Fiddler要简单很多，装在系统环境内虚拟环境同样可以调用，比较方便。

②.启动MimtProxy，由于环境是Windows环境所以mitmproxy命令无法使用。一般用mitmweb做调试，用mitmdump做代码分析，控制台输出数据可以通过添加参数mitmdump -q来屏蔽控制台输出。

③.服务启动后手机要与mitmproxy的服务器处在同一局域网，并且手机端要添加手动代理链接到mitmproxy服务器所在的本地IP地址。配置成功之后手机会有数据包被电脑端捕获。

④.手机端需要安装证书，否则进入抖音会提示没有网络连接，这个很重要。进入mitm.it安装对应安卓版本的证书即可。部份安卓手机无法直接打开下载的证数文件，解决办法是系统设置→其他设置→安全与隐私设置→从本机存储安装证书。不同设备可能有所差别仔细找找就行了，大同小异。

爬虫代码注意事项

真机的爬虫主要包含两部分，一个是操作手机部份，一个是数据处理部份。因为真机抓包的环境的十分不稳定，或许是数据线连接不稳定的原因，或许是Appium的原因。总之我们的脚本执行动作要行云流水，不得掺杂半点错误。这就要求我们做好异常处理，简单点讲就是try-catch。说起来简单但是实际应用起来却有他的巧妙之处。做过移动端自动化测试的同学应该轻车熟路，我这边分享一些我的笨方法。

因为Appium的操作几乎是盲点，做页面位置判断代码量有太大。我的解决思路是定义一个方法，这个方法的作用就是回到脚本起点。当页面出现异常的时候，例如：由于手机卡顿 ，脚本代码已经走到下个页面的执行位置，但是真是页面还没有跳转，代码由于找不到写死的element必然会抛出异常。我们通过捕获这个异常调用定义好的方法让页面返回到开始页面从新执行。简单来说就是工作流写死一气呵成，但是工作流抛出异常就回到起点从新开始。不管怎么处理一定要保证所有异常都能正常捕获，如果有没有捕获的异常会终止进程，Appium启动一次session的开销成本太大了，而且爬虫终止非常影响我们数据的采集。

Appium-定位

脚本的执行过程中，总是需要点击某个指定的按钮，向某个指定的输入框输入消息，这些按钮之类的在Appium中称作element。关于定位element框架提供了很多种方法供我们选择。安卓页面的布局和html大同小异，都是用许多不同的标记堆叠起来的。所以定位element对于一名合格的爬虫工程师来说都不是什么大问题，毕竟爬虫工程师不是还被人叫做xpath标注员🤣🤣🤣。最初级的写法就是通过resource-id定位元素，这个类似与html中的id选择器，速度快，方便快速，但是对于移动应用不推荐这种写法。因为app的resource-id会经常变动，app不更新的前提下resource-id都会经常更改。相信这是应用的一种安全机制，也是为了防止有人恶意使用脚本做一些灰色产业嘎嘎嘎。这也是我踩着坑总结出来的，昨天写的代码今天就不行了，电脑都砸了的冲动。解决方式所有的定位都不要用绝对定位，要用xpath写相对定位，通过固定的text等信息偏移过去。因为无论resource-id怎么变它的相对位置都不会改变。

同样的我们滑动操作也是必不可少，因为我们要不断地向下滑动获取用户信息。对于滑动的话appium提供了两个滑动方法，一个是swipe，这个主要针对的是慢动作的滑动，可以添加延迟参数。另外一个就是flick，翻译过来就是“抽动，弹”的意思，主要用在快速滑动。像我们要不停的下来获取新的数据的话用flick再好不过了，比起swipe来效率翻番。

Mitmproxy抓包

对于Mitmproxy的开发来说，其实我并没有很多经验。所用到的都是最基础最简单的调用方法，但是Mitmproxy的强大并不仅限于此。官方的examples里面有很多进阶的使用例子，比如通过中间人给数据包加上代理，通过中间人做相应规则的篡改和跳转。有精力的话可以深挖一下肯定有收获。

遇到的最多的一个问题就是如何在Pycharm中调用mitmproxy服务，这一点也是为难了我好久哦。启动mitmproxy服务我们可以使用os.system()命令。由于我们要先启动中间人服务在执行脚本，如果我们把两部分都放在主线程中的话mitmproxy就会阻塞线程，我们下面的脚本代码就无法执行。所以为了解决这个问题需要把mitmproxy的启动放在子线程执行。

关于抖音

抖音为了防止灰产行业现在的风控可以说是十分严格了。但是这也仅限于注册刷关注之类的，与我们的爬虫关系不是很大。只不过最新版本的抖音已经无法直接获取到视频的直链了，所以手头上存一个稳定的旧版app还是十分必要的。

Ⅰ.抖音的粉丝数默认只展示5000个，超出部份则不展示这个没办法破解。好的方法就是每天获取一次，每次都会有许多新数据出来，做个去重就好了。Ⅱ.经常刷的话有出验证码的可能性，但是概率极低，保证自己手机环境的真实性，账号是老号更佳。Ⅲ.安卓有Url Scheme的机制，有了这个可以很方便快捷的跳转到app指定的页面。但是这个一般是隐藏的，不是很好找到。分享一段通过Url Scheme直接跳转到用户页的Appium代码供大家参考一下。

def jump_to_user(device: WebDriver, uid: str):
    url = "snssdk1128://user/profile/{}?refer=web&gd_label=click_wap_profile_bottom&type=need_follow&needlaunchlog=1"
    device.get(url.format(uid))

手机卡顿十分影响脚本的运行，模拟器什么的都太卡了，还是上真机效率高哦！！！

爬取指定UID下的粉丝

屁话少说，放码过来（Talk is cheap，show me the code！）

提取指定UID下的所有粉丝，这个可以说很有用。因为对于抖音来说用户的UID在的话就可以获取一系列的用户信息，完成我们想要的一些业务功能。本身代码并不复杂，这里就仅仅是抛砖引玉，供有需要的朋友参考学习。

为了代码能够顺利执行，需要安装必要的支持库，这里就不一一罗列了。同时需要在代码根目录下创建data文件夹，data文件夹下包含一个名为followers的txt文件，此文件为读取用户的配置文件。请把想要读取的uid一行一个填入该文件。修改script.py中的设备信息双击执行就可以看到效果哦。

import threading
import time

from appium import webdriver
from appium.webdriver.webdriver import WebDriver
import os
import spider


def get_device_session(name: str, version: str, port: int = 8300, package='com.ss.android.ugc.aweme',
                       activity='com.ss.android.ugc.aweme.main.MainActivity') -> WebDriver:
    desired_caps = {}
    desired_caps['platformName'] = 'Android'
    desired_caps['udid'] = name
    desired_caps['deviceName'] = name
    desired_caps['platformVersion'] = version
    desired_caps['appPackage'] = package
    desired_caps['appActivity'] = activity
    # desired_caps['automationName'] = 'UiAutomator2'
    desired_caps['systemPort'] = port
    desired_caps["noReset"] = True
    desired_caps["newCommandTimeout"] = 600000
    device = webdriver.Remote('http://127.0.0.1:4723/wd/hub', desired_caps)
    device.implicitly_wait(15)
    return device


def run():
    device = get_device_session('45c6a7a8', '7.0.0') # 替换为自己的手机信息
    data = read_data()
    for user in data:
        file_path = os.path.join(os.getcwd(), "data", user)
        if not os.path.exists(file_path):
            os.mkdir(file_path)
        jump_to_user(device, user)
        time.sleep(5)
        try:
            script(device)
            remove_duplicates(file_path)
            print("用户:{}读取完毕".format(user))
        except Exception as e:

            print("读取UID:{}用户时发生异常,异常原因:{}".format(user, e.args))
        device.close_app()
        device.activate_app("com.ss.android.ugc.aweme")
        time.sleep(3)


def script(device: WebDriver):
    device.find_element_by_xpath('//android.widget.TextView[@text="粉丝"]').click()
    while True:
        elements = device.find_elements_by_xpath("//android.widget.TextView")
        temp = False
        for e in elements:
            if e.text == "没有更多了~":
                temp = True
                break
        if temp:
            break
        try:
            for _ in range(20):
                device.swipe(600, 600, 600, 1200)
                time.sleep(0.1)
                device.flick(600, 1200, 600, 600)
                time.sleep(0.1)
                device.flick(600, 1200, 600, 600)
        except Exception as e:
            print(e)
            pass


def jump_to_user(device: WebDriver, uid: str):
    url = "snssdk1128://user/profile/{}?refer=web&gd_label=click_wap_profile_bottom&type=need_follow&needlaunchlog=1"
    device.get(url.format(uid))


def read_data():
    os.chdir(os.path.abspath(os.path.dirname(__file__)))
    with open("./data/followers.txt", encoding="utf8", mode="r") as f:
        lines = f.readlines()
        result = []
        for line in lines:
            result.append(line.strip())
        return set(result)


def remove_duplicates(path):
    true_path = os.path.join(path, "followers.txt")

    with open(true_path, encoding="utf-8", mode="r") as file:
        data = file.readlines()

    result = sorted(set(data))

    with open(true_path, encoding="utf-8", mode="w") as file:
        file.write("".join(result))

    with open(os.path.join(path, "monitor.txt"), encoding="utf8", mode='a') as file:
        temp = "[{}] : 监控到粉丝数量为 {} 个\n".format(get_time(), len(result))
        print(temp)
        file.write(temp)


def get_time():
    time_now = time.strftime("%m-%d %H:%M:%S")
    return time_now


if __name__ == '__main__':
    thread = threading.Thread(target=spider.run)
    thread.start()
    run()
    # remove_duplicates("E:\\Python\\android\\monitor\\data\\73221748146")

import json
import time
import os

import mitmproxy.http


class Follower():
    def response(self, flow: mitmproxy.http.HTTPFlow):
        url = flow.request.url
        if "user/follower/list" in url:
            data = json.loads(flow.response.text)
            followers = data['followers']
            user_id = url.split("user_id=")[1].split('&max_time')[0]
            print("[{}] : 用户:{}读取到{}个粉丝".format(time.strftime("%d-%H:%M:%S"), user_id, len(followers)))
            with open("data/{}/followers.txt".format(user_id), mode="a", encoding="utf8") as f:
                for follower in followers:
                    nickname = follower["nickname"]
                    uid = follower["uid"]
                    if nickname != "已重置":
                        f.write(uid + "\n")


addons = [
    Follower()
]


def run():
    print("将会启动mitmproxy服务")
    os.system("mitmdump -q -s spider.py")


if __name__ == '__main__':
    run()

没写什么注释大家随缘看，代码丑陋，玻璃心勿喷！！！