Codex
1284 lines
49 KiB
Python
1284 lines
49 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
OAuth Service - OAuth 业务逻辑层
|
||
处理第三方 OAuth 登录逻辑（异步版本）
|
||
"""
|
||
import json
|
||
import logging
|
||
import re
|
||
from typing import Dict, Optional, Tuple
|
||
|
||
import httpx
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import settings
|
||
from core.oauth.base_oauth_service import BaseOAuthService
|
||
from core.user.model import User
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class GiteeOAuthService(BaseOAuthService):
|
||
"""Gitee OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'gitee'
|
||
AUTHORIZE_URL = "https://gitee.com/oauth/authorize"
|
||
TOKEN_URL = "https://gitee.com/oauth/token"
|
||
USER_INFO_URL = "https://gitee.com/api/v5/user"
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取 Gitee 客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'GITEE_CLIENT_ID', ''),
|
||
'client_secret': getattr(settings, 'GITEE_CLIENT_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'GITEE_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取 Gitee OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_gitee")
|
||
return {
|
||
'client_id': oauth_config.get("client_id") or getattr(settings, 'GITEE_CLIENT_ID', '') or '',
|
||
'client_secret': oauth_config.get("client_secret") or getattr(settings, 'GITEE_CLIENT_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'GITEE_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""
|
||
使用访问令牌获取 Gitee 用户信息
|
||
"""
|
||
try:
|
||
params = {'access_token': access_token}
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, params=params)
|
||
response.raise_for_status()
|
||
|
||
user_info = response.json()
|
||
|
||
if 'id' not in user_info:
|
||
logger.error(f"Gitee 用户信息格式错误: {user_info}")
|
||
return None
|
||
|
||
return user_info
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求 Gitee 用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取 Gitee 用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化 Gitee 用户信息"""
|
||
return {
|
||
'provider_id': str(raw_user_info.get('id')),
|
||
'username': raw_user_info.get('login'),
|
||
'name': raw_user_info.get('name', raw_user_info.get('login')),
|
||
'email': raw_user_info.get('email'),
|
||
'avatar': raw_user_info.get('avatar_url'),
|
||
'bio': raw_user_info.get('bio'),
|
||
}
|
||
|
||
|
||
class GitHubOAuthService(BaseOAuthService):
|
||
"""GitHub OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'github'
|
||
AUTHORIZE_URL = "https://github.com/login/oauth/authorize"
|
||
TOKEN_URL = "https://github.com/login/oauth/access_token"
|
||
USER_INFO_URL = "https://api.github.com/user"
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取 GitHub 客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'GITHUB_CLIENT_ID', ''),
|
||
'client_secret': getattr(settings, 'GITHUB_CLIENT_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'GITHUB_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取 GitHub OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_github")
|
||
return {
|
||
'client_id': oauth_config.get("client_id") or getattr(settings, 'GITHUB_CLIENT_ID', '') or '',
|
||
'client_secret': oauth_config.get("client_secret") or getattr(settings, 'GITHUB_CLIENT_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'GITHUB_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
def get_extra_authorize_params(cls) -> Dict[str, str]:
|
||
"""GitHub 需要 scope 参数"""
|
||
return {
|
||
'scope': 'user:email',
|
||
}
|
||
|
||
@classmethod
|
||
def get_token_request_headers(cls) -> Dict[str, str]:
|
||
"""GitHub 需要 Accept header 来获取 JSON 响应"""
|
||
return {
|
||
'Accept': 'application/json',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""使用访问令牌获取 GitHub 用户信息"""
|
||
try:
|
||
headers = {
|
||
'Authorization': f'Bearer {access_token}',
|
||
'Accept': 'application/json',
|
||
}
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
user_info = response.json()
|
||
|
||
if 'id' not in user_info:
|
||
logger.error(f"GitHub 用户信息格式错误: {user_info}")
|
||
return None
|
||
|
||
return user_info
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求 GitHub 用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取 GitHub 用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化 GitHub 用户信息"""
|
||
return {
|
||
'provider_id': str(raw_user_info.get('id')),
|
||
'username': raw_user_info.get('login'),
|
||
'name': raw_user_info.get('name') or raw_user_info.get('login'),
|
||
'email': raw_user_info.get('email'),
|
||
'avatar': raw_user_info.get('avatar_url'),
|
||
'bio': raw_user_info.get('bio'),
|
||
}
|
||
|
||
|
||
class QQOAuthService(BaseOAuthService):
|
||
"""QQ 互联 OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'qq'
|
||
AUTHORIZE_URL = "https://graph.qq.com/oauth2.0/authorize"
|
||
TOKEN_URL = "https://graph.qq.com/oauth2.0/token"
|
||
USER_INFO_URL = "https://graph.qq.com/user/get_user_info"
|
||
OPENID_URL = "https://graph.qq.com/oauth2.0/me"
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取 QQ 客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'QQ_APP_ID', ''),
|
||
'client_secret': getattr(settings, 'QQ_APP_KEY', ''),
|
||
'redirect_uri': getattr(settings, 'QQ_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取 QQ OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_qq")
|
||
return {
|
||
'client_id': oauth_config.get("app_id") or getattr(settings, 'QQ_APP_ID', '') or '',
|
||
'client_secret': oauth_config.get("app_key") or getattr(settings, 'QQ_APP_KEY', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'QQ_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
def get_extra_authorize_params(cls) -> Dict[str, str]:
|
||
"""QQ 需要 response_type 参数"""
|
||
return {
|
||
'response_type': 'code',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_access_token(cls, code: str, redirect_uri: Optional[str] = None) -> Optional[str]:
|
||
"""使用授权码获取访问令牌（QQ 返回 URL 参数格式）"""
|
||
try:
|
||
config = await cls.get_client_config_async()
|
||
|
||
params = {
|
||
'grant_type': 'authorization_code',
|
||
'client_id': config['client_id'],
|
||
'client_secret': config['client_secret'],
|
||
'code': code,
|
||
'redirect_uri': redirect_uri or config['redirect_uri'],
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.TOKEN_URL, params=params)
|
||
response.raise_for_status()
|
||
|
||
# QQ 返回的是 URL 参数格式: access_token=xxx&expires_in=xxx
|
||
response_text = response.text
|
||
|
||
match = re.search(r'access_token=([^&]+)', response_text)
|
||
if match:
|
||
access_token = match.group(1)
|
||
logger.info(f"QQ access_token 获取成功")
|
||
return access_token
|
||
else:
|
||
logger.error(f"QQ access_token 解析失败: {response_text}")
|
||
return None
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求 QQ access_token 失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取 QQ access_token 异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""使用访问令牌获取 QQ 用户信息"""
|
||
try:
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
# 1. 获取 openid
|
||
openid_response = await client.get(
|
||
cls.OPENID_URL,
|
||
params={'access_token': access_token}
|
||
)
|
||
openid_response.raise_for_status()
|
||
|
||
# QQ 返回的是 JSONP 格式: callback( {"client_id":"xxx","openid":"xxx"} );
|
||
openid_text = openid_response.text
|
||
|
||
match = re.search(r'callback\(\s*(\{.*?\})\s*\)', openid_text)
|
||
if not match:
|
||
logger.error(f"QQ openid 解析失败: {openid_text}")
|
||
return None
|
||
|
||
openid_data = json.loads(match.group(1))
|
||
openid = openid_data.get('openid')
|
||
|
||
if not openid:
|
||
logger.error(f"QQ openid 不存在: {openid_data}")
|
||
return None
|
||
|
||
logger.info(f"QQ openid 获取成功: {openid}")
|
||
|
||
# 2. 获取用户信息
|
||
config = await cls.get_client_config_async()
|
||
user_response = await client.get(
|
||
cls.USER_INFO_URL,
|
||
params={
|
||
'access_token': access_token,
|
||
'oauth_consumer_key': config['client_id'],
|
||
'openid': openid
|
||
}
|
||
)
|
||
user_response.raise_for_status()
|
||
|
||
user_info = user_response.json()
|
||
|
||
if user_info.get('ret') != 0:
|
||
logger.error(f"QQ 用户信息获取失败: {user_info.get('msg')}")
|
||
return None
|
||
|
||
user_info['openid'] = openid
|
||
return user_info
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求 QQ 用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取 QQ 用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化 QQ 用户信息"""
|
||
return {
|
||
'provider_id': raw_user_info.get('openid'),
|
||
'username': raw_user_info.get('nickname', '').replace(' ', '_'),
|
||
'name': raw_user_info.get('nickname'),
|
||
'email': None,
|
||
'avatar': raw_user_info.get('figureurl_qq_2') or raw_user_info.get('figureurl_qq_1'),
|
||
'bio': None,
|
||
}
|
||
|
||
|
||
class GoogleOAuthService(BaseOAuthService):
|
||
"""Google OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'google'
|
||
AUTHORIZE_URL = "https://accounts.google.com/o/oauth2/v2/auth"
|
||
TOKEN_URL = "https://oauth2.googleapis.com/token"
|
||
USER_INFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取 Google 客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'GOOGLE_CLIENT_ID', ''),
|
||
'client_secret': getattr(settings, 'GOOGLE_CLIENT_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'GOOGLE_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取 Google OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_google")
|
||
return {
|
||
'client_id': oauth_config.get("client_id") or getattr(settings, 'GOOGLE_CLIENT_ID', '') or '',
|
||
'client_secret': oauth_config.get("client_secret") or getattr(settings, 'GOOGLE_CLIENT_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'GOOGLE_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
def get_extra_authorize_params(cls) -> Dict[str, str]:
|
||
"""Google 需要 scope 和 access_type 参数"""
|
||
return {
|
||
'scope': 'openid email profile',
|
||
'access_type': 'offline',
|
||
'response_type': 'code',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""使用访问令牌获取 Google 用户信息"""
|
||
try:
|
||
headers = {
|
||
'Authorization': f'Bearer {access_token}',
|
||
}
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
user_info = response.json()
|
||
|
||
if 'id' not in user_info:
|
||
logger.error(f"Google 用户信息格式错误: {user_info}")
|
||
return None
|
||
|
||
return user_info
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求 Google 用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取 Google 用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化 Google 用户信息"""
|
||
return {
|
||
'provider_id': raw_user_info.get('id'),
|
||
'username': raw_user_info.get('email', '').split('@')[0],
|
||
'name': raw_user_info.get('name') or raw_user_info.get('email'),
|
||
'email': raw_user_info.get('email'),
|
||
'avatar': raw_user_info.get('picture'),
|
||
'bio': None,
|
||
}
|
||
|
||
|
||
class WeChatOAuthService(BaseOAuthService):
|
||
"""微信开放平台 OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'wechat'
|
||
AUTHORIZE_URL = "https://open.weixin.qq.com/connect/qrconnect"
|
||
TOKEN_URL = "https://api.weixin.qq.com/sns/oauth2/access_token"
|
||
USER_INFO_URL = "https://api.weixin.qq.com/sns/userinfo"
|
||
|
||
@classmethod
|
||
def get_user_id_field(cls) -> str:
|
||
"""微信使用 unionid 作为唯一标识"""
|
||
return 'wechat_unionid'
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取微信客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'WECHAT_APP_ID', ''),
|
||
'client_secret': getattr(settings, 'WECHAT_APP_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'WECHAT_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取微信 OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_wechat")
|
||
return {
|
||
'client_id': oauth_config.get("app_id") or getattr(settings, 'WECHAT_APP_ID', '') or '',
|
||
'client_secret': oauth_config.get("app_secret") or getattr(settings, 'WECHAT_APP_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'WECHAT_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_authorize_url(cls, state: str = None, redirect_uri: str = None) -> str:
|
||
"""获取微信授权 URL（微信参数名称与标准 OAuth 2.0 不同）"""
|
||
config = await cls.get_client_config_async()
|
||
|
||
params = {
|
||
'appid': config['client_id'],
|
||
'redirect_uri': redirect_uri or config['redirect_uri'],
|
||
'response_type': 'code',
|
||
'scope': 'snsapi_login',
|
||
}
|
||
|
||
if state:
|
||
params['state'] = state
|
||
|
||
query_string = '&'.join([f"{k}={v}" for k, v in params.items()])
|
||
return f"{cls.AUTHORIZE_URL}?{query_string}#wechat_redirect"
|
||
|
||
@classmethod
|
||
async def get_access_token(cls, code: str) -> Optional[Dict]:
|
||
"""使用授权码获取访问令牌（微信返回 access_token 和 openid）"""
|
||
try:
|
||
config = await cls.get_client_config_async()
|
||
params = {
|
||
'appid': config['client_id'],
|
||
'secret': config['client_secret'],
|
||
'code': code,
|
||
'grant_type': 'authorization_code',
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.TOKEN_URL, params=params)
|
||
response.raise_for_status()
|
||
|
||
token_data = response.json()
|
||
|
||
if 'errcode' in token_data:
|
||
logger.error(f"微信获取 token 失败: {token_data}")
|
||
return None
|
||
|
||
if 'access_token' not in token_data or 'openid' not in token_data:
|
||
logger.error(f"微信 token 响应格式错误: {token_data}")
|
||
return None
|
||
|
||
return token_data
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求微信 token 失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取微信 token 异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str, openid: str = None) -> Optional[Dict]:
|
||
"""使用访问令牌获取微信用户信息"""
|
||
try:
|
||
params = {
|
||
'access_token': access_token,
|
||
'openid': openid,
|
||
'lang': 'zh_CN',
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, params=params)
|
||
response.raise_for_status()
|
||
|
||
user_info = response.json()
|
||
|
||
if 'errcode' in user_info:
|
||
logger.error(f"微信获取用户信息失败: {user_info}")
|
||
return None
|
||
|
||
if 'openid' not in user_info:
|
||
logger.error(f"微信用户信息格式错误: {user_info}")
|
||
return None
|
||
|
||
return user_info
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求微信用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取微信用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化微信用户信息"""
|
||
provider_id = raw_user_info.get('unionid') or raw_user_info.get('openid')
|
||
nickname = raw_user_info.get('nickname', '')
|
||
username = nickname.replace(' ', '_')[:30] if nickname else f"wechat_{provider_id[:8]}"
|
||
|
||
return {
|
||
'provider_id': provider_id,
|
||
'username': username,
|
||
'name': nickname or username,
|
||
'email': None,
|
||
'avatar': raw_user_info.get('headimgurl'),
|
||
'bio': None,
|
||
}
|
||
|
||
@classmethod
|
||
async def handle_oauth_login(
|
||
cls,
|
||
db: AsyncSession,
|
||
code: str,
|
||
ip_address: str,
|
||
user_agent: str = None,
|
||
login_type: str = None,
|
||
device_id: str = None,
|
||
redirect_uri: str = None,
|
||
) -> Tuple[User, str, str, int]:
|
||
"""
|
||
处理微信 OAuth 登录流程（重写基类方法）
|
||
|
||
微信的 get_access_token 返回 Dict（含 access_token 和 openid），
|
||
且 get_user_info 需要同时传递 access_token 和 openid，
|
||
因此需要重写此方法以正确处理。
|
||
"""
|
||
# 1. 使用 code 换取 token_data（Dict，含 access_token + openid）
|
||
token_data = await cls.get_access_token(code)
|
||
if not token_data:
|
||
raise ValueError(f"获取 {cls.PROVIDER_NAME} 访问令牌失败")
|
||
|
||
access_token = token_data.get('access_token')
|
||
openid = token_data.get('openid')
|
||
|
||
if not access_token or not openid:
|
||
raise ValueError(f"获取 {cls.PROVIDER_NAME} 访问令牌格式错误")
|
||
|
||
# 2. 使用 access_token + openid 获取用户信息
|
||
raw_user_info = await cls.get_user_info(access_token, openid=openid)
|
||
if not raw_user_info:
|
||
raise ValueError(f"获取 {cls.PROVIDER_NAME} 用户信息失败")
|
||
|
||
# 3. 后续流程与基类一致，调用基类的通用逻辑
|
||
# 先标准化用户信息，然后手动执行基类 handle_oauth_login 中第222行之后的逻辑
|
||
# 为避免代码重复，将 access_token(str) 设置到类属性后调用基类
|
||
# 但基类的 handle_oauth_login 会重新调用 get_access_token，所以这里直接复用基类的用户处理逻辑
|
||
return await cls._handle_user_login(
|
||
db=db,
|
||
raw_user_info=raw_user_info,
|
||
ip_address=ip_address,
|
||
user_agent=user_agent,
|
||
login_type=login_type,
|
||
device_id=device_id,
|
||
)
|
||
|
||
@classmethod
|
||
async def _handle_user_login(
|
||
cls,
|
||
db: AsyncSession,
|
||
raw_user_info: Dict,
|
||
ip_address: str,
|
||
user_agent: str = None,
|
||
login_type: str = None,
|
||
device_id: str = None,
|
||
) -> Tuple[User, str, str, int]:
|
||
"""微信专用：从已获取的用户信息开始执行登录流程（复用基类逻辑）"""
|
||
from datetime import timedelta
|
||
from sqlalchemy import select
|
||
from utils.redis import RedisClient
|
||
from core.login_log.service import LoginLogService
|
||
from utils.security import create_access_token, create_refresh_token
|
||
from utils.user_info_cache import set_cached_user_info
|
||
from core.user.service import UserService
|
||
from core.oauth.base_oauth_service import REFRESH_TOKEN_PREFIX
|
||
|
||
# 标准化用户信息
|
||
user_info = cls.normalize_user_info(raw_user_info)
|
||
provider_id = user_info['provider_id']
|
||
username = user_info['username']
|
||
name = user_info['name']
|
||
email = user_info.get('email')
|
||
avatar = user_info.get('avatar')
|
||
bio = user_info.get('bio')
|
||
|
||
# 查找或创建用户
|
||
user_id_field = cls.get_user_id_field()
|
||
stmt = select(User).where(
|
||
getattr(User, user_id_field) == provider_id,
|
||
User.is_deleted == False
|
||
)
|
||
result = await db.execute(stmt)
|
||
user = result.scalar_one_or_none()
|
||
|
||
is_superadmin = getattr(settings, 'GRANT_ADMIN_TO_OAUTH_USER', False)
|
||
default_dept_id = getattr(settings, 'OAUTH_DEFAULT_DEPT_ID', None)
|
||
|
||
if user:
|
||
logger.info(f"{cls.PROVIDER_NAME} 用户已存在: {username} (ID: {provider_id})")
|
||
if email and not user.email:
|
||
user.email = email
|
||
if bio and not user.bio:
|
||
user.bio = bio
|
||
db.add(user)
|
||
await db.flush()
|
||
else:
|
||
logger.info(f"创建新的 {cls.PROVIDER_NAME} 用户: {username} (ID: {provider_id})")
|
||
unique_username = username
|
||
counter = 1
|
||
while True:
|
||
stmt = select(User).where(User.username == unique_username)
|
||
result = await db.execute(stmt)
|
||
if result.scalar_one_or_none() is None:
|
||
break
|
||
unique_username = f"{username}_{counter}"
|
||
counter += 1
|
||
|
||
create_kwargs = {
|
||
'username': unique_username,
|
||
'name': name,
|
||
'email': email,
|
||
'bio': bio,
|
||
user_id_field: provider_id,
|
||
'oauth_provider': cls.PROVIDER_NAME,
|
||
'user_type': 1,
|
||
'user_status': 1,
|
||
'is_active': True,
|
||
'is_superuser': is_superadmin,
|
||
'dept_id': default_dept_id,
|
||
}
|
||
user = User(**create_kwargs)
|
||
db.add(user)
|
||
await db.flush()
|
||
await db.refresh(user)
|
||
logger.info(f"{cls.PROVIDER_NAME} 用户创建成功: {unique_username}")
|
||
|
||
if login_type:
|
||
user.last_login_type = login_type
|
||
db.add(user)
|
||
await db.flush()
|
||
|
||
await db.commit()
|
||
await db.refresh(user)
|
||
|
||
if not user.is_active:
|
||
raise ValueError("账户已被禁用")
|
||
if user.user_status == 0:
|
||
raise ValueError("账户已被禁用")
|
||
if user.user_status == 2:
|
||
raise ValueError("账户已被锁定，请联系管理员")
|
||
|
||
# 生成 JWT token
|
||
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
||
refresh_token_expires = timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
|
||
token_data = {"sub": user.id, "username": user.username}
|
||
jwt_access_token = create_access_token(token_data, access_token_expires, device_id=device_id)
|
||
jwt_refresh_token = create_refresh_token(token_data, refresh_token_expires, device_id=device_id)
|
||
|
||
# 缓存用户信息到 Redis
|
||
role_ids = await UserService.get_user_role_ids(db, user.id)
|
||
await set_cached_user_info(user.id, role_ids, user.dept_id, user.is_superuser)
|
||
expire_time = settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
|
||
|
||
redis = await RedisClient.get_client()
|
||
|
||
if not settings.ALLOW_MULTI_DEVICE_LOGIN:
|
||
refresh_pattern = f"{REFRESH_TOKEN_PREFIX}{user.id}:*"
|
||
access_pattern = f"access_token:{user.id}:*"
|
||
cursor = 0
|
||
while True:
|
||
cursor, keys = await redis.scan(cursor, match=refresh_pattern, count=100)
|
||
if keys:
|
||
await redis.delete(*keys)
|
||
if cursor == 0:
|
||
break
|
||
cursor = 0
|
||
while True:
|
||
cursor, keys = await redis.scan(cursor, match=access_pattern, count=100)
|
||
if keys:
|
||
await redis.delete(*keys)
|
||
if cursor == 0:
|
||
break
|
||
|
||
redis_key = f"{REFRESH_TOKEN_PREFIX}{user.id}:{device_id}" if device_id else f"{REFRESH_TOKEN_PREFIX}{user.id}"
|
||
await redis.set(redis_key, jwt_refresh_token, ex=int(refresh_token_expires.total_seconds()))
|
||
|
||
if device_id:
|
||
await redis.set(
|
||
f"access_token:{user.id}:{device_id}",
|
||
jwt_access_token,
|
||
ex=int(access_token_expires.total_seconds())
|
||
)
|
||
|
||
await LoginLogService.record_login(
|
||
db=db,
|
||
username=user.username,
|
||
user_id=str(user.id),
|
||
status=1,
|
||
login_ip=ip_address,
|
||
user_agent=user_agent,
|
||
login_type=login_type or cls.PROVIDER_NAME,
|
||
)
|
||
|
||
return user, jwt_access_token, jwt_refresh_token, expire_time
|
||
|
||
|
||
class MicrosoftOAuthService(BaseOAuthService):
|
||
"""微软 OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'microsoft'
|
||
AUTHORIZE_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/authorize"
|
||
TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
|
||
USER_INFO_URL = "https://graph.microsoft.com/v1.0/me"
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取微软客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'MICROSOFT_CLIENT_ID', ''),
|
||
'client_secret': getattr(settings, 'MICROSOFT_CLIENT_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'MICROSOFT_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取微软 OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_microsoft")
|
||
return {
|
||
'client_id': oauth_config.get("client_id") or getattr(settings, 'MICROSOFT_CLIENT_ID', '') or '',
|
||
'client_secret': oauth_config.get("client_secret") or getattr(settings, 'MICROSOFT_CLIENT_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'MICROSOFT_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
def get_extra_authorize_params(cls) -> Dict[str, str]:
|
||
"""微软需要 scope 和 response_mode 参数"""
|
||
return {
|
||
'scope': 'openid email profile User.Read',
|
||
'response_type': 'code',
|
||
'response_mode': 'query',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""使用 Microsoft Graph API 获取用户信息"""
|
||
try:
|
||
headers = {
|
||
'Authorization': f'Bearer {access_token}',
|
||
}
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
user_info = response.json()
|
||
|
||
if 'id' not in user_info:
|
||
logger.error(f"Microsoft 用户信息格式错误: {user_info}")
|
||
return None
|
||
|
||
return user_info
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求 Microsoft 用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取 Microsoft 用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化微软用户信息"""
|
||
user_principal_name = raw_user_info.get('userPrincipalName', '')
|
||
username = user_principal_name.split('@')[0] if '@' in user_principal_name else user_principal_name
|
||
email = raw_user_info.get('mail') or raw_user_info.get('userPrincipalName')
|
||
|
||
return {
|
||
'provider_id': raw_user_info.get('id'),
|
||
'username': username or f"ms_{raw_user_info.get('id', '')[:8]}",
|
||
'name': raw_user_info.get('displayName') or username,
|
||
'email': email,
|
||
'avatar': None,
|
||
'bio': raw_user_info.get('jobTitle'),
|
||
}
|
||
|
||
|
||
class DingTalkOAuthService(BaseOAuthService):
|
||
"""钉钉 OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'dingtalk'
|
||
AUTHORIZE_URL = "https://login.dingtalk.com/oauth2/auth"
|
||
TOKEN_URL = "https://api.dingtalk.com/v1.0/oauth2/userAccessToken"
|
||
USER_INFO_URL = "https://api.dingtalk.com/v1.0/contact/users/me"
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取钉钉客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'DINGTALK_APP_ID', ''),
|
||
'client_secret': getattr(settings, 'DINGTALK_APP_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'DINGTALK_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取钉钉 OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_dingtalk")
|
||
return {
|
||
'client_id': oauth_config.get("app_id") or getattr(settings, 'DINGTALK_APP_ID', '') or '',
|
||
'client_secret': oauth_config.get("app_secret") or getattr(settings, 'DINGTALK_APP_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'DINGTALK_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
def get_extra_authorize_params(cls) -> Dict[str, str]:
|
||
"""钉钉需要的额外授权参数"""
|
||
return {
|
||
'response_type': 'code',
|
||
'scope': 'openid',
|
||
'prompt': 'consent',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_access_token(cls, code: str) -> Optional[str]:
|
||
"""使用授权码获取访问令牌（钉钉使用 JSON body）"""
|
||
try:
|
||
config = await cls.get_client_config_async()
|
||
|
||
data = {
|
||
'clientId': config['client_id'],
|
||
'clientSecret': config['client_secret'],
|
||
'code': code,
|
||
'grantType': 'authorization_code',
|
||
}
|
||
|
||
headers = {
|
||
'Content-Type': 'application/json',
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.post(cls.TOKEN_URL, json=data, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
|
||
if 'accessToken' in result:
|
||
logger.info(f"钉钉 access_token 获取成功")
|
||
return result['accessToken']
|
||
else:
|
||
logger.error(f"钉钉 token 响应格式错误: {result}")
|
||
return None
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求钉钉 access_token 失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取钉钉 access_token 异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""使用访问令牌获取钉钉用户信息"""
|
||
try:
|
||
headers = {
|
||
'x-acs-dingtalk-access-token': access_token,
|
||
'Content-Type': 'application/json',
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
user_info = response.json()
|
||
|
||
if 'unionId' not in user_info:
|
||
logger.error(f"钉钉用户信息格式错误: {user_info}")
|
||
return None
|
||
|
||
return user_info
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求钉钉用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取钉钉用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化钉钉用户信息"""
|
||
provider_id = raw_user_info.get('unionId', '')
|
||
nick = raw_user_info.get('nick', '')
|
||
username = nick if nick else f"dingtalk_{provider_id[:8]}"
|
||
|
||
return {
|
||
'provider_id': provider_id,
|
||
'username': username,
|
||
'name': nick or username,
|
||
'email': raw_user_info.get('email'),
|
||
'avatar': raw_user_info.get('avatarUrl'),
|
||
'mobile': raw_user_info.get('mobile'),
|
||
'bio': f"钉钉用户 - {nick}" if nick else "钉钉用户",
|
||
}
|
||
|
||
@classmethod
|
||
def get_user_id_field(cls) -> str:
|
||
"""获取用户 ID 字段名"""
|
||
return 'dingtalk_unionid'
|
||
|
||
|
||
|
||
class FeishuOAuthService(BaseOAuthService):
|
||
"""飞书 OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'feishu'
|
||
AUTHORIZE_URL = "https://open.feishu.cn/open-apis/authen/v1/authorize"
|
||
TOKEN_URL = "https://open.feishu.cn/open-apis/authen/v1/oidc/access_token"
|
||
USER_INFO_URL = "https://open.feishu.cn/open-apis/authen/v1/user_info"
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取飞书客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'FEISHU_APP_ID', ''),
|
||
'client_secret': getattr(settings, 'FEISHU_APP_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'FEISHU_REDIRECT_URI', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取飞书 OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_feishu")
|
||
return {
|
||
'client_id': oauth_config.get("app_id") or getattr(settings, 'FEISHU_APP_ID', '') or '',
|
||
'client_secret': oauth_config.get("app_secret") or getattr(settings, 'FEISHU_APP_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'FEISHU_REDIRECT_URI', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
def get_extra_authorize_params(cls) -> Dict[str, str]:
|
||
"""飞书需要的额外授权参数"""
|
||
return {
|
||
'response_type': 'code',
|
||
'scope': 'contact:user.base:readonly',
|
||
}
|
||
|
||
@classmethod
|
||
async def _get_app_access_token(cls) -> Optional[str]:
|
||
"""获取应用级别的 access_token"""
|
||
try:
|
||
config = await cls.get_client_config_async()
|
||
|
||
url = "https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal"
|
||
data = {
|
||
'app_id': config['client_id'],
|
||
'app_secret': config['client_secret'],
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.post(url, json=data)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0:
|
||
return result.get('app_access_token')
|
||
|
||
logger.error(f"获取飞书 app_access_token 失败: {result}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取飞书 app_access_token 异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
async def get_access_token(cls, code: str) -> Optional[str]:
|
||
"""使用授权码获取访问令牌"""
|
||
try:
|
||
app_access_token = await cls._get_app_access_token()
|
||
if not app_access_token:
|
||
return None
|
||
|
||
data = {
|
||
'grant_type': 'authorization_code',
|
||
'code': code,
|
||
}
|
||
|
||
headers = {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': f'Bearer {app_access_token}',
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.post(cls.TOKEN_URL, json=data, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0 and 'data' in result:
|
||
access_token = result['data'].get('access_token')
|
||
if access_token:
|
||
logger.info(f"飞书 access_token 获取成功")
|
||
return access_token
|
||
|
||
logger.error(f"飞书 token 响应格式错误: {result}")
|
||
return None
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求飞书 access_token 失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取飞书 access_token 异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""使用访问令牌获取飞书用户信息"""
|
||
try:
|
||
headers = {
|
||
'Authorization': f'Bearer {access_token}',
|
||
'Content-Type': 'application/json',
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0 and 'data' in result:
|
||
user_info = result['data']
|
||
if 'union_id' not in user_info:
|
||
logger.error(f"飞书用户信息格式错误: {result}")
|
||
return None
|
||
return user_info
|
||
|
||
logger.error(f"飞书用户信息响应错误: {result}")
|
||
return None
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求飞书用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取飞书用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化飞书用户信息"""
|
||
provider_id = raw_user_info.get('union_id', '')
|
||
name = raw_user_info.get('name', '')
|
||
en_name = raw_user_info.get('en_name', '')
|
||
username = en_name or name or f"feishu_{provider_id[:8]}"
|
||
username = username.replace(' ', '_')
|
||
|
||
mobile = raw_user_info.get('mobile', '')
|
||
if mobile and mobile.startswith('+86-'):
|
||
mobile = mobile[4:]
|
||
|
||
return {
|
||
'provider_id': provider_id,
|
||
'username': username,
|
||
'name': name or username,
|
||
'email': raw_user_info.get('email'),
|
||
'avatar': raw_user_info.get('avatar_url') or raw_user_info.get('avatar_big'),
|
||
'mobile': mobile,
|
||
'bio': f"飞书用户 - {name}" if name else "飞书用户",
|
||
}
|
||
|
||
@classmethod
|
||
def get_user_id_field(cls) -> str:
|
||
"""获取用户 ID 字段名"""
|
||
return 'feishu_union_id'
|
||
|
||
|
||
class WeComOAuthService(BaseOAuthService):
|
||
"""企业微信 OAuth 服务类"""
|
||
|
||
PROVIDER_NAME = 'wecom'
|
||
AUTHORIZE_URL = "https://login.work.weixin.qq.com/wwlogin/sso/login"
|
||
TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
|
||
USER_INFO_URL = "https://qyapi.weixin.qq.com/cgi-bin/auth/getuserinfo"
|
||
USER_DETAIL_URL = "https://qyapi.weixin.qq.com/cgi-bin/user/get"
|
||
|
||
@classmethod
|
||
def get_user_id_field(cls) -> str:
|
||
"""企业微信使用 wecom_userid 作为唯一标识"""
|
||
return 'wecom_userid'
|
||
|
||
@classmethod
|
||
def get_client_config(cls) -> Dict[str, str]:
|
||
"""获取企业微信客户端配置（同步回退）"""
|
||
return {
|
||
'client_id': getattr(settings, 'WECOM_CORP_ID', ''),
|
||
'client_secret': getattr(settings, 'WECOM_APP_SECRET', ''),
|
||
'redirect_uri': getattr(settings, 'WECOM_REDIRECT_URI', ''),
|
||
'agent_id': getattr(settings, 'WECOM_AGENT_ID', ''),
|
||
}
|
||
|
||
@classmethod
|
||
async def get_client_config_async(cls) -> Dict[str, str]:
|
||
"""从 config_manager 三级配置获取企业微信 OAuth 配置"""
|
||
from app.config_manager import config_manager
|
||
oauth_config = await config_manager.get_group("oauth_wecom")
|
||
return {
|
||
'client_id': oauth_config.get("corp_id") or getattr(settings, 'WECOM_CORP_ID', '') or '',
|
||
'client_secret': oauth_config.get("app_secret") or getattr(settings, 'WECOM_APP_SECRET', '') or '',
|
||
'redirect_uri': oauth_config.get("redirect_uri") or getattr(settings, 'WECOM_REDIRECT_URI', '') or '',
|
||
'agent_id': oauth_config.get("agent_id") or getattr(settings, 'WECOM_AGENT_ID', '') or '',
|
||
}
|
||
|
||
@classmethod
|
||
async def get_authorize_url(cls, state: str = None, redirect_uri: str = None) -> str:
|
||
"""获取企业微信授权 URL（企业微信使用特殊的 SSO 登录页面）"""
|
||
config = await cls.get_client_config_async()
|
||
|
||
params = {
|
||
'login_type': 'CorpApp',
|
||
'appid': config['client_id'],
|
||
'agentid': config['agent_id'],
|
||
'redirect_uri': redirect_uri or config['redirect_uri'],
|
||
}
|
||
|
||
if state:
|
||
params['state'] = state
|
||
|
||
query_string = '&'.join([f"{k}={v}" for k, v in params.items()])
|
||
return f"{cls.AUTHORIZE_URL}?{query_string}"
|
||
|
||
@classmethod
|
||
async def _get_corp_access_token(cls) -> Optional[str]:
|
||
"""获取企业微信的企业级 access_token"""
|
||
try:
|
||
config = await cls.get_client_config_async()
|
||
params = {
|
||
'corpid': config['client_id'],
|
||
'corpsecret': config['client_secret'],
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.TOKEN_URL, params=params)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
|
||
if result.get('errcode') == 0 and 'access_token' in result:
|
||
logger.info("企业微信 corp access_token 获取成功")
|
||
return result['access_token']
|
||
|
||
logger.error(f"获取企业微信 corp access_token 失败: {result}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取企业微信 corp access_token 异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
async def get_access_token(cls, code: str) -> Optional[str]:
|
||
"""
|
||
企业微信不使用标准的 code 换 token 流程。
|
||
这里获取 corp access_token 并返回，code 在 get_user_info 中使用。
|
||
返回格式: "corp_access_token|code" 以便后续使用。
|
||
"""
|
||
corp_token = await cls._get_corp_access_token()
|
||
if not corp_token:
|
||
return None
|
||
# 将 corp_token 和 code 拼接传递，在 get_user_info 中拆分使用
|
||
return f"{corp_token}|{code}"
|
||
|
||
@classmethod
|
||
async def get_user_info(cls, access_token: str) -> Optional[Dict]:
|
||
"""
|
||
使用 corp access_token + code 获取企业微信用户信息。
|
||
access_token 格式: "corp_access_token|code"
|
||
"""
|
||
try:
|
||
parts = access_token.split('|', 1)
|
||
if len(parts) != 2:
|
||
logger.error("企业微信 access_token 格式错误")
|
||
return None
|
||
|
||
corp_token, code = parts
|
||
|
||
# 第一步：用 code 获取 userid
|
||
params = {
|
||
'access_token': corp_token,
|
||
'code': code,
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(cls.USER_INFO_URL, params=params)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
|
||
if result.get('errcode') != 0:
|
||
logger.error(f"企业微信获取用户身份失败: {result}")
|
||
return None
|
||
|
||
userid = result.get('userid') or result.get('UserId')
|
||
if not userid:
|
||
logger.error(f"企业微信用户身份响应中无 userid: {result}")
|
||
return None
|
||
|
||
# 第二步：用 userid 获取用户详情
|
||
detail_params = {
|
||
'access_token': corp_token,
|
||
'userid': userid,
|
||
}
|
||
|
||
detail_response = await client.get(cls.USER_DETAIL_URL, params=detail_params)
|
||
detail_response.raise_for_status()
|
||
|
||
user_detail = detail_response.json()
|
||
|
||
if user_detail.get('errcode') != 0:
|
||
logger.error(f"企业微信获取用户详情失败: {user_detail}")
|
||
# 即使获取详情失败，也返回基本信息
|
||
return {
|
||
'userid': userid,
|
||
'name': userid,
|
||
}
|
||
|
||
return user_detail
|
||
|
||
except httpx.RequestError as e:
|
||
logger.error(f"请求企业微信用户信息失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取企业微信用户信息异常: {str(e)}")
|
||
return None
|
||
|
||
@classmethod
|
||
def normalize_user_info(cls, raw_user_info: Dict) -> Dict:
|
||
"""标准化企业微信用户信息"""
|
||
userid = raw_user_info.get('userid', '')
|
||
name = raw_user_info.get('name', '')
|
||
username = userid if userid else f"wecom_{name}"
|
||
|
||
email = raw_user_info.get('biz_mail') or raw_user_info.get('email')
|
||
avatar = raw_user_info.get('thumb_avatar') or raw_user_info.get('avatar')
|
||
mobile = raw_user_info.get('mobile', '')
|
||
position = raw_user_info.get('position', '')
|
||
|
||
return {
|
||
'provider_id': userid,
|
||
'username': username,
|
||
'name': name or username,
|
||
'email': email,
|
||
'avatar': avatar,
|
||
'bio': f"企业微信用户 - {position}" if position else "企业微信用户",
|
||
}
|
||
|
||
|
||
# OAuth 提供商映射
|
||
OAUTH_PROVIDERS = {
|
||
'gitee': GiteeOAuthService,
|
||
'github': GitHubOAuthService,
|
||
'qq': QQOAuthService,
|
||
'google': GoogleOAuthService,
|
||
'wechat': WeChatOAuthService,
|
||
'microsoft': MicrosoftOAuthService,
|
||
'dingtalk': DingTalkOAuthService,
|
||
'feishu': FeishuOAuthService,
|
||
'wecom': WeComOAuthService,
|
||
}
|