1. 程式人生 > >NO.15——使用Appium自動化測試爬取微信朋友圈資料

NO.15——使用Appium自動化測試爬取微信朋友圈資料

     一、解析過程

本人使用錘子手機做測試,型號是YQ601,首先開啟開發者模式確保手機能與mac相連,開啟Appium客戶端,配置引數如圖


可以理解為Appuim繼承自web端的selenium,同樣可以執行一些自動化操作。Appium自帶了一個XPATH選擇器,給使用者提供了選擇結果,如圖


這個選擇器給出的結果太繁瑣,所以可以改成通過查詢ID的方式來構造爬蟲程式。但是這裡要注意,估計微信提升了自己的反爬能力,在測試時發現,每次重新連線手機,對應特定節點的ID都會發生變化,保險起見,每次重新連線手機,都要對節點ID作更新。

       這裡把程式分為三部分:(1)模擬登陸(2)進入朋友圈(3)抓取動態(4)存入資料庫

二、程式碼

import os
from appium import webdriver
from appium.webdriver.common.touch_action import TouchAction
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from pymongo import MongoClient
from time import sleep
from processor import Processor
from config import *


class Moments():
    def __init__(self):
        """
        初始化
        """
        # 驅動配置
        self.desired_caps = {
            'platformName': PLATFORM,
            'deviceName': DEVICE_NAME,
            'appPackage': APP_PACKAGE,
            'appActivity': APP_ACTIVITY
        }
        self.driver = webdriver.Remote(DRIVER_SERVER, self.desired_caps)
        self.wait = WebDriverWait(self.driver, TIMEOUT)
        self.client = MongoClient(MONGO_URL)
        self.db = self.client[MONGO_DB]
        self.collection = self.db[MONGO_COLLECTION]
        # 處理器
        self.processor = Processor()
    
    def login(self):
        """
        登入微信
        :return:
        """
        # 登入按鈕
        login = self.wait.until(EC.presence_of_element_located((By.ID, 'com.tencent.mm:id/d75')))
        login.click()
        # 手機輸入
        phone = self.wait.until(EC.presence_of_element_located((By.ID, 'com.tencent.mm:id/hz')))
        phone.set_text(USERNAME)
        # 下一步
        next = self.wait.until(EC.element_to_be_clickable((By.ID, 'com.tencent.mm:id/alr')))
        next.click()
        # 密碼
        password = self.wait.until(
            EC.presence_of_element_located((By.XPATH, '//*[@resource-id="com.tencent.mm:id/hz"][1]')))
        password.set_text(PASSWORD)
        # 提交
        submit = self.wait.until(EC.element_to_be_clickable((By.ID, 'com.tencent.mm:id/alr')))
        submit.click()
        # 是否檢視通訊錄
        yesORnot = self.wait.until(EC.element_to_be_clickable((By.ID, 'com.tencent.mm:id/an2')))
        yesORnot.click()
    
    def enter(self):
        """
        進入朋友圈
        :return:
        """
        # 選項卡
        tab = self.wait.until(
            EC.presence_of_element_located((By.XPATH, '//*[@resource-id="com.tencent.mm:id/b0w"][3]')))
        tab.click()
        # 朋友圈
        moments = self.wait.until(EC.presence_of_element_located((By.ID, 'com.tencent.mm:id/a7f')))
        moments.click()
        
    
    def crawl(self):
        """
        爬取
        :return:
        """
        while True:
            # 當前頁面顯示的所有狀態
            items = self.wait.until(
                EC.presence_of_all_elements_located(
                    (By.XPATH, '//*[@resource-id="com.tencent.mm:id/d58"]//android.widget.FrameLayout')))
            # 上滑
            self.driver.swipe(FLICK_START_X, FLICK_START_Y + FLICK_DISTANCE, FLICK_START_X, FLICK_START_Y)
            # 遍歷每條狀態
            for item in items:
                try:
                    # 暱稱
                    nickname = item.find_element_by_id('com.tencent.mm:id/as6').get_attribute('text')
                    # 正文
                    content = item.find_element_by_id('com.tencent.mm:id/ib').get_attribute('text')
                    # 日期
                    date = item.find_element_by_id('com.tencent.mm:id/dfw').get_attribute('text')
                    # 處理日期
                    date = self.processor.date(date)
                    print(nickname, content, date)
                    data = {
                        'nickname': nickname,
                        'content': content,
                        'date': date,
                    }
                    # 插入MongoDB
                    self.collection.update({'nickname': nickname, 'content': content}, {'$set': data}, True)
                    sleep(SCROLL_SLEEP_TIME)
                except NoSuchElementException:
                    pass
    
    def main(self):
        """
        入口
        :return:
        """
        # 登入
        self.login()
        # 進入朋友圈
        self.enter()
        # 爬取
        self.crawl()


if __name__ == '__main__':
    moments = Moments()
    moments.main()
import time
import re


class Processor():
    def date(self, datetime):
        """
        處理時間,轉化成釋出時間的時間戳
        :param datetime: 原始時間
        :return: 處理後時間
        """
        if re.match('\d+分鐘前', datetime):
            minute = re.match('(\d+)', datetime).group(1)
            datetime = time.strftime('%Y-%m-%d', time.localtime(time.time() - float(minute) * 60))
        if re.match('\d+小時前', datetime):
            hour = re.match('(\d+)', datetime).group(1)
            datetime = time.strftime('%Y-%m-%d', time.localtime(time.time() - float(hour) * 60 * 60))
        if re.match('昨天', datetime):
            datetime = time.strftime('%Y-%m-%d', time.localtime(time.time() - 24 * 60 * 60))
        if re.match('\d+天前', datetime):
            day = re.match('(\d+)', datetime).group(1)
            datetime = time.strftime('%Y-%m-%d', time.localtime(time.time()) - float(day) * 24 * 60 * 60)
        return datetime