关于微信小程序爬虫token自动更新问题

目录
  • 一、微信模拟点击
  • 二、配置fiddler获取请求头的信息写到本地文件
  • 三、主爬虫业务代码

  现在很多的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分钟左右。

  这里是具体流程:https://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是增加的钉钉发送消息,此处不再添加,有需求的可以自己查看钉钉机器人文档或者依据自己的需求调整自己的消息提醒。

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):
        """
        初始化
        """
        # 驱动配置
        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))

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中,或者你可以依照你的想法调整

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)  # 解析当前经纬度范围内店铺点位

    # 处理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()

二、配置fiddler获取请求头的信息写到本地文件

修改fiddlerscript添加以下内容,在做数据请求的以下增加下面内容

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();
             }

三、主爬虫业务代码

此处按照自己的需求逻辑调整自己的业务代码。

到此这篇关于关于微信小程序爬虫token自动更新问题的文章就介绍到这了,更多相关小程序爬虫token自动更新内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • 微信小程序配置服务器提示验证token失败的解决方法

    最近在学习微信小程序,遇到的第一个问题就是需要配置服务器 关于这个服务器的配置我也是绕了好多弯路,说白了腾讯就是想通过你填的这个URL和Token去验证你有一个自己的服务器(外网可以访问的服务器),其实就是想让你证明你是你自己,呵呵.... 关于这个token随便填不要纠结,下面直接看JAVA代码 package com.base.action; import java.io.PrintWriter; import java.security.MessageDigest; import java

  • 微信小程序登录换取token的教程

    前言: 这次主要是介绍些业务逻辑,技术点倒是没有多少.不过在开发中,优秀的编程思路同样是非常值得学习的. 最近小程序可以说在开发届狠狠的火了一把.微信小程序可以开发游戏,腾讯率先带头,做出了一个跳一跳也是点爆朋友圈.所谓落后就要挨打,那么今天就开始学习小程序的一些小知识吧(本文基于十年磨一剑的tp5) 目录: 微信登录换取token的流程 如何将code变成openid和session_key 抛出错误异常和派发令牌 一:微信登录换取token的流程 多说无益,直接上图 小程序获取token.p

  • 微信小程序url与token设置详解

    微信小程序url与token设置详解 新浪云应用sae的代码里创建一个weixin.php文件,写入以下代码 define("TOKEN","myToken");// 后台填写的token,在微信公众平台启用 $wechatObj = new wechatAPI(); $wechatObj->isValid(); class wechatAPI { public function isValid()//验证微信接口,验证函数以外的代码和微信公众号开发token

  • 小程序开发实现access_token统一管理

    目录 TOKEN 定时刷新器 一.背景 二.access_token的内部设计 2.1 access_token的时效性 2.2 access_token 的逐渐失效性 三.access_token的统一管理 参考文档 TOKEN 定时刷新器 一.背景 对于使用过公众平台的API功能的开发者来说,access_token绝对不会陌生,它就像一个打开家门的钥匙,只要拿着它,就能使用公众平台绝大部分的API功能.因此,对于开发者而言,access_token的使用方式就变得尤其的重要.在日常API接

  • 关于微信小程序爬虫token自动更新问题

    目录 一.微信模拟点击 二.配置fiddler获取请求头的信息写到本地文件 三.主爬虫业务代码 现在很多的app都很喜欢在微信或者支付宝的小程序内做开发,毕竟比较方便.安全.有流量.不需要再次下载app,好多人会因为加入你让他下载app他会扭头就走不用你的app,毕竟做类似产品的不是你一家. 之前做过很多微信小程序的爬虫任务,今天做下记录,防止很久不用后就会忘记,微信小程序分为两大类: 1.是不需要登录的(这种的话不做分析,毕竟没什么反爬) 2.需要登录的 2.1 登录一次之后token永久有效

  • 微信小程序 Tab页切换更新数据

    微信小程序 Tab页切换更新数据 微信小程序还处于内测阶段,最不方便的莫过于官方在不停的更新,前几天写的功能隔个几天忽然发现不能用了_(:зゝ∠)_ 功能需求如下: 我在首页点击"5万以上"他会把跳转到买车页然后同时把"5万以上"这个筛选条件带到买车页. 之前navigator导航是可以跳转并携带数据的,但这一次官方更新加了个新东西-----switchTab,专门用来实现tab页的跳转,但禁止携带数据 那么如果还想要实现我们的效果只能用别的方法了 想了一下有两种思

  • 微信小程序如何处理token过期问题

    目录 先说结论 问题 解决方案 使用Promise封装回调函数 总结 先说结论 业务流程:  从网络日志中检查到token过期,则跳转到登录页面,要求用户重新登录. 代码逻辑:使用自定义的HttpUtil封装wx.request API,全局捕获过期token并自动处理后,下发给上层业务. 问题 Token过期的现象: 在网络请求中,客户端token会过段时间过期,使得后续的网络请求失败,抛出异常日志如下: data: {code: "99997", date: 16341748313

  • Fundebug支持监控微信小程序HTTP请求错误的方法

    摘要: Fundebug的微信小程序错误监控插件更新至0.5.0,支持监控HTTP请求错误. 接入插件 接入Fundebug的错误监控插件非常简单,只需要下载fundebug.0.5.0.min.js,在app.js中引入并配置apikey: var fundebug = require('./libs/fundebug.0.5.0.min.js') fundebug.apikey = "API-KEY"; 获取apikey需要[免费注册](https://www.fundebug.c

  • 微信小程序实现倒计时补零功能

    微信小程序中 "倒计时自动补零" 的一点代码,方法比较简单粗暴,想着以后怎么也能用到,就先总结出来了. 代码: js: //index.js var num = 10//计时 var strH = '' var strM = '' var strS = '' var timer = '' Page({ data: { timeText:''//展示 }, onLoad: function () { this.move() //计时开始 后面的1000是毫秒 每1000毫秒跳一次 tim

  • 使用python实现微信小程序自动签到功能

    功能描述目标 完成多账号微信小程序每天自动签到 输出 签到成功则向微信群发送签到成功的信息 否则提示用户签到失败,需手动签到 包管理 requests itchat time threading 程序的结构设计步骤1 获取要发送的json数据:地址 步骤2 向服务器发送请求 步骤3 根据服服务器响应判断签到是否完成 步骤4 微信交互 代码实现使用findler抓包工具查看请求类型 https://reserve.25team.com/wxappv1/yi/index?version=13 可以看

  • 微信小程序 自动登陆PHP源码实例(源码下载)

    微信小程序 自动登陆PHP源码实例 app.js 初始化APP自动登陆 您也可以在任何地方进行用户登陆验证 用法:首先在js文件中定义 var app = getApp(); app.getUserDataToken(); App({ onLaunch: function () { /*初始化APP自动登陆 * 您也可以在任何地方进行用户登陆验证 *用法:首先在js文件中定义 var app = getApp(); app.getUserDataToken(); */ this.getUserD

  • 微信小程序 Storage更新详解

    前言 1.近期项目开发中接触到微信小程序,其中有部分业务需要用到数据本地存储,而微信的官方文档中并没用提供直接更新Storage的API. 2.通过wx.setStorage(Object object)可以达到覆盖原有数据达到更新的目的,但是在只修改某个对象中某个属性的值中使用这个API总感觉有点蹩脚. 3.下面笔者简单封装了一个方法用来修改Storage,若是有人知道更好的方法或者我的写法有问题还望不吝赐教. 问题描述 小程序首次登录时需要输入账号密码,首次登录成功后将账号与token保存本

  • 微信小程序实现分类菜单激活状态随列表滚动而自动切换效果详解

    目录 view结构 js部分 属性解释 说明 注意 这里主要用到微信小程序提供的SelectorQuery获取页面节点信息实现,组件用的是微信小程序的scroll-view 逻辑就是获取右侧盒子的节点信息,获取右侧子分类的节点信息,当子分类滑动到顶部的之后,则切换左侧分类状态,而且当右侧子分类的位置处于触顶以及包含顶部位置的状态下,同样激活左侧分类状态. view结构 左侧父级分类 <scroll-view class="left" scroll-y > <view

随机推荐