获取有效的 token（带缓存，所有实例共享） Args: force_refresh: 是否强制刷新 Returns: token 字符串
(self, force_refresh: bool = False)
| 130 | raise EmailServiceError(f"生成 token 失败: {e}") |
| 131 | |
| 132 | def _get_token(self, force_refresh: bool = False) -> str: |
| 133 | """ |
| 134 | 获取有效的 token（带缓存，所有实例共享） |
| 135 | |
| 136 | Args: |
| 137 | force_refresh: 是否强制刷新 |
| 138 | |
| 139 | Returns: |
| 140 | token 字符串 |
| 141 | """ |
| 142 | base_url = self.config["base_url"] |
| 143 | |
| 144 | with CloudMailService._token_lock: |
| 145 | # 检查共享缓存（token 有效期设为 1 小时） |
| 146 | if not force_refresh and base_url in CloudMailService._shared_tokens: |
| 147 | token, expires_at = CloudMailService._shared_tokens[base_url] |
| 148 | if time.time() < expires_at: |
| 149 | return token |
| 150 | |
| 151 | # 生成新 token |
| 152 | token = self._generate_token() |
| 153 | expires_at = time.time() + 3600 # 1 小时后过期 |
| 154 | CloudMailService._shared_tokens[base_url] = (token, expires_at) |
| 155 | return token |
| 156 | |
| 157 | def _get_headers(self, token: Optional[str] = None) -> Dict[str, str]: |
| 158 | """构造请求头""" |
no test coverage detected