sixGuoDjango/wechat_token/token工具包.py

"""

Author: xiao 792282@qq.com
Date: 2025-08-10 11:36:05
LastEditors: xiao 792282@qq.com
LastEditTime: 2025-08-17 13:38:39
FilePath: django_公共认证接口/微信TOKEN.py
Description: 这是默认设置,可以在设置》工具》File Description中进行配置

"""

import os
from zoneinfo import ZoneInfo

import django

os.environ.setdefault(key="DJANGO_SETTINGS_MODULE", value="sixGuoDjango.settings")
django.setup()
from wechat_token.token数据库 import weitoken

import jmespath
import pydash
import requests
from datetime import timedelta, datetime


class sixToken:
    """注释"""
    corp_id = "wwbe6369dc26a9d83a"  # 企业微信(当前的值为 江苏六果科技有限公司 企业微信ID)
    corpsecret = "SIRyum7vQh-Bt36AxnqKMX0HIiE-zkYbjYgSat7reZw"  # 企业微信 secret(重点,这里的值为 创建的应用的 元宝零售的 值) 各个应用都会有不同的值,需要注意分别的;
    通讯录Secret = "WNxZZdcU3FRS5msq_kHZaV021qxsZh9kCNExxXCbM9I"
    mini_appid = "wxf612e825377328f0"  # 小程序appid(当前的值为 元宝零售 江苏六果科技有限公司)
    mini_appsecret = "8691f774e3c0c5a876b21a3e84109ef0"  # 小程序属性
    # wechat_appid = "wx92441a2398fa8cc0"  # 服务号的appid
    wechat_appid = "wx654fffabd4b1f5cf"  # TODO 服务号的测试的
    # wechat_appsecret = "1c76fbfea91f36bede14e2cf2dc756dc"  #TODO 服务号的appsecret
    wechat_appsecret = "b1017a44b93fb08ae95b408fd19ff9d9"  # 服务号测试的appsecret
    
    @classmethod
    def get_wecom_token(cls):
        """"""
        print("每次都会被调用的wecom")
        查询对象, 是否存在 = weitoken.objects.get_or_create(token_name="wecom_token")
        if 查询对象.token_value is None:
            print("为空,get_wecom_token 新调用一下")
            创建的新对象 = cls.token统一调用企微获取接口()
            return 创建的新对象.token_value
        if 查询对象.token_value is not None:
            # 对时间进行比较 (这里有一个点: 从数据中取出的时间值为utc时间)
            nowDateTime = datetime.now(tz=ZoneInfo("utc"))
            if (nowDateTime - 查询对象.updateTime) > timedelta(seconds=6000):
                print("tokentoken超过时间了需要重新创建")
                创建的新对象 = cls.token统一调用企微获取接口()
                return 创建的新对象.token_value
            else:
                return 查询对象.token_value
        return 查询对象.token_value
    
    @classmethod
    def get_wecom_tick(cls):
        pass
        """"""
        查询对象, 是否存在 = weitoken.objects.get_or_create(token_name="wecom_token")
        if 查询对象.tick_value is None:
            print("为空,所以需要重 新调用一下")
            创建的新对象 = cls.token统一调用企微获取接口()
            return 创建的新对象.phone_token_value
        if 查询对象.tick_value is not None:
            # 对时间进行比较 (这里有一个点: 从数据中取出的时间值为utc时间)
            nowDateTime = datetime.now(tz=ZoneInfo("utc"))
            if (nowDateTime - 查询对象.updateTime) > timedelta(seconds=6000):
                print("token超过时间了需要重新创建")
                创建的新对象 = cls.token统一调用企微获取接口()
                return 创建的新对象.tick_value
            else:
                print("在有效时间内")
                return 查询对象.tick_value
        return 查询对象.tick_value
    
    @classmethod
    def mini_token(cls):
        """"""
        url = "https://api.weixin.qq.com/cgi-bin/stable_token"  # 固定请求地址官网请求的地址
        data = {"grant_type": "client_credential", "appid": cls.mini_appid, "secret": cls.mini_appsecret, "force_refresh": True}
        token_data = requests.post(url=url, json=data)
        token: dict = token_data.json()
        小程序token: str = jmespath.search(data=token, expression="access_token", )
        返回对象, _ = weitoken.objects.update_or_create(token_name="mini_token", defaults={"token_value": 小程序token})
        return 返回对象
    
    @classmethod
    def get_mini_token(cls):
        """
        实时的获取mini_token的值
        :return:
        :rtype:
        """
        查询对象, 是否存在 = weitoken.objects.get_or_create(token_name="mini_token")
        if 查询对象.token_value is None:
            创建的新对象 = cls.mini_token()
            return 创建的新对象.token_value
        if 查询对象.token_value is not None:
            # 对时间进行比较 (这里有一个点: 从数据中取出的时间值为utc时间)
            nowDateTime = datetime.now(tz=ZoneInfo("utc"))
            if (nowDateTime - 查询对象.updateTime) > timedelta(seconds=6000):
                print("token超过时间了需要重新创建")
                创建的新对象 = cls.mini_token()
                return 创建的新对象.token_value
            else:
                return 查询对象.token_value
        return None
    
    @classmethod
    def token统一调服务号(cls):
        """获取了六果服务号的token:注意不是授权时的token"""
        params = {"appid": cls.wechat_appid, "secret": cls.wechat_appsecret, "grant_type": "client_credential", "force_refresh": False, }
        token_data = requests.post(url="https://api.weixin.qq.com/cgi-bin/stable_token", json=params)
        token: dict = token_data.json()
        wechat_token服务号 = token.get("access_token")
        jsapo_url = f"https://api.weixin.qq.com/cgi-bin/ticket/getticket?access_token={wechat_token服务号}&type=jsapi"
        json_data = requests.get(url=jsapo_url).json()
        ticket服务号 = pydash.get(obj=json_data, path="ticket", default=None)
        返回对象, _ = weitoken.objects.update_or_create(token_name="mini_token", defaults={"token_value": wechat_token服务号, "tick_value": ticket服务号})
        return 返回对象
    
    @classmethod
    def get_wechat_token(cls):
        """"""
        
        查询对象, 是否存在 = weitoken.objects.get_or_create(token_name="wechat_token")
        if 查询对象.token_value is None:
            print("为空,所以需要重 新调用一下")
            创建的新对象 = cls.token统一调服务号()
            return 创建的新对象.token_value
        if 查询对象.token_value is not None:
            # 对时间进行比较 (这里有一个点: 从数据中取出的时间值为utc时间)
            nowDateTime = datetime.now(tz=ZoneInfo("utc"))
            if (nowDateTime - 查询对象.updateTime) > timedelta(seconds=6000):
                print("token超过时间了需要重新创建")
                创建的新对象 = cls.token统一调服务号()
                return 创建的新对象.token_value
            else:
                print("在有效时间内")
                return 查询对象.token_value
        return None
    
    @classmethod
    def get_wechat_tick_token(cls):
        查询对象, 是否存在 = weitoken.objects.get_or_create(token_name="wechat_token")
        if 查询对象.tick_value is None:
            print("为空,所以需要重 新调用一下")
            创建的新对象 = cls.token统一调服务号()
            return 创建的新对象.tick_value
        if 查询对象.tick_value is not None:
            # 对时间进行比较 (这里有一个点: 从数据中取出的时间值为utc时间)
            nowDateTime = datetime.now(tz=ZoneInfo("utc"))
            if (nowDateTime - 查询对象.updateTime) > timedelta(seconds=6000):
                print("token超过时间了需要重新创建")
                创建的新对象 = cls.token统一调服务号()
                return 创建的新对象.tick_value
            else:
                print("在有效时间内")
                return 查询对象.tick_value
        return None
    
    @classmethod
    def token统一调用企微获取接口(cls):
        """手机端的token"""
        phone_token_uri = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
        phone_token_data = requests.get(url=phone_token_uri, params={"corpid": "wwbe6369dc26a9d83a", "corpsecret": cls.通讯录Secret})
        phone_token = jmespath.search(data=phone_token_data.json(), expression="access_token")
        # 拿到企微的正常token
        wecom_token_uri = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
        # 发起请求
        token_data = requests.get(url=wecom_token_uri, params={"corpid": cls.corp_id, "corpsecret": cls.corpsecret, }, ).json()
        企微token = jmespath.search(data=token_data, expression="access_token", )
        print(企微token)
        # 调用发起 拿到wecom-tick-value
        wecom_tick_url = f"https://qyapi.weixin.qq.com/cgi-bin/get_jsapi_ticket?access_token={企微token}"
        tick_data = requests.get(url=wecom_tick_url)
        wecom_tick_data = jmespath.search(data=tick_data.json(), expression="ticket", )
        返回对象, _ = weitoken.objects.update_or_create(token_name="wecom_token", defaults={
            "phone_token_value": phone_token, "token_value": 企微token, "tick_value": wecom_tick_data,
            })
        return 返回对象
    
    @classmethod
    def get_weom_phone_token(cls):
        查询对象, 是否存在 = weitoken.objects.get_or_create(token_name="wecom_token")
        if 查询对象.phone_token_value is None:
            print("为空,所以需要重 新调用一下")
            创建的新对象 = cls.token统一调用企微获取接口()
            return 创建的新对象.phone_token_value
        if 查询对象.phone_token_value is not None:
            # 对时间进行比较 (这里有一个点: 从数据中取出的时间值为utc时间)
            nowDateTime = datetime.now(tz=ZoneInfo("utc"))
            if (nowDateTime - 查询对象.updateTime) > timedelta(seconds=6000):
                print("token超过时间了需要重新创建")
                创建的新对象 = cls.token统一调用企微获取接口()
                return 创建的新对象.phone_token_value
            else:
                print("在有效时间内")
                return 查询对象.phone_token_value
        return None
    
    @classmethod
    def 服务号网页授权TOKEN(cls, code: str):
        """用户授权时,需要的token;
        https://developers.weixin.qq.com/doc/offiaccount/OA_Web_Apps/Wechat_webpage_authorization.html
        """
        params = {
            "appid": "wx92441a2398fa8cc0",  # 公众号的唯一标识
            "secret": "1c76fbfea91f36bede14e2cf2dc756dc",  # 公众号的appsecret
            "grant_type": "authorization_code",  # "code": code
            "code": code,  # 填写第一步获取的code参数
            }
        
        oauth_token_url = f"https://api.weixin.qq.com/sns/oauth2/access_token"
        oauth_token_data = requests.get(url=oauth_token_url, params=params)
        # oauth_token: dict = oauth_token_data.json()
        # oauth_token_str = oauth_token.get("access_token")
        # refresh_token = oauth_token.get("refresh_token")
        # unionid = oauth_token.get("unionid")


if __name__ == "__main__":
    pass
    # mini_token = sixToken.get_mini_token()
    # wecom_token = sixToken.get_wecom_token()
    sixToken.token统一调用企微获取接口()
    # wecom_phone_token = sixToken.get_weom_phone_token()
    # print(wecom_phone_token)