69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
# -*- coding = utf-8 -*-
|
|
# @Time : 2025/2/17 下午4:49
|
|
# @File : api.py
|
|
# @Software : PyCharm
|
|
# @Author : xingc
|
|
# @Desc :
|
|
import time
|
|
|
|
import httpx
|
|
import ujson
|
|
|
|
from core.db.engine import get_cache
|
|
|
|
|
|
class WechatApi:
|
|
def __init__(self, appid, secret):
|
|
self.appid = appid
|
|
self.secret = secret
|
|
self._access_token = None
|
|
self._access_token_expire_time = None
|
|
self.session = httpx.AsyncClient()
|
|
|
|
@property
|
|
async def access_token(self):
|
|
if any([
|
|
self._access_token is None,
|
|
self._access_token_expire_time < time.time(),
|
|
]):
|
|
await self.get_access_token()
|
|
return self._access_token
|
|
|
|
async def get_access_token(self):
|
|
async with get_cache() as redis_conn:
|
|
result_str = await redis_conn.get('access_token')
|
|
|
|
if result_str:
|
|
access_token, expire_time = result_str.split('|')
|
|
self._access_token = access_token
|
|
self._access_token_expire_time = int(expire_time)
|
|
else:
|
|
url = 'https://api.weixin.qq.com/cgi-bin/token'
|
|
params = {
|
|
'appid': self.appid,
|
|
'secret': self.secret,
|
|
'grant_type': 'client_credential'
|
|
}
|
|
r = await self.session.get(url, params=params)
|
|
result = r.json()
|
|
if 'errcode' not in result_str:
|
|
expires_in = result['expires_in']
|
|
self._access_token_expire_time = time.time() + expires_in
|
|
self._access_token = result['access_token']
|
|
await redis_conn.set(
|
|
'access_token', f'{self._access_token}|{self._access_token_expire_time}', ex=expires_in
|
|
)
|
|
|
|
return self._access_token
|
|
|
|
async def code_to_session(self, code: str):
|
|
url = 'https://api.weixin.qq.com/sns/jscode2session'
|
|
params = {
|
|
'appid': self.appid,
|
|
'secret': self.secret,
|
|
'js_code': code,
|
|
'grant_type': 'authorization_code'
|
|
}
|
|
r = await self.session.get(url, params=params)
|
|
return r.json()
|