54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
import hashlib
|
||
import json
|
||
import time
|
||
import uuid
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
|
||
from config import APP_ID, TOKEN
|
||
|
||
|
||
TOKEN_URL = "https://openaiapi.weixin.qq.com/v2/token"
|
||
|
||
|
||
def _md5_hex(data: str) -> str:
|
||
return hashlib.md5(data.encode("utf-8")).hexdigest()
|
||
|
||
|
||
async def fetch_access_token(account: Optional[str] = None) -> str:
|
||
"""
|
||
调用微信智能对话 /v2/token 接口，换取 AccessToken。
|
||
文档参考：https://developers.weixin.qq.com/doc/aispeech/confapi/dialog/token.html
|
||
"""
|
||
timestamp = int(time.time())
|
||
nonce = uuid.uuid4().hex[:16]
|
||
|
||
body_obj = {}
|
||
if account:
|
||
body_obj["account"] = account
|
||
body_str = json.dumps(body_obj, separators=(",", ":"), ensure_ascii=False)
|
||
|
||
# sign = md5(Token + str(unix_timestamp) + nonce + md5(body))
|
||
body_md5 = _md5_hex(body_str) if body_str else _md5_hex("")
|
||
sign_src = f"{TOKEN}{timestamp}{nonce}{body_md5}"
|
||
sign = _md5_hex(sign_src)
|
||
|
||
headers = {
|
||
"X-APPID": APP_ID,
|
||
"request_id": str(uuid.uuid4()),
|
||
"timestamp": str(timestamp),
|
||
"nonce": nonce,
|
||
"sign": sign,
|
||
"content-type": "application/json",
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=5) as client:
|
||
resp = await client.post(TOKEN_URL, headers=headers, content=body_str.encode("utf-8"))
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(f"fetch_access_token error: {data}")
|
||
return data["data"]["access_token"]
|
||
|