關于微信小程序爬蟲token自動更新問題
更新時間：2021年09月27日 08:45:56 作者：blackpearl9
本文主要介紹了關于微信小程序爬蟲關于token自動更新問題，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
現(xiàn)在很多的app都很喜歡在微信或者支付寶的小程序內(nèi)做開發(fā)，畢竟比較方便、安全、有流量、不需要再次下載app，好多人會因為加入你讓他下載app他會扭頭就走不用你的app，畢竟做類似產(chǎn)品的不是你一家。
之前做過很多微信小程序的爬蟲任務，今天做下記錄，防止很久不用后就會忘記，微信小程序分為兩大類：
1、是不需要登錄的（這種的話不做分析，畢竟沒什么反爬）
2、需要登錄的
2.1 登錄一次之后token永久有效
2.2 登錄一次token幾分鐘內(nèi)到幾小時內(nèi)失效
2.2.1 登錄后一段時間后token時候需要再次調(diào)用微信內(nèi)部方法生成code去換取token（本次主要做的）
2.2.2 跟2.2.1類似，然后又加了一道校驗，比如圖片驗證碼，這個類似于微信公眾號的茅臺預約那種（本次不做分析）
微信小程序的登錄其實跟其他的web登錄不太一樣，一般的web登錄或者是app登錄基本上就是用戶名+密碼+驗證碼（圖片或者短信）就可以，微信的邏輯是假如你需要登錄的話需要獲得用戶的授權，之后調(diào)用微信的內(nèi)部方法生成一個code，code只能用一次之后就實效，微信解釋這個code有效期是5分鐘左右。
這里是具體流程：https://developers.weixin.qq.com/community/develop/doc/000c2424654c40bd9c960e71e5b009?highLine=code
之前爬取過的一個小程序他的反爬是token有效期一個小時，然后單次token可用大概100次左右，當單個token使用次數(shù)或者單小時內(nèi)使用次數(shù)超過100次就直接封號處理，24小時內(nèi)也有頻率控制，所以就需要我每小時一次每小時一次的去獲取token，當然，因為我是個程序猿，所以我不能每小時手動的去獲取這個token，比較這不是我們的風格。
這里需要的是python+fiddler+appium+模擬器，大致的思路是通過appium去操控模擬器模擬點擊微信的小程序，定期的去做點擊，然后fiddler去從請求的頭部信息中獲取到token，之后寫到本地文件中，然后python程序定時的去判斷這個本地文件是否進行了更新，更新了的話通過正則來獲取到token_list之后去最后一個，因為有可能是當前保存的token已經(jīng)失效了，小程序還會再次去拿這個token嘗試請求一下，假如失效了會調(diào)用微信的內(nèi)部方法生成code來換取token，我這里的爬蟲主代碼是運行在服務器的，所有又增加了Redis來存儲token。
一、微信模擬點擊
微信按照需求條件時間頻率模擬點擊、滑動、退出等操作，以下的ding_talk的send_msg是增加的釘釘發(fā)送消息，此處不再添加，有需求的可以自己查看釘釘機器人文檔或者依據(jù)自己的需求調(diào)整自己的消息提醒。
weixin.py
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):
"""
初始化
"""
# 驅(qū)動配置
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 # 根據(jù)周控制頻率控制再次開啟時間
self.week_start_xiu = time.time() # xd的開始時間
self.week_end_xiu = 0 # 根據(jù)周控制頻率控制再次開啟時間
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 ')
# 手機輸入
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="請?zhí)顚懳⑿琶艽a"]')))
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:
"""
# 滑動之后等待出現(xiàn)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('需要發(fā)送失敗消息')
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):
"""
每次進來之后都需要判斷是否到了時間，若時間到了之后才可執(zhí)行點擊操作
:param : en 代表en; xd 代表xd; xiu 代表xiu.
:return: None 無返回值
"""
print(self.date_end_en, time.time())
if self.date_end_en < time.time(): # 此時的時候已經(jīng)超時，需要再次從新進行點擊
print('en模擬點擊')
# 從新定義開始結(jié)束時間
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"http://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, "http://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"http://android.widget.FrameLayout[2]/android.widget.ImageButton")))
close_button.click()
print('點擊了關閉小程序')
def touch_xd(self):
"""
需要考慮是否已經(jīng)登錄狀態(tài)還是需要再次登錄
:return:
"""
# 點擊后進入到小程序
logging.info('click app xd')
xd_app = self.wait.until(EC.presence_of_element_located((By.XPATH, "http://android.widget.TextView[@text='textname']")))
xd_app.click()
time.sleep(20)
# 頁面出現(xiàn)需要獲取到你的定位的時候需要點擊允許
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"http://android.widget.FrameLayout[2]/android.widget.ImageButton")))
close_button.click()
print('結(jié)束')
time.sleep(30)
def touch_xiu(self):
"""
xiu模擬點擊，需要考慮是否需要登錄狀態(tài)下
:return:
"""
# 點擊后進入到小程序
logging.info('click app xiu')
xiu_app = self.wait.until(EC.presence_of_element_located((By.XPATH, "http://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, "http://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, "http://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, "http://android.view.View[@content-desc='立即登錄']")))
login.click()
except Exception as e:
print('xiu已經(jīng)登錄，不需要再次點擊確認登錄')
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: # 監(jiān)控Redis情況，當Redis中無數(shù)據(jù)后開始運行一次
res = we_chat.run() # 操作微信做操作點擊en小程序生成token
if not res:
count_1 += 1
if count_1 > 10:
break # 當失敗十次之后跳出循環(huán)
# 此處增加限制，每次生成token之后一個小時后才會產(chǎn)生新的token，防止一個token多次使用導致被封號
time.sleep(60*60)
else:
time.sleep(60*60) # 當有數(shù)據(jù)的時候等待五分鐘
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'業(yè)務報警：' \
f'\n en獲取token出現(xiàn)問題' \
f'\n{e}'
send_msg(msg)
# print(e, type(e))
config.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 = 'http://localhost:4723/wd/hub'
# 等待元素加載時間
TIMEOUT = 10
# 微信手機號密碼
USERNAME = 'wechatname'
PASSWORD = 'wechatpwd'
# 滑動點
FLICK_START_X = 300
FLICK_START_Y = 300
FLICK_DISTANCE = 700
以下是處理文件，將token獲取到后放到Redis中，或者你可以依照你的想法調(diào)整
handle_file.py
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) # 解析當前經(jīng)緯度范圍內(nèi)店鋪點位
# 處理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存在的話 則根據(jù)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()
二、配置fiddler獲取請求頭的信息寫到本地文件
修改fiddlerscript添加以下內(nèi)容，在做數(shù)據(jù)請求的以下增加下面內(nèi)容
fiddler
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();
}
三、主爬蟲業(yè)務代碼
此處按照自己的需求邏輯調(diào)整自己的業(yè)務代碼。
到此這篇關于關于微信小程序爬蟲token自動更新問題的文章就介紹到這了,更多相關小程序爬蟲token自動更新內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
文本轉(zhuǎn)語音，一般會用在無障礙開發(fā)。下面介紹如何使用Python實現(xiàn)將文本文件轉(zhuǎn)換成語音輸出。跟著小編一起來看看吧。2016-08-08
這篇文章主要介紹了Python ollama的搭建與使用流程分析,詳細介紹了ollama的安裝方式,本文結(jié)合實例給大家介紹的非常詳細,感興趣的朋友跟隨小編一起看看吧2024-04-04
python如何使用socketserver模塊實現(xiàn)并發(fā)聊天
這篇文章主要介紹了python如何使用socketserver模塊實現(xiàn)并發(fā)聊天,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-12-12
這篇文章主要介紹了python asyncio 協(xié)程庫的使用，幫助大家更好的理解和使用python，感興趣的朋友可以了解下2021-01-01
Python使用APScheduler實現(xiàn)定時任務過程解析
這篇文章主要介紹了Python使用APScheduler實現(xiàn)定時任務過程解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-09-09
今天我們就用 Python 來實現(xiàn)自動簽到，省得我每天打開 APP 來操作了。感興趣的朋友跟隨小編一起看看吧2018-09-09
python實戰(zhàn)之利用pygame實現(xiàn)貪吃蛇游戲（二）
這篇文章主要介紹了python實戰(zhàn)之利用pygame實現(xiàn)貪吃蛇游戲（二）,文中有非常詳細的代碼示例,對正在學習python的小伙伴們有很好的幫助,需要的朋友可以參考下2021-05-05
這篇文章主要為大家介紹了python?json-rpc?規(guī)范的源碼閱讀，以及jsonrpcclient與jsonrpcserver的實現(xiàn)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-10-10
2048作為經(jīng)典的手機端游戲一直深受大家的喜愛，本文將利用Python中的Pygame模塊制作簡單版的2048游戲，感興趣的小伙伴可以了解一下2022-02-02
這篇文章主要介紹了Python模塊Uvicorn實戰(zhàn),文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2020-01-01