用授权码换取token
(self, code: str)
| 204 | return f"{self.auth_endpoint}?{urlencode(params)}" |
| 205 | |
| 206 | async def exchange_code(self, code: str) -> Credentials: |
| 207 | """用授权码换取token""" |
| 208 | data = { |
| 209 | "client_id": self.client_id, |
| 210 | "client_secret": self.client_secret, |
| 211 | "redirect_uri": self.redirect_uri, |
| 212 | "code": code, |
| 213 | "grant_type": "authorization_code", |
| 214 | } |
| 215 | |
| 216 | try: |
| 217 | oauth_base_url = await get_oauth_proxy_url() |
| 218 | token_url = f"{oauth_base_url.rstrip('/')}/token" |
| 219 | response = await post_async( |
| 220 | token_url, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"} |
| 221 | ) |
| 222 | response.raise_for_status() |
| 223 | |
| 224 | token_data = response.json() |
| 225 | |
| 226 | # 计算过期时间 |
| 227 | expires_at = None |
| 228 | if "expires_in" in token_data: |
| 229 | expires_in = int(token_data["expires_in"]) |
| 230 | expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in) |
| 231 | |
| 232 | # 创建凭证对象 |
| 233 | self.credentials = Credentials( |
| 234 | access_token=token_data["access_token"], |
| 235 | refresh_token=token_data.get("refresh_token"), |
| 236 | client_id=self.client_id, |
| 237 | client_secret=self.client_secret, |
| 238 | expires_at=expires_at, |
| 239 | ) |
| 240 | |
| 241 | return self.credentials |
| 242 | |
| 243 | except Exception as e: |
| 244 | error_msg = f"获取token失败: {str(e)}" |
| 245 | log.error(error_msg) |
| 246 | raise TokenError(error_msg) |
| 247 | |
| 248 | |
| 249 | class ServiceAccount: |
no test coverage detected