确保 token 的 AT 可用,并在必要时返回刷新后的最新对象。
(self, token: Optional[Token])
| 520 | return self._should_refresh_at(token) |
| 521 | |
| 522 | async def ensure_valid_token(self, token: Optional[Token]) -> Optional[Token]: |
| 523 | """确保 token 的 AT 可用,并在必要时返回刷新后的最新对象。""" |
| 524 | if not token: |
| 525 | return None |
| 526 | |
| 527 | if not self._should_refresh_at(token): |
| 528 | if self._has_recent_at_validation(token.id): |
| 529 | return token |
| 530 | try: |
| 531 | credits_result = await self._get_credits_for_token(token, token.at) |
| 532 | await self.db.update_token( |
| 533 | token.id, |
| 534 | credits=credits_result.get("credits", 0), |
| 535 | user_paygate_tier=credits_result.get("userPaygateTier"), |
| 536 | ) |
| 537 | self._mark_at_valid(token.id) |
| 538 | return await self.db.get_token(token.id) or token |
| 539 | except Exception as e: |
| 540 | debug_logger.log_warning( |
| 541 | f"[AT_CHECK] Token {token.id}: 本地判定未过期,但上游校验失败,准备刷新 AT/ST - {e}" |
| 542 | ) |
| 543 | |
| 544 | if not await self._refresh_at(token.id): |
| 545 | return None |
| 546 | |
| 547 | return await self.db.get_token(token.id) |
| 548 | |
| 549 | async def is_at_valid(self, token_id: int, token: Optional[Token] = None) -> bool: |
| 550 | """检查AT是否有效,如果无效或即将过期则自动刷新 |
no test coverage detected