關於微信小程式爬蟲關於token自動更新問題
現在很多的app都很喜歡在微信或者支付寶的小程式內做開發，畢竟比較方便、安全、有流量、不需要再次下載app，好多人會因為加入你讓他下載app他會扭頭就走不用你的app，畢竟做類似產品的不是你一家。
之前做過很多微信小程式的爬蟲任務，今天做下記錄，防止很久不用後就會忘記，微信小程式分為兩大類：
1、是不需要登錄的（這種的話不做分析，畢竟沒什麼反爬）
2、需要登錄的
2.1 登錄一次之後token永久有效
2.2 登錄一次token幾分鐘內到幾小時內失效
2.2.1 登錄後一段時間後token時候需要再次調用微信內部方法生成code去換取token（本次主要做的）
2.2.2 跟2.2.1類似，然後又加了一道校驗，比如圖片驗證碼，這個類似於微信公眾號的茅台預約那種（本次不做分析）
微信小程式的登錄其實跟其他的web登錄不太一樣，一般的web登錄或者是app登錄基本上就是用戶名+密碼+驗證碼（圖片或者簡訊）就可以，微信的邏輯是假如你需要登錄的話需要獲得用戶的授權，之後調用微信的內部方法生成一個code，code只能用一次之後就實效，微信解釋這個code有效期是5分鐘左右。
這裡是具體流程：//developers.weixin.qq.com/community/develop/doc/000c2424654c40bd9c960e71e5b009?highLine=code
之前爬取過的一個小程式他的反爬是token有效期一個小時，然後單次token可用大概100次左右，當單個token使用次數或者單小時內使用次數超過100次就直接封號處理，24小時內也有頻率控制，所以就需要我每小時一次每小時一次的去獲取token，當然，因為我是個程式猿，所以我不能每小時手動的去獲取這個token，比較這不是我們的風格。
這裡需要的是python+fiddler+appium+模擬器，大致的思路是通過appium去操控模擬器模擬點擊微信的小程式，定期的去做點擊，然後fiddler去從請求的頭部資訊中獲取到token，之後寫到本地文件中，然後python程式定時的去判斷這個本地文件是否進行了更新，更新了的話通過正則來獲取到token_list之後去最後一個，因為有可能是當前保存的token已經失效了，小程式還會再次去拿這個token嘗試請求一下，假如失效了會調用微信的內部方法生成code來換取token，我這裡的爬蟲主程式碼是運行在伺服器的，所有又增加了Redis來存儲token。
一、微信模擬點擊
微信按照需求條件時間頻率模擬點擊、滑動、退出等操作，以下的ding_talk的send_msg是增加的釘釘發送消息，此處不再添加，有需求的可以自己查看釘釘機器人文檔或者依據自己的需求調整自己的消息提醒。
import time import logging from appium import webdriver from ding_talk import send_msg from handle_file import EnToken from conf.dbr import RedisClient from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from config import * LOG_FORMAT = "%(asctime)s - %(levelname)s - line:%(lineno)s - msg:%(message)s" logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) # logging.FileHandler(filename='app.log', encoding='utf-8') # 微信獲取en token class WeChat(object): def __init__(self): """ 初始化 """ # 驅動配置 self.desired_caps = { 'platformName': PLATFORM, 'deviceName': DEVICE_NAME, 'appPackage': APP_PACKAGE, 'appActivity': APP_ACTIVITY, 'noReset': True } self.driver = webdriver.Remote(DRIVER_SERVER, self.desired_caps) self.wait = WebDriverWait(self.driver, TIMEOUT) self.hours_en = 60 * 60 * 1.1 # en控制1.1小時模擬點擊一次 self.date_start_en = time.time() # en開始時間 self.date_end_en = 0 # en超過此時間後再次運行 # self.date_end_en = self.date_start_en + self.hours_en # en超過此時間後再次運行 self.week = 60 * 60 * 24 * 7 # 按照周的頻率對xd進行token更新 self.week_start_xd = time.time() # xd的開始時間 self.week_end_xd = 0 # 根據周控制頻率控制再次開啟時間 self.week_start_xiu = time.time() # xd的開始時間 self.week_end_xiu = 0 # 根據周控制頻率控制再次開啟時間 def login(self): """ 登錄微信 :return: """ # 登錄按鈕 a = time.time() try: login = self.wait.until(EC.presence_of_element_located((By.ID, 'com.tencent.mm:id/f34'))) login.click() except Exception as e: # print(e) logging.info(f'failed login {e}') b = time.time() - a # print('點擊登錄', b) logging.info(f'click login,use time {b}') # 手機輸入 try: phone = self.wait.until(EC.presence_of_element_located((By.ID, 'com.tencent.mm:id/bem'))) phone.set_text(USERNAME) except Exception as e: # print(e) logging.info(f'something wrong{e}') c = time.time() - a - b # print('手機號輸入', c) logging.info(f'send keys phone nums use time {c}') # 下一步 try: next = self.wait.until(EC.element_to_be_clickable((By.ID, 'com.tencent.mm:id/dw1'))) next.click() except Exception as e: logging.info(f'something wrong{e}') d = time.time() - a - b - c logging.info(f'click next bottom use time {c}') # 密碼 password = self.wait.until(EC.presence_of_element_located((By.XPATH, '//*[@text="請填寫微信密碼"]'))) password.set_text(PASSWORD) e = time.time() - a - b - c - d logging.info(f'send keys password use time {e}') # 提交 # submit = self.wait.until(EC.element_to_be_clickable((By.ID, 'com.tencent.mm:id/dw1'))) submit = self.wait.until(EC.element_to_be_clickable((By.XPATH, '//*[@text="登錄"]'))) submit.click() f = time.time() - a - b - c - d - e logging.info(f'commit password use time {f}') def run(self): """ 入口 :return: """ # 滑動之後等待出現en小程式 self.slide_down() time.sleep(10) # 點擊進入en小程式 self.touch_en() if self.week_end_xd < self.week_start_xd: self.week_start_xd = time.time() self.week_end_xd = self.week_start_xd + self.week print('xd點擊') self.touch_xd() elif self.week_end_xiu < self.week_start_xiu: self.week_end_xiu = time.time() + self.week print('xiu') self.touch_xiu() time.sleep(10) # 退出小程式 self.driver_closed() print('driver closed') emt = EnToken() token_res = emt.token_2_redis() if not token_res: print('需要發送失敗消息') return False return True def slide_down(self): """ 滑動微信螢幕之後點擊小程式 :return: """ window_size_phone = self.driver.get_window_size() # print(window_size_phone) phone_width = window_size_phone.get('width') phone_height = window_size_phone.get('height') # print(phone_width, phone_height) time.sleep(15) x1 = phone_width * 0.5 y1 = phone_height * 0.7 y2 = phone_height * 0.26 # print('準備向下滑動') logging.info(f'prepare slide down') a = time.time() self.driver.swipe(x1, y2, x1, y1, 2050) # print('向下滑動完成', time.time() - a) logging.info(f'slide down success use time {time.time() - a}') def touch_en(self): """ 每次進來之後都需要判斷是否到了時間，若時間到了之後才可執行點擊操作 :param : en 代表en; xd 代表xd; xiu 代表xiu. :return: None 無返回值 """ print(self.date_end_en, time.time()) if self.date_end_en < time.time(): # 此時的時候已經超時，需要再次從新進行點擊 print('en模擬點擊') # 從新定義開始結束時間 print(self.date_end_en, time.time()) self.date_end_en = time.time() + self.hours_en # 再次更改end time為n小時後 print(self.date_end_en, time.time()) try: # print('id定位en') en_app = self.wait.until( EC.presence_of_element_located((By.XPATH, f"//android.widget.TextView[@text='textname…']"))) # en_master = self.wait.until(EC.presence_of_element_located((By.ID, 'com.tencent.mm:id/hu'))) # en_master = self.wait.until( # EC.presence_of_element_located((By.XPATH, "//android.widget.TextView[@text='textname']"))) en_app.click() logging.info(f'located by app_name en') except Exception as error: # print(e, 'id定位失敗') logging.info(f'failed located by id:{error}') time.sleep(20) # 關閉小程式按鈕點擊 print('close the en app') close_button = self.wait.until(EC.presence_of_element_located((By.XPATH, f"//android.widget.FrameLayout[2]/android.widget.ImageButton"))) close_button.click() print('點擊了關閉小程式') def touch_xd(self): """ 需要考慮是否已經登錄狀態還是需要再次登錄 :return: """ # 點擊後進入到小程式 logging.info('click app xd') xd_app = self.wait.until(EC.presence_of_element_located((By.XPATH, "//android.widget.TextView[@text='textname']"))) xd_app.click() time.sleep(20) # 頁面出現需要獲取到你的定位的時候需要點擊允許 print('點擊確認獲取當前位置') self.driver.tap([(510, 679)], 500) # 點擊進入到個人中心 time.sleep(10) logging.info('click personal xd') self.driver.tap([(540, 1154)], 500) # 點擊快速登錄進行登錄 time.sleep(10) logging.info('click login xd') self.driver.tap([(270, 1030)], 500) # 點擊同意獲取頭像資訊 time.sleep(10) logging.info('同意獲取頭像等相關資訊') self.driver.tap([(510, 775)], 500) time.sleep(20) # 關閉小程式按鈕點擊 print('close the guaishou app') close_button = self.wait.until( EC.presence_of_element_located((By.XPATH, f"//android.widget.FrameLayout[2]/android.widget.ImageButton"))) close_button.click() print('結束') time.sleep(30) def touch_xiu(self): """ xiu模擬點擊，需要考慮是否需要登錄狀態下 :return: """ # 點擊後進入到小程式 logging.info('click app xiu') xiu_app = self.wait.until(EC.presence_of_element_located((By.XPATH, "//android.widget.TextView[@text='xiu']"))) xiu_app.click() # 若頁面顯示需要確認獲取當前位置的話需要點擊確認 logging.info('click confirm xiu') time.sleep(15) confirm_loc = self.wait.until( EC.presence_of_element_located((By.XPATH, "//android.widget.Button[@text='確定']"))) confirm_loc.click() # 點擊個人中心 logging.info('click personal xiu') time.sleep(5) try: personal = self.wait.until( EC.presence_of_element_located((By.XPATH, "//android.view.View[@content-desc='個人中心']"))) personal.click() except Exception as e: print(e) # 點擊快速登錄進行登錄 logging.info('click login xiu') time.sleep(5) try: login = self.wait.until(EC.presence_of_element_located((By.XPATH, "//android.view.View[@content-desc='立即登錄']"))) login.click() except Exception as e: print('xiu已經登錄，不需要再次點擊確認登錄') time.sleep(30) def driver_closed(self): self.driver.quit() if __name__ == '__main__': conn_r = RedisClient(db=10) count_1 = 0 # start_time = time.time() # end_time = time.time() + 60 * 60 * 1 we_chat = WeChat() try: while 1: if conn_r.r_size() < 3: # 監控Redis情況，當Redis中無數據後開始運行一次 res = we_chat.run() # 操作微信做操作點擊en小程式生成token if not res: count_1 += 1 if count_1 > 10: break # 當失敗十次之後跳出循環 # 此處增加限制，每次生成token之後一個小時後才會產生新的token，防止一個token多次使用導致被封號 time.sleep(60*60) else: time.sleep(60*60) # 當有數據的時候等待五分鐘 we_chat.driver = webdriver.Remote(DRIVER_SERVER, we_chat.desired_caps) we_chat.wait = WebDriverWait(we_chat.driver, TIMEOUT) except Exception as e: msg = f'業務報警：' \ f'\n en獲取token出現問題' \ f'\n{e}' send_msg(msg) # print(e, type(e)) logging.info(msg)
weixin.py
import os # 平台 PLATFORM = 'Android' # 設備名稱 通過 adb devices -l 獲取 DEVICE_NAME = 'MI_9' # APP路徑 APP = os.path.abspath('.') + '/weixin.apk' # APP包名 APP_PACKAGE = 'com.tencent.mm' # 入口類名 APP_ACTIVITY = '.ui.LauncherUI' # Appium地址 DRIVER_SERVER = '//localhost:4723/wd/hub' # 等待元素載入時間 TIMEOUT = 10 # 微信手機號密碼 USERNAME = 'wechatname' PASSWORD = 'wechatpwd' # 滑動點 FLICK_START_X = 300 FLICK_START_Y = 300 FLICK_DISTANCE = 700
config.py
以下是處理文件，將token獲取到後放到Redis中，或者你可以依照你的想法調整
import re import os import logging from conf.dbr import RedisClient LOG_FORMAT = "%(asctime)s - %(levelname)s - line:%(lineno)s - msg:%(message)s" logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT) # 處理en token到Redis class EnToken(object): def __init__(self): # self.token_path = 'F:\\en.txt' # self.token_path = 'F:\\xiu.txt' # self.token_path = 'F:\\xd.txt' self.conn = RedisClient(db=10) # 解析日維度價格 self.conn_en = RedisClient(db=9) # 解析當前經緯度範圍內店鋪點位 # 處理en token文件，從文件中讀取到token之後只取最後一個，取到之後刪除本地文件 @staticmethod def handle_en_txt(): token_dict = {} path_token_list = [ ('en', '>(e.*?)-->'), ('xd', 'headers-->(.*?)-->'), ('xiu', r'>(\d+)-->'), ] for i in path_token_list: token_path = f'F:\\{i[0]}.txt' token_re = i[-1] if os.path.exists(token_path): with open(token_path, mode='r', encoding='utf-8') as f: token_str = f.read() # print(token_str) # token_list = re.findall('>(e.*?)-->', token_str) # token_list = re.findall('>(Q.*?)-->', token_str) # token_list = re.findall('>(\d+)-->', token_str) token_list = re.findall(token_re, token_str) print(token_list) if token_list: token = token_list[-1] print(token) token_dict[i[0]] = token os.remove(token_path) # 刪除掉 # return token else: # print('file_en_dont_exit') logging.info('file_en_dont_exit') return token_dict # 將token放到Redis中 def token_2_redis(self): """ 假如token存在的話 則根據token的最後幾位做key放入到Redis中 :return: """ token_dict = self.handle_en_txt() print(token_dict) if token_dict: for token_items in token_dict.items(): token_key = token_items[0] token_val = token_items[-1] self.conn.set(token_key, token_val, over_time=None) # self.conn.set(token_key, token, over_time=60*65) # 設置有效時長65分鐘之後失效 # self.conn_en.set(token_key, token, over_time=60*65) # 設置有效時長65分鐘之後失效 logging.info(f'token success {token_key,token_val}') return True else: logging.info('token dons"t exist') self.conn.close() self.conn_en.close() if __name__ == '__main__': en = EnToken() en.token_2_redis()
handle_file.py
二、配置fiddler獲取請求頭的資訊寫到本地文件
修改fiddlerscript添加以下內容，在做數據請求的以下增加下面內容
if (oSession.oRequest["Host"]=="這裡是請求的host") { var filename = "F:\en.txt"; var curDate = new Date(); var logContent = 'en' + "[" + curDate.toLocaleString() + "]"; var sw : System.IO.StreamWriter; if (System.IO.File.Exists(filename)){ sw = System.IO.File.AppendText(filename); sw.Write(logContent + 'oSession.oRequest.headers-->' + oSession.oRequest.headers['x-wx-token'] + '-->' + oSession.oRequest.headers +'\n'); // sw.Write("Request header:" + "\n" + oSession.oRequest.headers); // sw.Write(wap_s + '\n\n') } else{ sw = System.IO.File.CreateText(filename); sw.Write(logContent + 'oSession.oRequest.headers-->' + oSession.oRequest.headers['x-wx-token'] + '-->' + '\n'); // sw.Write("Request header:" + "\n" + oSession.oRequest.headers); // sw.Write(wap_s + '\n\n') } sw.Close(); sw.Dispose(); }
fiddler
三、主爬蟲業務程式碼
此處按照自己的需求邏輯調整自己的業務程式碼。
如果對你有所幫助就請作者喝杯咖啡吧😊
