使用Appium爬取噹噹網App的全部數據

01 目標

場景：有時候經過傳統的方法去爬一些 Web 網頁或者 APP，受限於對方的反爬方案，很難爬到想要的數據，這個時候能夠考慮使用「Appium」結合「mitmproxy」的方式去爬取數據。

其中，Appium 負責驅動 App 端自動化運行，mitmproxy 負責截取請求數據並解析保存到數據庫。

今天的目的是爬取「噹噹網」的全部數據，並保存到 MongoDB 數據庫當中。

02 準備工做

首先，須要在 PC 上安裝好 Charles 和 Appium Desktop，並配置好 mitmproxy 環境。

# 安裝mitmproxy依賴包
pip3 install mitmproxy

# 安裝pymongodb
pip3 install pymongo
複製代碼

另外，須要準備一臺 Android 手機，另外 PC 端配置好 Android 開發環境。

03 爬取思路

1. 在配置好手動代理的狀況下，打開 Charles 實時捕獲客戶端的發起的網絡請求。

打開噹噹網搜索商品的頁面，搜索關鍵字「Python」,能夠在 Charles 查看到當前請求的 URL 地址包含：「word=Python」

編寫 mitmproxy 的執行腳本文件，重寫 response() 函數，經過對請求的 URL 進行過濾，對有用的數據進行整理並保存到 MongoDB 數據庫當中。

class DangDangMongo(object):
    """ 初始化MongoDB數據庫 """
    def __init__(self):
        self.client = MongoClient('localhost')
        self.db = self.client['admin']
        self.db.authenticate("root", "xag")
        self.dangdang_book_collection = self.db['dangdang_book']

def response(flow):

    # 過濾請求的URL
    if 'keyword=Python' in request.url:

        data = json.loads(response.text.encode('utf-8'))

        # 書籍
        products = data.get('products') or None

        product_datas = []

        for product in products:
            # 書ID
            product_id = product.get('id')

            # 書名
            product_name = product.get('name')

            # 書價格
            product_price = product.get('price')

            # 做者
            authorname = product.get('authorname')

            # 出版社
            publisher = product.get('publisher')

            product_datas.append({
                'product_id': product_id,
                'product_name': product_name,
                'product_price': product_price,
                'authorname': authorname,
                'publisher': publisher
            })

        DangDangMongo().dangdang_book_collection.insert_many(product_datas)
        print('成功插入數據成功')
複製代碼

先打開客戶端的手動代理監聽 8080 端口，而後執行「mitmdump」命令，而後滾動商品界面，發現數據到寫入到數據庫中了。

mitmdump -s script_dangdang.py 複製代碼

2. 下面咱們要利用 Appium 幫咱們實現 自動化。

首先打開 Appium Desktop，並啓動服務。

打開 Android Studio，利用菜單欄的 Build-Analyze APK 分析噹噹網的安卓應用，打開 AndroidManifest.xml

能夠發現應用包名和初始化 Activity 分別爲：

com.dangdang.buy二、com.dangdang.buy2.StartupActivity

獲取到包名和初始 Activity 後，就能夠利用 WebDriver 去模擬打開噹噹網 APP。

self.caps = {
            'automationName': DRIVER,
            'platformName': PLATFORM,
            'deviceName': DEVICE_NAME,
            'appPackage': APP_PACKAGE,
            'appActivity': APP_ACTIVITY,
            'platformVersion': ANDROID_VERSION,
            'autoGrantPermissions': AUTO_GRANT_PERMISSIONS,
            'unicodeKeyboard': True,
            'resetKeyboard': True
        }
self.driver = webdriver.Remote(DRIVER_SERVER, self.caps)
複製代碼

接着使用 Android SDK 自帶的工具 uiautomatorviewer 獲取到元素信息，使用 Appium 中的 WebDriver 去操做 UI 元素。

第一次打開應用的時候，可能會出現紅包雨對話框、新人專享紅包對話框、切換城市對話框，這裏須要經過元素 ID 獲取到關閉按鈕，執行點擊操做來關閉這些對話框。

這裏建立一個 新的線程 來單獨處理這些對話框。

class ExtraJob(threading.Thread):
   def run(self):
        while self.__running.isSet():

            # 爲True時當即返回, 爲False時阻塞直到內部的標識位爲True後返回
            self.__flag.wait()

            # 1.0 【紅包雨】對話框
            red_packet_element = is_element_exist(self.driver, 'com.dangdang.buy2:id/close')
            if red_packet_element:
                red_packet_element.click()

            # 1.1 【新人專享券】對話框
            new_welcome_page_sure_element = is_element_exist(self.driver, 'com.dangdang.buy2:id/dialog_cancel_tv')
            if new_welcome_page_sure_element:
                new_welcome_page_sure_element.click()

            # 1.2 【切換位置】對話框
            change_city_cancle_element = is_element_exist(self.driver, 'com.dangdang.buy2:id/left_bt')
            if change_city_cancle_element:
                change_city_cancle_element.click()

extra_job = ExtraJob(dangdang.driver)
extra_job.start()
複製代碼

接下來就是點擊搜索按鈕，而後輸入內容，執行點擊搜索對話框。

# 1.搜索框
search_element_pro = self.wait.until(
            EC.presence_of_element_located((By.ID, 'com.dangdang.buy2:id/index_search')))
search_element_pro.click()

search_input_element = self.wait.until(
            EC.presence_of_element_located((By.ID, 'com.dangdang.buy2:id/search_text_layout')))
search_input_element.set_text(KEY_WORD)

# 2.搜索對話框，開始檢索
search_btn_element = self.wait.until(
            EC.element_to_be_clickable((By.ID, 'com.dangdang.buy2:id/search_btn_search')))
search_btn_element.click()

# 3.休眠3秒，保證第一頁的內容加載徹底
time.sleep(3)複製代碼

待第一頁的數據加載徹底以後，能夠一直向上滾動頁面，直到數據所有被加載徹底，數據會由 mitmproxy 自動保存到 MongoDB 數據庫當中。

while True:
      str1 = self.driver.page_source
      self.driver.swipe(FLICK_START_X, FLICK_START_Y + FLICK_DISTANCE, FLICK_START_X, FLICK_START_X)
      time.sleep(1)
      str2 = self.driver.page_source
      if str1 == str2:
            print('中止滑動')
            # 中止線程
            extra_job.stop()
            break
      print('繼續滑動'複製代碼

04 結果

首先使用 mitmdump 開啓請求監聽的服務，而後執行爬取腳本。

App 會自動打開，執行一系列操做後，到達商品界面，而後自動滑動界面，經過 mitmproxy 自動把有用的數據保存到 MongoDB 數據庫中。

本文首發於公衆號「 AirPython 」，後臺回覆「 Appium」便可獲取完整代碼。