Delete token
(self, token_id: int)
| 219 | return await self.db.get_token(token_id) |
| 220 | |
| 221 | async def delete_token(self, token_id: int): |
| 222 | """Delete token""" |
| 223 | token = await self.db.get_token(token_id) |
| 224 | project_ids: List[str] = [] |
| 225 | if token: |
| 226 | current_project_id = str(token.current_project_id or "").strip() |
| 227 | if current_project_id: |
| 228 | project_ids.append(current_project_id) |
| 229 | |
| 230 | for project in await self.db.get_projects_by_token(token_id): |
| 231 | project_id = str(project.project_id or "").strip() |
| 232 | if project_id and project_id not in project_ids: |
| 233 | project_ids.append(project_id) |
| 234 | |
| 235 | await self.db.delete_token(token_id) |
| 236 | |
| 237 | refresh_task = self._refresh_futures.pop(token_id, None) |
| 238 | if refresh_task and not refresh_task.done(): |
| 239 | refresh_task.cancel() |
| 240 | try: |
| 241 | await refresh_task |
| 242 | except asyncio.CancelledError: |
| 243 | pass |
| 244 | except Exception: |
| 245 | pass |
| 246 | self._refresh_locks.pop(token_id, None) |
| 247 | self._project_locks.pop(token_id, None) |
| 248 | |
| 249 | if config.captcha_method == "personal" and project_ids: |
| 250 | try: |
| 251 | from .browser_captcha_personal import BrowserCaptchaService |
| 252 | service = await BrowserCaptchaService.get_instance(self.db) |
| 253 | for project_id in project_ids: |
| 254 | await service.stop_resident_mode(project_id) |
| 255 | except Exception as e: |
| 256 | debug_logger.log_warning(f"[DELETE_TOKEN] 清理 personal 浏览器状态失败: {e}") |
| 257 | |
| 258 | async def enable_token(self, token_id: int): |
| 259 | """Enable a token and reset error count""" |
nothing calls this directly
no test coverage detected