deliveryman/deliveryman-api
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
aaron 2025-02-26 18:22:43 +08:00
parent 4debd58dd4
commit 442d6f0862
1 changed files with 47 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
app/core/wecomclient.py
View File

@ -1,47 +1,49 @@
from typing import Optional, Dict
import aiohttp import aiohttp
import logging
from app.core.config import settings from app.core.config import settings
import logging
from typing import Optional, Dict
class WecomClient: class WecomClient:
def __init__(self): def __init__(self):
self.corpid = settings.WECHAT_CORP_ID self.corpid = settings.WECHAT_CORP_ID
self.provider_secret = settings.WECHAT_CORP_SECRET self.secret = settings.WECHAT_CORP_SECRET # 注意：这里使用普通的 secret，不是 provider_secret
self.access_token = None self.access_token = None
async def get_provider_token(self) -> str: async def get_access_token(self) -> str:
"""获取服务商token""" """获取企业微信普通access_token"""
try: try:
url = "https://qyapi.weixin.qq.com/cgi-bin/service/get_provider_token" url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
data = { params = {
"corpid": self.corpid, "corpid": self.corpid,
"provider_secret": self.provider_secret "corpsecret": self.secret
} }
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
async with session.post(url, json=data) as response: async with session.get(url, params=params) as response:
result = await response.json() result = await response.json()
if result.get("errcode") == 0: if result.get("errcode") == 0:
self.access_token = result["provider_access_token"] self.access_token = result["access_token"]
return self.access_token return self.access_token
else: else:
raise Exception(f"获取provider_token失败: {result}") logging.error(f"获取access_token返回: {result}")
raise Exception(f"获取access_token失败: {result}")
except Exception as e: except Exception as e:
logging.error(f"获取provider_token失败: {str(e)}") logging.error(f"获取access_token失败: {str(e)}")
raise raise
async def miniprogram_to_userid(self, openid: str) -> Optional[Dict]: async def miniprogram_to_userid(self, code: str) -> Optional[Dict]:
""" """
小程序openid转换为企业微信userid 小程序code转换为企业微信userid
注意：这里使用code而不是openid
""" """
try: try:
url = "https://qyapi.weixin.qq.com/cgi-bin/service/miniprogram/jscode2session" url = "https://qyapi.weixin.qq.com/cgi-bin/miniprogram/jscode2session"
params = { params = {
"provider_access_token": await self.get_provider_token(), "access_token": await self.get_access_token(),
"appid": settings.WECHAT_APPID, "js_code": code,
"code": openid "grant_type": "authorization_code"
} }
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
@ -51,13 +53,37 @@ class WecomClient:
if result.get("errcode") == 0: if result.get("errcode") == 0:
data = { data = {
"userid": result.get("userid"), "userid": result.get("userid"),
"pending_id": result.get("pending_id") "pending_id": result.get("pending_id"),
"session_key": result.get("session_key")
} }
return data return data
else: else:
logging.error(f"openid转换失败: {result}") logging.error(f"code转换失败: {result}")
return None return None
except Exception as e: except Exception as e:
logging.error(f"openid转换失败: {str(e)}") logging.error(f"code转换失败: {str(e)}")
return None
async def get_user_info(self, userid: str) -> Optional[Dict]:
"""获取企业成员信息"""
try:
url = "https://qyapi.weixin.qq.com/cgi-bin/user/get"
params = {
"access_token": await self.get_access_token(),
"userid": userid
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
result = await response.json()
if result.get("errcode") == 0:
return result
else:
logging.error(f"获取用户信息失败: {result}")
return None
except Exception as e:
logging.error(f"获取用户信息失败: {str(e)}")
return None return None