feat(accounts): add server-side concurrent batch operations

src/core/openai/token_refresh.py

@@ -323,10 +323,24 @@ def validate_account_token(account_id: int, proxy_url: Optional[str] = None) ->
     with get_db() as db:
         account = crud.get_account_by_id(db, account_id)
         if not account:
+            logger.warning(f"验证账号失败，账号不存在: {account_id}")
             return False, "账号不存在"
 
         if not account.access_token:
+            logger.warning(f"验证账号 {account.email} 失败: 缺少 access_token")
+            crud.update_account(db, account_id, status="failed")
             return False, "账号没有 access_token"
 
+        logger.info(f"开始验证账号 Token: {account.email}")
         manager = TokenRefreshManager(proxy_url=proxy_url)
-        return manager.validate_token(account.access_token)
+        is_valid, error = manager.validate_token(account.access_token)
+        crud.update_account(
+            db,
+            account_id,
+            status="active" if is_valid else "failed",
+        )
+        if is_valid:
+            logger.info(f"账号 Token 验证成功: {account.email}")
+        else:
+            logger.warning(f"账号 Token 验证失败: {account.email}, 原因: {error}")
+        return is_valid, error

src/web/routes/accounts.py

@@ -5,10 +5,11 @@
 import json
 import logging
 import zipfile
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from typing import List, Optional
 
-from fastapi import APIRouter, HTTPException, Query, BackgroundTasks, Body
+from fastapi import APIRouter, HTTPException, Query, Body
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
@@ -99,6 +100,22 @@ class BatchUpdateRequest(BaseModel):
 
 # ============== Helper Functions ==============
 
