您现在的位置是:网站首页>技术文章>Python个人网站

Appium+mitmproxy实现自动化抓取抖音数据

Petrichor2019-12-12 16:00Python470人已围观

简介个人网站,用来做什么?我刚开始就把它当做一个我吐槽心情的地方,也就相当于一个网络记事本,也会放上一些照片。后来也用来记录自己所学习的知识。

1. mitmproxy

      mitmproxy 就是用于 MITM 的 proxy,MITM 即中间人攻击(Man-in-the-middle attack)。用于中间人攻击的代理首先会向正常的代理一样转发请求,保障服务端与客户端的通信,其次,会适时的查、记录其截获的数据,或篡改数据,引发服务端或客户端特定的行为。

      不同于 fiddler 或 wireshark 等抓包工具,mitmproxy 不仅可以截获请求帮助开发者查看、分析,更可以通过自定义脚本进行二次开发。举例来说,利用 fiddler 可以过滤出浏览器对某个特定 url 的请求,并查看、分析其数据,但实现不了高度定制化的需求,类似于:“截获对浏览器对该 url 的请求,将返回内容置空,并将真实的返回内容存到某个数据库,出现异常时发出邮件通知”。而对于 mitmproxy,这样的需求可以通过载入自定义 python 脚本轻松实现。

2. 安装

pip install mitmproxy

      安装之后我们就可以使用mitmdump,mitmweb,mitmproxy这三个命令。mitmdump可以允许我们使用自定义脚本来运行,mitmweb会打开一个网页。

3. 使用

      首先在终端中执行mitmdump命令,就可以运行起一个代理服务器,然后在手机上设置代理,填入电脑的IP和端口,当我们在手机上操作时就可以在终端上看到手机所发送的HTTP请求,但我们会发现当时HTTPS的请求时会出现报错,此时我们就需要在手机的浏览器访问mitm.it,按对应系统安装好证书后,就可以访问HTTPS请求了。

4. 自定义脚本启动代理

      mitmdump还支持自定义脚本启动代理,使用mitmdump -s <脚本文件>命令就可以启动了。在脚本里我们可以修改请求的一些参数,也可以获取请求的响应。

import json
from mitmproxy import http


class Spider(object):

    def response(self, flow: http.HTTPFlow):
        user_info_url = 'https://aweme-eagle-ipv6.snssdk.com/aweme/v1/user/'

        if flow.request.url.startswith(user_info_url):
            print('用户信息------------')
            response = flow.response.get_text()
            self.parse_user_info_response(response)


    def parse_user_info_response(self, response):
        print('running ....... parse user info response')
        with open('user_data.txt', 'w') as w:
            w.write(response)


addons = [
    Spider()
]

5. Appium

import json
import time
from appium import webdriver
import os


class AutoControl(object):
    desired_caps = dict()
    desired_caps['platformName'] = 'Android'
    desired_caps['platformVersion'] = '5.1.1'
    desired_caps['deviceName'] = 'UHESY7W6NH'
    desired_caps['appPackage'] = 'com.ss.android.ugc.aweme'
    desired_caps['appActivity'] = 'com.ss.android.ugc.aweme.main.MainActivity'
    desired_caps['noReset'] = 'true'

    def __init__(self):
        self.driver = webdriver.Remote('http://localhost:4723/wd/hub', self.desired_caps)
        self.size = self.driver.get_window_size()
        self.width = self.size['width']
        self.height = self.size['height']
        self.duration = 500

    def swipe_up(self):
        x1 = self.width * 0.5
        y1 = self.height * 0.6
        y2 = self.height * 0.4
        self.driver.swipe(x1, y1, x1, y2, self.duration)

    def swipe_down(self):
        x1 = self.width * 0.5
        y1 = self.height * 0.4
        y2 = self.height * 0.6
        self.driver.swipe(x1, y1, x1, y2, self.duration)

    def swipe_left(self):
        x1 = self.width * 0.8
        y1 = self.height * 0.5
        x2 = self.width * 0.2
        self.driver.swipe(x1, y1, x2, y1, self.duration)

    def swipe_right(self):
        x1 = self.width * 0.3
        y1 = self.height * 0.5
        x2 = self.width * 0.6
        self.driver.swipe(x1, y1, x2, y1, self.duration)

    def follow(self):
        self.driver.tap([(557, 572)], 100)

    def star(self):
        self.driver.tap([(544, 612)], 100)

    def get_comment(self):
        self.driver.tap([(549, 689)], 100)

    def close_comment(self):
        self.driver.tap([(280, 216)], 100)
        time.sleep(1)
        self.driver.tap([(280, 200)], 100)

    def search(self, keyword):
        self.driver.tap([(22, 48)], 100)
        time.sleep(3)
        self.driver.find_element_by_id('com.ss.android.ugc.aweme:id/ahw').send_keys(keyword)
        time.sleep(2)
        self.driver.tap([(552, 785)], 100)

    def comment(self, content):
        time.sleep(3)
        comment_count = self.driver.find_element_by_id('com.ss.android.ugc.aweme:id/ac5').text
        self.driver.tap([(549, 689)], 100)
        time.sleep(2)
        if comment_count != 0:
            self.driver.tap([(96, 947)], 100)
        time.sleep(2)
        self.driver.find_element_by_id('com.ss.android.ugc.aweme:id/a1b').send_keys(content)
        time.sleep(2)
        self.driver.find_element_by_id('com.ss.android.ugc.aweme:id/a1r').click()
        time.sleep(1)
        self.driver.tap([(280, 200)], 100)

 

很赞哦!(5)

文章评论