刷新访问令牌
(self)
| 71 | return True |
| 72 | |
| 73 | async def refresh(self): |
| 74 | """刷新访问令牌""" |
| 75 | if not self.refresh_token: |
| 76 | raise TokenError("无刷新令牌") |
| 77 | |
| 78 | data = { |
| 79 | "client_id": self.client_id, |
| 80 | "client_secret": self.client_secret, |
| 81 | "refresh_token": self.refresh_token, |
| 82 | "grant_type": "refresh_token", |
| 83 | } |
| 84 | |
| 85 | try: |
| 86 | oauth_base_url = await get_oauth_proxy_url() |
| 87 | token_url = f"{oauth_base_url.rstrip('/')}/token" |
| 88 | response = await post_async( |
| 89 | token_url, |
| 90 | data=data, |
| 91 | headers={"Content-Type": "application/x-www-form-urlencoded"}, |
| 92 | ) |
| 93 | response.raise_for_status() |
| 94 | |
| 95 | token_data = response.json() |
| 96 | self.access_token = token_data["access_token"] |
| 97 | |
| 98 | if "expires_in" in token_data: |
| 99 | expires_in = int(token_data["expires_in"]) |
| 100 | current_utc = datetime.now(timezone.utc) |
| 101 | self.expires_at = current_utc + timedelta(seconds=expires_in) |
| 102 | log.debug( |
| 103 | f"Token刷新: 当前UTC时间={current_utc.isoformat()}, " |
| 104 | f"有效期={expires_in}秒, " |
| 105 | f"过期时间={self.expires_at.isoformat()}" |
| 106 | ) |
| 107 | |
| 108 | if "refresh_token" in token_data: |
| 109 | self.refresh_token = token_data["refresh_token"] |
| 110 | |
| 111 | log.debug(f"Token刷新成功，过期时间: {self.expires_at}") |
| 112 | |
| 113 | except Exception as e: |
| 114 | error_msg = str(e) |
| 115 | status_code = None |
| 116 | if hasattr(e, 'response') and hasattr(e.response, 'status_code'): |
| 117 | status_code = e.response.status_code |
| 118 | error_msg = f"Token刷新失败 (HTTP {status_code}): {error_msg}" |
| 119 | else: |
| 120 | error_msg = f"Token刷新失败: {error_msg}" |
| 121 | |
| 122 | log.error(error_msg) |
| 123 | token_error = TokenError(error_msg) |
| 124 | token_error.status_code = status_code |
| 125 | raise token_error |
| 126 | |
| 127 | @classmethod |
| 128 | def from_dict(cls, data: Dict[str, Any]) -> "Credentials": |
no test coverage detected