凭证检验并行化优化，添加实时进度推送

- 新增 verify_credentials_batch() 批量并行检验方法
- 立即检验和后台检验改为并行执行，性能提升约 16 倍
- 添加 SSE 进度推送，前端显示检验进度条
- 使用 Semaphore(20) 控制并发数

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: sortbyiky

main.py

@@ -116,6 +116,22 @@ async def broadcast_stats(stats_data: dict):
             pass
 
 
+async def broadcast_verify_progress(completed: int, total: int, filename: str, success: bool):
+    """Broadcast verify progress to all SSE clients"""
+    progress_data = {
+        "completed": completed,
+        "total": total,
+        "percent": int(completed / total * 100) if total > 0 else 0,
+        "filename": filename,
+        "success": success,
+    }
+    for queue in sse_clients:
+        try:
+            await queue.put({"type": "verify_progress", "data": progress_data})
+        except Exception:
+            pass
+
+
 async def quota_refresh_loop():
     """Background task to refresh quota and stats data every minute and push to clients"""
     while True:
@@ -143,6 +159,7 @@ async def quota_refresh_loop():
 
 # Set SSE callback for auto_verify_service and log_forwarder
 auto_verify_service.set_log_callback(broadcast_log)
+auto_verify_service.set_progress_callback(broadcast_verify_progress)
 log_forwarder.set_log_callback(broadcast_log)
 
 

services/api_client.py

@@ -1,6 +1,6 @@
 import asyncio
 import httpx
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 import logging
 
 logger = logging.getLogger(__name__)
@@ -72,6 +72,59 @@ async def verify_credential(self, filename: str, mode: str = "antigravity") -> D
         resp.raise_for_status()
         return resp.json()
 
+    async def verify_credentials_batch(
+        self,
+        credentials: List[Dict[str, Any]],
+        max_concurrent: int = DEFAULT_MAX_CONCURRENT,
+        progress_callback: Optional[Callable] = None,
+        mode: str = "antigravity",
+    ) -> List[Dict[str, Any]]:
+        """Verify multiple credentials in parallel with progress callback"""
+        if not credentials:
+            return []
+
+        semaphore = asyncio.Semaphore(max_concurrent)
+        completed = 0
+        total = len(credentials)
+
+        async def verify_one(cred: Dict[str, Any]) -> Dict[str, Any]:
+            nonlocal completed
+            filename = cred.get("filename", "")
+            async with semaphore:
+                try:
+                    result = await self.verify_credential(filename, mode)
+                    success = True
+                except Exception as e:
+                    logger.warning(f"Failed to verify {filename}: {e}")
+                    result = {"success": False, "error": str(e)}
+                    success = False
+
+                completed += 1
+                if progress_callback:
+                    try:
+                        await progress_callback(completed, total, filename, success)
+                    except Exception as cb_err:
+                        logger.warning(f"Progress callback error: {cb_err}")
+
+                return {
+                    "filename": filename,
+                    "result": result,
+                    "success": success,
+                }
+
+        tasks = [verify_one(c) for c in credentials]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        # Filter out exceptions
+        valid_results = []
+        for r in results:
+            if isinstance(r, Exception):
+                logger.warning(f"Task exception: {r}")
+            else:
+                valid_results.append(r)
+
+        return valid_results
+
     async def get_credential_quota(self, filename: str) -> Dict[str, Any]:
         """Get quota for a credential (antigravity mode only)"""
         url = f"{self.base_url}/creds/quota/{filename}"

services/auto_verify.py

@@ -1,7 +1,7 @@
 import asyncio
 import logging
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from .api_client import GcliApiClient
 
@@ -16,11 +16,16 @@ def __init__(self):
         self._history: List[Dict[str, Any]] = []
         self._max_history = 100
         self._on_new_log = None  # SSE callback
+        self._on_progress = None  # Progress callback for SSE
 
     def set_log_callback(self, callback):
         """Set callback for new log entries (for SSE)"""
         self._on_new_log = callback
 
+    def set_progress_callback(self, callback: Optional[Callable]):
+        """Set callback for verify progress (for SSE)"""
+        self._on_progress = callback
+
     @property
     def is_running(self) -> bool:
         return self._running
@@ -93,56 +98,45 @@ async def _check_and_verify(self, error_codes: List[int]):
         logger.info(f"Found {len(to_verify)} credentials to verify")
         await self._add_history({
             "type": "info",
-            "message": f"发现 {len(to_verify)} 个需要恢复的凭证"
+            "message": f"发现 {len(to_verify)} 个需要恢复的凭证，开始并行检验..."
         })
 
