You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 
flyapps/fir_ser/api/utils/mp/wechat.py

151 lines
5.3 KiB

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# project: 9月
# author: NinEveN
# date: 2021/9/6
from hashlib import sha1
import requests
import logging
import json
from django.core.cache import cache
from fir_ser.settings import THIRDLOGINCONF, CACHE_KEY_TEMPLATE
from api.utils.mp.utils import WxMsgCryptBase
logger = logging.getLogger(__name__)
wx_login_info = THIRDLOGINCONF.wx_official
def format_req_json(j_data, func, *args, **kwargs):
if j_data.get("errcode") == 40001 or 'invalid credential' in j_data.get('errmsg', ''):
logger.error(f"error j_data {j_data}")
sync_wx_access_token(True)
return func(*args, **kwargs)[1]
return j_data
def sync_wx_access_token(force=False):
wx_access_token_key = CACHE_KEY_TEMPLATE.get("wx_access_token_key")
access_token_info = cache.get(wx_access_token_key)
if not access_token_info or force:
access_token_info = make_wx_auth_obj().get_access_token()
expires_in = access_token_info.get('expires_in')
if expires_in:
cache.set(wx_access_token_key, access_token_info, expires_in - 60)
return access_token_info
def get_wx_access_token_cache(c_count=1, ):
if c_count > 10:
return ''
wx_access_token_key = CACHE_KEY_TEMPLATE.get("wx_access_token_key")
access_token = cache.get(wx_access_token_key)
if access_token:
return access_token.get('access_token')
sync_wx_access_token(True)
return get_wx_access_token_cache(c_count + 1)
def create_menu():
menu_json = {
"button": [
{
"type": "click",
"name": "",
"key": "good"
},
{
"name": "分发平台",
"sub_button": [
{
"type": "view",
"name": "官方地址",
"url": "https://flyapps.cn"
},
{
"type": "view",
"name": "留言反馈",
"url": "https://flyapps.cn/gbook/"
},
]
},
{
"type": "media_id",
"name": "联系我们",
"media_id": "qvQxPuAb4GnUgjkxl2xVnbsnldxawf4DXM09biqgP30"
}
]
}
p_url = f"https://api.weixin.qq.com/cgi-bin/menu/create?access_token={get_wx_access_token_cache()}"
req = requests.post(url=p_url, data=json.dumps(menu_json, ensure_ascii=False).encode('utf-8'))
print(req.json())
def show_qrcode_url(ticket):
return f'https://mp.weixin.qq.com/cgi-bin/showqrcode?ticket={ticket}'
def make_wx_login_qrcode(scene_str='web.login', expire_seconds=600):
"""
:param scene_str: 场景值ID(字符串形式的ID),字符串类型,长度限制为1到64
:param expire_seconds: 该二维码有效时间,以秒为单位。 最大不超过2592000(即30天),此字段如果不填,则默认有效期为30秒。
:return: {
"ticket":"gQH47joAAAAAAAAAASxodHRwOi8vd2VpeGluLnFxLmNvbS9xL2taZ2Z3TVRtNzJXV1Brb3ZhYmJJAAIEZ23sUwMEmm3sUw==",
"expire_seconds":60,
"url":"http://weixin.qq.com/q/kZgfwMTm72WWPkovabbI"
}
https://developers.weixin.qq.com/doc/offiaccount/Account_Management/Generating_a_Parametric_QR_Code.html
"""
t_url = f'https://api.weixin.qq.com/cgi-bin/qrcode/create?access_token={get_wx_access_token_cache()}'
data = {"expire_seconds": expire_seconds, "action_name": "QR_STR_SCENE",
"action_info": {"scene": {"scene_str": scene_str}}}
req = requests.post(t_url, json=data)
if req.status_code == 200:
return True, format_req_json(req.json(), make_wx_login_qrcode, scene_str, expire_seconds)
logger.error(f"make wx login qrcode failed {req.status_code} {req.text}")
return False, req.text
def get_userinfo_from_openid(open_id):
t_url = f'https://api.weixin.qq.com/cgi-bin/user/info?access_token={get_wx_access_token_cache()}&openid={open_id}&lang=zh_CN'
req = requests.get(t_url)
if req.status_code == 200:
return True, format_req_json(req.json(), get_userinfo_from_openid, open_id)
logger.error(f"get userinfo from openid failed {req.status_code} {req.text}")
return False, req.text
class WxOfficialBase(object):
def __init__(self, app_id, app_secret, token, encoding_aes_key):
self.app_id = app_id
self.app_secret = app_secret
self.token = token
self.encoding_aes_key = encoding_aes_key
def get_access_token(self):
t_url = f'https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={self.app_id}&secret={self.app_secret}'
req = requests.get(t_url)
if req.status_code == 200:
return req.json()
logger.error(f"get access token failed {req.status_code} {req.text}")
return req.text
def make_wx_auth_obj():
return WxOfficialBase(**wx_login_info.get('auth'))
def check_signature(params):
tmp_list = sorted([wx_login_info.get('auth', {}).get('token'), params.get("timestamp"), params.get("nonce")])
tmp_str = "".join(tmp_list)
tmp_str = sha1(tmp_str.encode("utf-8")).hexdigest()
if tmp_str == params.get("signature"):
return int(params.get("echostr"))
return ''
class WxMsgCrypt(WxMsgCryptBase):
def __init__(self):
super().__init__(**wx_login_info.get('auth'))