91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
import aiohttp
|
|
from app.core.config import settings
|
|
import requests
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives import serialization
|
|
import json
|
|
import base64
|
|
import time
|
|
import random
|
|
import string
|
|
from cryptography.x509 import load_pem_x509_certificate
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
def generate_random_string(length=32):
|
|
"""生成指定长度的随机字符串"""
|
|
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
|
|
|
|
|
|
class WeChatClient:
|
|
"""微信客户端"""
|
|
|
|
def __init__(self):
|
|
self.appid = settings.WECHAT_APPID
|
|
self.secret = settings.WECHAT_SECRET
|
|
|
|
async def get_access_token(self):
|
|
"""获取小程序全局接口调用凭据"""
|
|
url = "https://api.weixin.qq.com/cgi-bin/token"
|
|
params = {
|
|
"grant_type": "client_credential",
|
|
"appid": self.appid,
|
|
"secret": self.secret
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, params=params) as response:
|
|
text = await response.text()
|
|
try:
|
|
result = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
raise Exception(f"解析微信返回数据失败: {text}")
|
|
|
|
if "errcode" in result and result["errcode"] != 0:
|
|
raise Exception(result.get("errmsg", "获取access_token失败"))
|
|
|
|
return result.get("access_token")
|
|
|
|
async def get_phone_number(self, code: str) -> dict:
|
|
"""获取用户手机号"""
|
|
access_token = await self.get_access_token()
|
|
url = f"https://api.weixin.qq.com/wxa/business/getuserphonenumber?access_token={access_token}"
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(url, json={"code": code}) as response:
|
|
text = await response.text()
|
|
try:
|
|
result = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
raise Exception(f"解析微信返回数据失败: {text}")
|
|
|
|
if result.get("errcode", 0) != 0:
|
|
raise Exception(result.get("errmsg", "获取手机号失败"))
|
|
|
|
return result.get("phone_info", {})
|
|
|
|
|
|
async def code2session(self, code: str) -> dict:
|
|
"""通过 code 获取用户 openid"""
|
|
url = f"https://api.weixin.qq.com/sns/jscode2session"
|
|
params = {
|
|
"appid": self.appid,
|
|
"secret": self.secret,
|
|
"js_code": code,
|
|
"grant_type": "authorization_code"
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, params=params) as response:
|
|
text = await response.text()
|
|
try:
|
|
result = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
raise Exception(f"解析微信返回数据失败: {text}")
|
|
|
|
if "errcode" in result and result["errcode"] != 0:
|
|
raise Exception(result.get("errmsg", "获取openid失败"))
|
|
|
|
return result
|
|
|
|
|