(method: str)
| 168 | |
| 169 | |
| 170 | async def _prepare_captcha_runtime(method: str): |
| 171 | runtime_method = _normalize_runtime_method(method) |
| 172 | try: |
| 173 | if runtime_method == "browser": |
| 174 | start_runtime_prepare( |
| 175 | runtime_method, |
| 176 | "已开始准备有头浏览器打码运行环境,安装进度将自动显示。", |
| 177 | ) |
| 178 | else: |
| 179 | start_runtime_prepare( |
| 180 | runtime_method, |
| 181 | "已开始准备内置浏览器打码运行环境,安装进度将自动显示。", |
| 182 | ) |
| 183 | |
| 184 | if runtime_method == "browser": |
| 185 | module = await asyncio.to_thread(importlib.import_module, "src.services.browser_captcha") |
| 186 | service_cls = getattr(module, "BrowserCaptchaService") |
| 187 | service = await service_cls.get_instance(db) |
| 188 | if hasattr(service, "reload_browser_count"): |
| 189 | await service.reload_browser_count() |
| 190 | if hasattr(service, "warmup_browser_slots"): |
| 191 | await service.warmup_browser_slots() |
| 192 | finish_runtime_prepare(runtime_method, "Chromium 浏览器环境已就绪,可以开始使用有头浏览器打码。") |
| 193 | return |
| 194 | |
| 195 | module = await asyncio.to_thread(importlib.import_module, "src.services.browser_captcha_personal") |
| 196 | service_cls = getattr(module, "BrowserCaptchaService") |
| 197 | service = await service_cls.get_instance(db) |
| 198 | await service.reload_config() |
| 199 | finish_runtime_prepare(runtime_method, "内置浏览器环境已就绪,可以开始使用 personal 打码。") |
| 200 | except HTTPException: |
| 201 | raise |
| 202 | except Exception as e: |
| 203 | fail_runtime_prepare(runtime_method, f"浏览器环境准备失败: {type(e).__name__}: {e}") |
| 204 | finally: |
| 205 | captcha_runtime_prepare_tasks.pop(runtime_method, None) |
| 206 | |
| 207 | |
| 208 | def _schedule_captcha_runtime_prepare(method: str) -> bool: |
no test coverage detected