NO.15——使用Appium自動化測試爬取微信朋友圈資料
阿新 • • 發佈:2019-01-02
一、解析過程
本人使用錘子手機做測試,型號是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