-        # Verify each credential
+        # Use parallel verification with progress callback
+        async def progress_callback(completed: int, total: int, filename: str, success: bool):
+            if self._on_progress:
+                await self._on_progress(completed, total, filename, success)
+
+        results = await self._client.verify_credentials_batch(
+            to_verify,
+            progress_callback=progress_callback
+        )
+
+        # Process results and add to history
         success_count = 0
         fail_count = 0
-        for cred in to_verify:
-            filename = cred.get("filename")
-            if not filename:
-                continue
-            old_errors = cred.get("error_codes", [])
-            user_email = cred.get("user_email", "")
-
-            # Log before restore
-            await self._add_history({
-                "type": "info",
-                "message": f"正在恢复 {filename}，错误码: {old_errors}"
-            })
-
-            try:
-                result = await self._client.verify_credential(filename)
-                success = result.get("success", False)
-                message = result.get("message", "")
-
-                if success:
-                    success_count += 1
-                    await self._add_history({
-                        "type": "verify",
-                        "filename": filename,
-                        "success": True,
-                        "message": f"恢复成功 - {message}" if message else "恢复成功，凭证已启用",
-                    })
-                else:
-                    fail_count += 1
-                    await self._add_history({
-                        "type": "verify",
-                        "filename": filename,
-                        "success": False,
-                        "message": f"恢复失败 - {message}" if message else "恢复失败",
-                    })
-                logger.info(f"Verified {filename}: success={success}")
-            except Exception as e:
+        for item in results:
+            filename = item.get("filename", "")
+            success = item.get("success", False)
+            result = item.get("result", {})
+            message = result.get("message", "")
+
+            if success:
+                success_count += 1
+                await self._add_history({
+                    "type": "verify",
+                    "filename": filename,
+                    "success": True,
+                    "message": f"恢复成功 - {message}" if message else "恢复成功，凭证已启用",
+                })
+            else:
                 fail_count += 1
+                error_msg = result.get("error", message)
                 await self._add_history({
                     "type": "verify",
                     "filename": filename,
                     "success": False,
-                    "message": f"恢复异常: {str(e)}",
+                    "message": f"恢复失败 - {error_msg}" if error_msg else "恢复失败",
                 })
-                logger.warning(f"Failed to verify {filename}: {e}")
 
         # Summary log
         await self._add_history({
@@ -151,37 +145,62 @@ async def _check_and_verify(self, error_codes: List[int]):
         })
 
     async def trigger_now(self, error_codes: List[int] = None) -> Dict[str, Any]:
-        """Manually trigger verification for all credentials"""
+        """Manually trigger verification for all credentials (parallel execution)"""
         if not self._client:
             return {"success": False, "message": "Not connected"}
 
-        results = []
         # Get all credentials (not just disabled)
         all_creds = await self._client.get_all_credentials()
+        if not all_creds:
+            return {"success": True, "total": 0, "verified": 0, "results": []}
 
-        for cred in all_creds:
-            filename = cred.get("filename")
-            if not filename:
-                continue
-            try:
-                result = await self._client.verify_credential(filename)
-                results.append({
-                    "filename": filename,
-                    "success": result.get("success", False),
-                    "message": result.get("message", ""),
-                })
-                await self._add_history({
-                    "type": "verify",
-                    "filename": filename,
-                    "success": result.get("success", False),
-                    "message": result.get("message", ""),
-                })
-            except Exception as e:
-                results.append({
-                    "filename": filename,
-                    "success": False,
-                    "message": str(e),
-                })
+        await self._add_history({
+            "type": "info",
+            "message": f"开始立即检验，共 {len(all_creds)} 个凭证，并行执行中..."
+        })
+
+        # Use parallel verification with progress callback
+        async def progress_callback(completed: int, total: int, filename: str, success: bool):
+            if self._on_progress:
+                await self._on_progress(completed, total, filename, success)
+
+        batch_results = await self._client.verify_credentials_batch(
+            all_creds,
+            progress_callback=progress_callback
+        )
+
+        # Process results and add to history
+        results = []
+        success_count = 0
+        fail_count = 0
+        for item in batch_results:
+            filename = item.get("filename", "")
+            success = item.get("success", False)
+            result = item.get("result", {})
+            message = result.get("message", "")
+
+            results.append({
+                "filename": filename,
+                "success": success,
+                "message": message,
+            })
+
+            if success:
+                success_count += 1
+            else:
+                fail_count += 1
+
+            await self._add_history({
+                "type": "verify",
+                "filename": filename,
+                "success": success,
+                "message": message if message else ("检验成功" if success else "检验失败"),
+            })
+
+        await self._add_history({
+            "type": "info",
+            "message": f"立即检验完成: 成功 {success_count} 个, 失败 {fail_count} 个"
+        })
 
         return {
             "success": True,