根据当前 token 快照判断是否需要刷新 AT。
(self, token: Token)
| 488 | # ========== AT自动刷新逻辑 (核心) ========== |
| 489 | |
| 490 | def _should_refresh_at(self, token: Token) -> bool: |
| 491 | """根据当前 token 快照判断是否需要刷新 AT。""" |
| 492 | if not token.at: |
| 493 | debug_logger.log_info(f"[AT_CHECK] Token {token.id}: AT不存在,需要刷新") |
| 494 | return True |
| 495 | |
| 496 | if not token.at_expires: |
| 497 | debug_logger.log_info(f"[AT_CHECK] Token {token.id}: AT过期时间未知,尝试刷新") |
| 498 | return True |
| 499 | |
| 500 | now = datetime.now(timezone.utc) |
| 501 | if token.at_expires.tzinfo is None: |
| 502 | at_expires_aware = token.at_expires.replace(tzinfo=timezone.utc) |
| 503 | else: |
| 504 | at_expires_aware = token.at_expires |
| 505 | |
| 506 | time_until_expiry = at_expires_aware - now |
| 507 | if time_until_expiry.total_seconds() < 3600: |
| 508 | debug_logger.log_info( |
| 509 | f"[AT_CHECK] Token {token.id}: AT即将过期 " |
| 510 | f"(剩余 {time_until_expiry.total_seconds():.0f} 秒),需要刷新" |
| 511 | ) |
| 512 | return True |
| 513 | |
| 514 | return False |
| 515 | |
| 516 | def needs_at_refresh(self, token: Optional[Token]) -> bool: |
| 517 | """供调度层快速判断当前 token 是否大概率会触发 AT 刷新。""" |
no test coverage detected