+def normalize_batch_concurrency(concurrency: Optional[int]) -> int:
+    """限制批处理并发数在 2-10 之间。"""
+    if concurrency is None:
+        return 4
+    return max(2, min(10, concurrency))
+
+
+def run_batch_concurrently(items: List[int], concurrency: Optional[int], worker):
+    """使用本地线程池并发执行批处理任务。"""
+    if not items:
+        return []
+
+    max_workers = min(len(items), normalize_batch_concurrency(concurrency))
+    with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="account_batch") as executor:
+        return list(executor.map(worker, items))
+
 def resolve_account_ids(
     db,
     ids: List[int],
@@ -193,7 +210,6 @@ async def list_accounts(
             accounts=[account_to_response(acc) for acc in accounts]
         )
 
-
 @router.get("/{account_id}", response_model=AccountResponse)
 async def get_account(account_id: int):
     """获取单个账号详情"""
@@ -576,6 +592,7 @@ class BatchRefreshRequest(BaseModel):
     """批量刷新请求"""
     ids: List[int] = []
     proxy: Optional[str] = None
+    concurrency: Optional[int] = 4
     select_all: bool = False
     status_filter: Optional[str] = None
     email_service_filter: Optional[str] = None
@@ -591,14 +608,15 @@ class BatchValidateRequest(BaseModel):
     """批量验证请求"""
     ids: List[int] = []
     proxy: Optional[str] = None
+    concurrency: Optional[int] = 4
     select_all: bool = False
     status_filter: Optional[str] = None
     email_service_filter: Optional[str] = None
     search_filter: Optional[str] = None
 
 
 @router.post("/batch-refresh")
-async def batch_refresh_tokens(request: BatchRefreshRequest, background_tasks: BackgroundTasks):
+def batch_refresh_tokens(request: BatchRefreshRequest):
     """批量刷新账号 Token"""
     proxy = _get_proxy(request.proxy)
 
@@ -614,23 +632,30 @@ async def batch_refresh_tokens(request: BatchRefreshRequest, background_tasks: B
             request.status_filter, request.email_service_filter, request.search_filter
         )
 
-    for account_id in ids:
+    def worker(account_id: int):
         try:
-            result = do_refresh(account_id, proxy)
-            if result.success:
-                results["success_count"] += 1
-            else:
-                results["failed_count"] += 1
-                results["errors"].append({"id": account_id, "error": result.error_message})
+            return {"id": account_id, "result": do_refresh(account_id, proxy)}
         except Exception as e:
+            return {"id": account_id, "error": str(e)}
+
+    for item in run_batch_concurrently(ids, request.concurrency, worker):
+        if item.get("error") is not None:
             results["failed_count"] += 1
-            results["errors"].append({"id": account_id, "error": str(e)})
+            results["errors"].append({"id": item["id"], "error": item["error"]})
+            continue
+
+        result = item["result"]
+        if result.success:
+            results["success_count"] += 1
+        else:
+            results["failed_count"] += 1
+            results["errors"].append({"id": item["id"], "error": result.error_message})
 
     return results
 
 
 @router.post("/{account_id}/refresh")
-async def refresh_account_token(account_id: int, request: Optional[TokenRefreshRequest] = Body(default=None)):
+def refresh_account_token(account_id: int, request: Optional[TokenRefreshRequest] = Body(default=None)):
     """刷新单个账号的 Token"""
     proxy = _get_proxy(request.proxy if request else None)
     result = do_refresh(account_id, proxy)
@@ -649,7 +674,7 @@ async def refresh_account_token(account_id: int, request: Optional[TokenRefreshR
 
 
 @router.post("/batch-validate")
-async def batch_validate_tokens(request: BatchValidateRequest):
+def batch_validate_tokens(request: BatchValidateRequest):
     """批量验证账号 Token 有效性"""
     proxy = _get_proxy(request.proxy)
 
@@ -665,31 +690,37 @@ async def batch_validate_tokens(request: BatchValidateRequest):
             request.status_filter, request.email_service_filter, request.search_filter
         )
 
-    for account_id in ids:
+    def worker(account_id: int):
         try:
             is_valid, error = do_validate(account_id, proxy)
-            results["details"].append({
+            return {
                 "id": account_id,
                 "valid": is_valid,
-                "error": error
-            })
-            if is_valid:
-                results["valid_count"] += 1
-            else:
-                results["invalid_count"] += 1
+                "error": error,
+            }
         except Exception as e:
-            results["invalid_count"] += 1
-            results["details"].append({
+            return {
                 "id": account_id,
                 "valid": False,
-                "error": str(e)
-            })
+                "error": str(e),
+            }
+
+    for item in run_batch_concurrently(ids, request.concurrency, worker):
+        results["details"].append({
+            "id": item["id"],
+            "valid": item["valid"],
+            "error": item["error"]
+        })
+        if item["valid"]:
+            results["valid_count"] += 1
+        else:
+            results["invalid_count"] += 1
 
     return results
 
 
 @router.post("/{account_id}/validate")
-async def validate_account_token(account_id: int, request: Optional[TokenValidateRequest] = Body(default=None)):
+def validate_account_token(account_id: int, request: Optional[TokenValidateRequest] = Body(default=None)):
     """验证单个账号的 Token 有效性"""
     proxy = _get_proxy(request.proxy if request else None)
     is_valid, error = do_validate(account_id, proxy)

src/web/routes/payment.py

@@ -13,7 +13,7 @@
 from ...database.models import Account
 from ...database import crud
 from ...config.settings import get_settings
-from .accounts import resolve_account_ids
+from .accounts import resolve_account_ids, run_batch_concurrently
 from ...core.openai.payment import (
     generate_plus_link,
     generate_team_link,
@@ -50,12 +50,12 @@ class MarkSubscriptionRequest(BaseModel):
 class BatchCheckSubscriptionRequest(BaseModel):
     ids: List[int] = []
     proxy: Optional[str] = None
+    concurrency: Optional[int] = 4
     select_all: bool = False
     status_filter: Optional[str] = None
     email_service_filter: Optional[str] = None
     search_filter: Optional[str] = None
 
-
 # ============== 支付链接生成 ==============
 
 @router.post("/generate-link")
@@ -122,6 +122,40 @@ def open_browser_incognito(request: OpenIncognitoRequest):
 
 # ============== 订阅状态 ==============
 
+def _check_subscription_for_account(account: Account, proxy: Optional[str], db) -> dict:
+    logger.info(f"开始检测账号订阅状态: {account.email}")
+    try:
+        status = check_subscription_status(account, proxy)
+        account.subscription_type = None if status == "free" else status
+        account.subscription_at = datetime.utcnow() if status != "free" else account.subscription_at
+        db.commit()
+        logger.info(f"账号订阅检测成功: {account.email}, 结果: {status}")
+        return {
+            "id": account.id,
+            "email": account.email,
+            "success": True,
+            "subscription_type": status,
+        }
+    except Exception as e:
+        db.rollback()
+        logger.warning(f"账号订阅检测失败: {account.email}, 原因: {e}")
+        return {
+            "id": account.id,
+            "email": account.email,
+            "success": False,
+            "error": str(e),
+        }
+
+
+def _check_subscription_for_account_id(account_id: int, proxy: Optional[str]) -> dict:
+    with get_db() as db:
+        account = db.query(Account).filter(Account.id == account_id).first()
+        if not account:
+            return {"id": account_id, "email": None, "success": False, "error": "账号不存在"}
+
+        return _check_subscription_for_account(account, proxy, db)
+
+
 @router.post("/accounts/batch-check-subscription")
 def batch_check_subscription(request: BatchCheckSubscriptionRequest):
     """批量检测账号订阅状态"""
@@ -134,29 +168,16 @@ def batch_check_subscription(request: BatchCheckSubscriptionRequest):
             db, request.ids, request.select_all,
             request.status_filter, request.email_service_filter, request.search_filter
         )
-        for account_id in ids:
-            account = db.query(Account).filter(Account.id == account_id).first()
-            if not account:
-                results["failed_count"] += 1
-                results["details"].append(
-                    {"id": account_id, "email": None, "success": False, "error": "账号不存在"}
-                )
-                continue
-
-            try:
-                status = check_subscription_status(account, proxy)
-                account.subscription_type = None if status == "free" else status
-                account.subscription_at = datetime.utcnow() if status != "free" else account.subscription_at
-                db.commit()
-                results["success_count"] += 1
-                results["details"].append(
-                    {"id": account_id, "email": account.email, "success": True, "subscription_type": status}
-                )
-            except Exception as e:
-                results["failed_count"] += 1
-                results["details"].append(
-                    {"id": account_id, "email": account.email, "success": False, "error": str(e)}
-                )
+
+    def worker(account_id: int):
+        return _check_subscription_for_account_id(account_id, proxy)
+
+    for detail in run_batch_concurrently(ids, request.concurrency, worker):
+        results["details"].append(detail)
+        if detail["success"]:
+            results["success_count"] += 1
+        else:
+            results["failed_count"] += 1
 
     return results
 
@@ -178,5 +199,3 @@ def mark_subscription(account_id: int, request: MarkSubscriptionRequest):
         db.commit()
 
     return {"success": True, "subscription_type": request.subscription_type}
-
-

static/css/style.css

@@ -776,6 +776,18 @@ body {
     box-shadow: 0 0 0 3px var(--primary-light);
 }
 
+.batch-concurrency-control {
+    display: inline-flex;
+    align-items: center;
+    gap: 8px;
+    color: var(--text-secondary);
+    font-size: 0.875rem;
+}
+
+.batch-concurrency-control .form-select {
+    min-width: 84px;
+}
+
 /* ============================================
    分页
    ============================================ */
@@ -1065,6 +1077,40 @@ body {
     to { transform: translateX(100%); }
 }
 
+.batch-progress-panel {
+    flex: 1 1 100%;
+    background: var(--surface-hover);
+    border: 1px solid var(--border);
+    border-radius: var(--radius-lg);
+    padding: var(--spacing-md);
+}
+
+.batch-progress-header {
+    display: flex;
+    justify-content: space-between;
+    align-items: flex-start;
+    gap: var(--spacing-md);
+    margin-bottom: var(--spacing-sm);
+}
+
+.batch-progress-title {
+    font-size: 0.95rem;
+    font-weight: 600;
+    color: var(--text-primary);
+}
+
+.batch-progress-meta,
+.batch-progress-stats {
+    font-size: 0.875rem;
+    color: var(--text-secondary);
+}
+
+.batch-progress-percent {
+    font-size: 1rem;
+    font-weight: 700;
+    color: var(--primary-color);
+}
+
 /* ============================================
    空状态
    ============================================ */
@@ -1260,6 +1306,15 @@ body {
         align-items: stretch;
     }
 
+    .batch-concurrency-control {
+        justify-content: space-between;
+    }
+
+    .batch-progress-header {
+        align-items: stretch;
+        flex-direction: column;
+    }
+
     .form-row {
         grid-template-columns: 1fr;
     }