Refactor and enhance OAuth2.0 implementation; update models and routes

- Refactored AdminSummaryData and AdminSummaryResponse classes for better clarity.
- Added OAUTH type to SettingsType enum.
- Cleaned up imports in webdav.py.
- Updated admin router to improve summary data retrieval and response handling.
- Enhanced file management routes with better condition handling and user storage updates.
- Improved group management routes by optimizing data retrieval.
- Refined task management routes for better condition handling.
- Updated user management routes to streamline access token retrieval.
- Implemented a new captcha verification structure with abstract base class.
- Removed deprecated env.md file and replaced with a new structured version.
- Introduced a unified OAuth2.0 client base class for GitHub and QQ integrations.
- Enhanced password management with improved hashing strategies.
- Added detailed comments and documentation throughout the codebase for clarity.
This commit is contained in:
2026-01-12 18:07:44 +08:00
parent 61ddc96f17
commit d2c914cff8
29 changed files with 814 additions and 4609 deletions

View File

@@ -1,7 +1,158 @@
from pydantic import BaseModel
"""QQ OAuth2.0 认证实现"""
import aiohttp
async def get_access_token(
from pydantic import BaseModel
from . import AccessTokenBase, OAuthBase
class QQAccessToken(AccessTokenBase):
"""QQ 访问令牌响应"""
expires_in: int
"""access token 的有效期,单位为秒"""
refresh_token: str
"""用于刷新 access token 的令牌"""
class QQOpenIDResponse(BaseModel):
"""QQ OpenID 响应"""
client_id: str
"""应用的 appid"""
openid: str
"""用户的唯一标识"""
class QQUserData(BaseModel):
"""QQ 用户数据"""
ret: int
"""返回码0 表示成功"""
msg: str
"""返回信息"""
nickname: str | None
"""用户昵称"""
gender: str | None
"""性别"""
figureurl: str | None
"""头像 URL"""
figureurl_1: str | None
"""头像 URL大图"""
figureurl_2: str | None
"""头像 URL更大图"""
figureurl_qq_1: str | None
"""QQ 头像 URL大图"""
figureurl_qq_2: str | None
"""QQ 头像 URL更大图"""
is_yellow_vip: str | None
"""是否黄钻用户"""
vip: str | None
"""是否 VIP 用户"""
yellow_vip_level: str | None
"""黄钻等级"""
level: str | None
"""等级"""
is_yellow_year_vip: str | None
"""是否年费黄钻"""
class QQUserInfoResponse(BaseModel):
"""QQ 用户信息响应"""
code: str
):
...
"""状态码"""
openid: str
"""用户 OpenID"""
user_data: QQUserData
"""用户数据"""
class QQOAuth(OAuthBase):
"""QQ OAuth2.0 客户端"""
access_token_url = "https://graph.qq.com/oauth2.0/token"
"""获取 Access Token 的 API 地址"""
user_info_url = "https://graph.qq.com/user/get_user_info"
"""获取用户信息的 API 地址"""
openid_url = "https://graph.qq.com/oauth2.0/me"
"""获取 OpenID 的 API 地址"""
http_method = "GET"
"""获取 token 的 HTTP 方法"""
async def get_access_token(self, code: str, redirect_uri: str) -> QQAccessToken:
"""
通过 Authorization Code 获取 Access Token
Args:
code: 授权码
redirect_uri: 与授权时传入的 redirect_uri 保持一致,需要 URLEncode
Returns:
QQAccessToken: 访问令牌
文档:
https://wiki.connect.qq.com/%E4%BD%BF%E7%94%A8authorization_code%E8%8E%B7%E5%8F%96access_token
"""
params = {
'grant_type': 'authorization_code',
'client_id': self.client_id,
'client_secret': self.client_secret,
'code': code,
'redirect_uri': redirect_uri,
'fmt': 'json',
'need_openid': 1,
}
async with aiohttp.ClientSession() as session:
async with session.get(url=self.access_token_url, params=params) as access_resp:
access_data = await access_resp.json()
return QQAccessToken(
access_token=access_data.get('access_token'),
expires_in=access_data.get('expires_in'),
refresh_token=access_data.get('refresh_token'),
)
async def get_openid(self, access_token: str) -> QQOpenIDResponse:
"""
获取用户 OpenID
注意:如果在 get_access_token 时传入了 need_openid=1响应中已包含 openid
无需额外调用此接口。此函数用于单独获取 openid 的场景。
Args:
access_token: 访问令牌
Returns:
QQOpenIDResponse: 包含 client_id 和 openid
文档:
https://wiki.connect.qq.com/%E8%8E%B7%E5%8F%96%E7%94%A8%E6%88%B7openid%E7%9A%84oauth2.0%E6%8E%A5%E5%8F%A3
"""
async with aiohttp.ClientSession() as session:
async with session.get(
url=self.openid_url,
params={
'access_token': access_token,
'fmt': 'json',
},
) as resp:
data = await resp.json()
return QQOpenIDResponse(
client_id=data.get('client_id'),
openid=data.get('openid'),
)
def _build_user_info_params(self, access_token: str, **kwargs) -> dict:
"""构建 QQ 用户信息请求参数"""
return {
'access_token': access_token,
'oauth_consumer_key': kwargs.get('app_id', self.client_id),
'openid': kwargs.get('openid', ''),
}
def _parse_user_response(self, data: dict) -> QQUserInfoResponse:
"""解析 QQ 用户信息响应"""
return QQUserInfoResponse(
code='0' if data.get('ret') == 0 else str(data.get('ret')),
openid=data.get('openid', ''),
user_data=QQUserData(**data),
)