refactor: remove unused files and streamline batch processing configuration in services

app/__init__.py

@@ -1 +0,0 @@
-"""App Package"""

app/api/v1/admin.py

@@ -816,15 +816,9 @@ async def refresh_tokens_api(data: dict):
         # 去重
         unique_tokens = _dedupe_tokens(tokens)
 
-        # 批量执行配置
-        max_concurrent = get_config("usage.concurrent")
-        batch_size = get_config("usage.batch_size")
-
         raw_results = await UsageService.batch(
             unique_tokens,
             mgr,
-            max_concurrent=max_concurrent,
-            batch_size=batch_size,
         )
 
         results = {}
@@ -854,9 +848,6 @@ async def refresh_tokens_api_async(data: dict):
     # 去重
     unique_tokens = _dedupe_tokens(tokens)
 
-    max_concurrent = get_config("usage.concurrent")
-    batch_size = get_config("usage.batch_size")
-
     task = create_task(len(unique_tokens))
 
     async def _run():
@@ -868,8 +859,6 @@ async def refresh_tokens_api_async(data: dict):
             raw_results = await UsageService.batch(
                 unique_tokens,
                 mgr,
-                max_concurrent=max_concurrent,
-                batch_size=batch_size,
                 on_item=_on_item,
                 should_cancel=lambda: task.cancelled,
             )
@@ -939,15 +928,9 @@ async def enable_nsfw_api(data: dict):
         # 去重
         unique_tokens = _dedupe_tokens(tokens)
 
-        # 批量执行配置
-        max_concurrent = get_config("nsfw.concurrent")
-        batch_size = get_config("nsfw.batch_size")
-
         raw_results = await NSFWService.batch(
             unique_tokens,
             mgr,
-            max_concurrent=max_concurrent,
-            batch_size=batch_size,
         )
 
         # 构造返回结果（mask token）
@@ -1005,9 +988,6 @@ async def enable_nsfw_api_async(data: dict):
     # 去重
     unique_tokens = _dedupe_tokens(tokens)
 
-    max_concurrent = get_config("nsfw.concurrent")
-    batch_size = get_config("nsfw.batch_size")
-
     task = create_task(len(unique_tokens))
 
     async def _run():
@@ -1020,8 +1000,6 @@ async def enable_nsfw_api_async(data: dict):
             raw_results = await NSFWService.batch(
                 unique_tokens,
                 mgr,
-                max_concurrent=max_concurrent,
-                batch_size=batch_size,
                 on_item=_on_item,
                 should_cancel=lambda: task.cancelled,
             )
@@ -1123,15 +1101,11 @@ async def get_cache_stats_api(request: Request):
         }
         online_details = []
         account_map = {a["token"]: a for a in accounts}
-        batch_size = max(1, int(get_config("asset.list_batch_size")))
-        max_concurrent = batch_size
         if selected_tokens:
             total = 0
             raw_results = await ListService.fetch_assets_details(
                 selected_tokens,
                 account_map,
-                max_concurrent=max_concurrent,
-                batch_size=batch_size,
             )
             for token, res in raw_results.items():
                 if res.get("ok"):
@@ -1164,8 +1138,6 @@ async def get_cache_stats_api(request: Request):
             raw_results = await ListService.fetch_assets_details(
                 tokens,
                 account_map,
-                max_concurrent=max_concurrent,
-                batch_size=batch_size,
             )
             for token, res in raw_results.items():
                 if res.get("ok"):
@@ -1197,8 +1169,6 @@ async def get_cache_stats_api(request: Request):
                 raw_results = await ListService.fetch_assets_details(
                     [token],
                     account_map,
-                    max_concurrent=1,
-                    batch_size=1,
                 )
                 res = raw_results.get(token, {})
                 data = res.get("data", {})
@@ -1289,9 +1259,6 @@ async def load_online_cache_api_async(data: dict):
     else:
         raise HTTPException(status_code=400, detail="No tokens provided")
 
-    batch_size = get_config("asset.list_batch_size")
-    max_concurrent = batch_size
-
     task = create_task(len(selected_tokens))
 
     async def _run():
@@ -1307,8 +1274,6 @@ async def load_online_cache_api_async(data: dict):
             raw_results = await ListService.fetch_assets_details(
                 selected_tokens,
                 account_map,
-                max_concurrent=max_concurrent,
-                batch_size=batch_size,
                 include_ok=True,
                 on_item=_on_item,
                 should_cancel=lambda: task.cancelled,
@@ -1426,14 +1391,9 @@ async def clear_online_cache_api(data: dict):
             token_list = list(dict.fromkeys(token_list))
 
             results = {}
-            batch_size = max(1, int(get_config("asset.delete_batch_size")))
-            max_concurrent = batch_size
-
             raw_results = await DeleteService.clear_assets(
                 token_list,
                 mgr,
-                max_concurrent=max_concurrent,
-                batch_size=batch_size,
             )
             for token, res in raw_results.items():
                 if res.get("ok"):
@@ -1452,8 +1412,6 @@ async def clear_online_cache_api(data: dict):
         raw_results = await DeleteService.clear_assets(
             [token],
             mgr,
-            max_concurrent=1,
-            batch_size=1,
         )
         res = raw_results.get(token, {})
         data = res.get("data", {})
@@ -1480,9 +1438,6 @@ async def clear_online_cache_api_async(data: dict):
     if not token_list:
         raise HTTPException(status_code=400, detail="No tokens provided")
 
-    batch_size = get_config("asset.delete_batch_size")
-    max_concurrent = batch_size
-
     task = create_task(len(token_list))
 
     async def _run():
@@ -1494,8 +1449,6 @@ async def clear_online_cache_api_async(data: dict):
             raw_results = await DeleteService.clear_assets(
                 token_list,
                 mgr,
-                max_concurrent=max_concurrent,
-                batch_size=batch_size,
                 include_ok=True,
                 on_item=_on_item,
                 should_cancel=lambda: task.cancelled,

app/core/batch.py

@@ -19,7 +19,6 @@ async def run_batch(
     items: List[str],
     worker: Callable[[str], Awaitable[T]],
     *,
-    max_concurrent: int = 10,
     batch_size: int = 50,
     task: Optional["BatchTask"] = None,
     on_item: Optional[Callable[[str, Dict[str, Any]], Awaitable[None]]] = None,
@@ -31,52 +30,43 @@ async def run_batch(
     Args:
         items: 待处理项列表
         worker: 异步处理函数
-        max_concurrent: 最大并发数
         batch_size: 每批大小
 
     Returns:
         {item: {"ok": bool, "data": ..., "error": ...}}
     """
-    try:
-        max_concurrent = int(max_concurrent)
-    except Exception:
-        max_concurrent = 10
     try:
         batch_size = int(batch_size)
     except Exception:
         batch_size = 50
 
-    max_concurrent = max(1, max_concurrent)
     batch_size = max(1, batch_size)
 
-    sem = asyncio.Semaphore(max_concurrent)
-
     async def _one(item: str) -> tuple[str, dict]:
         if (should_cancel and should_cancel()) or (task and task.cancelled):
             return item, {"ok": False, "error": "cancelled", "cancelled": True}
-        async with sem:
-            try:
-                data = await worker(item)
-                result = {"ok": True, "data": data}
-                if task:
-                    task.record(True)
-                if on_item:
-                    try:
-                        await on_item(item, result)
-                    except Exception:
-                        pass
-                return item, result
-            except Exception as e:
-                logger.warning(f"Batch item failed: {item[:16]}... - {e}")
-                result = {"ok": False, "error": str(e)}
-                if task:
-                    task.record(False, error=str(e))
-                if on_item:
-                    try:
-                        await on_item(item, result)
-                    except Exception:
-                        pass
-                return item, result
+        try:
+            data = await worker(item)
+            result = {"ok": True, "data": data}
+            if task:
+                task.record(True)
+            if on_item:
+                try:
+                    await on_item(item, result)
+                except Exception:
+                    pass
+            return item, result
+        except Exception as e:
+            logger.warning(f"Batch item failed: {item[:16]}... - {e}")
+            result = {"ok": False, "error": str(e)}
+            if task:
+                task.record(False, error=str(e))
+            if on_item:
+                try:
+                    await on_item(item, result)
+                except Exception:
+                    pass
+            return item, result
 
     results: Dict[str, dict] = {}
 

app/services/grok/batch_services/__init__.py

@@ -1,5 +0,0 @@
-"""Batch services."""
-
-from .usage import UsageService
-
-__all__ = ["UsageService"]

app/services/grok/batch_services/assets.py

@@ -106,8 +106,6 @@ class ListService(BaseAssetsService):
         tokens: List[str],
         account_map: dict,
         *,
-        max_concurrent: int,
-        batch_size: int,
         include_ok: bool = False,
         on_item=None,
         should_cancel=None,
@@ -115,6 +113,7 @@ class ListService(BaseAssetsService):
         """Batch fetch assets details for tokens."""
         account_map = account_map or {}
         shared_service = ListService()
+        batch_size = max(1, int(get_config("asset.list_batch_size")))
 
         async def _fetch_detail(token: str):
             account = account_map.get(token)
@@ -152,7 +151,6 @@ class ListService(BaseAssetsService):
             return await run_batch(
                 tokens,
                 _fetch_detail,
-                max_concurrent=max_concurrent,
                 batch_size=batch_size,
                 on_item=on_item,
                 should_cancel=should_cancel,
@@ -194,8 +192,6 @@ class DeleteService(BaseAssetsService):
         tokens: List[str],
         mgr,
         *,
-        max_concurrent: int,
-        batch_size: int,
         include_ok: bool = False,
         on_item=None,
         should_cancel=None,
@@ -203,6 +199,7 @@ class DeleteService(BaseAssetsService):
         """Batch clear assets for tokens."""
         delete_service = DeleteService()
         list_service = ListService()
+        batch_size = max(1, int(get_config("asset.delete_batch_size")))
 
         async def _clear_one(token: str):
             try:
@@ -222,7 +219,6 @@ class DeleteService(BaseAssetsService):
             return await run_batch(
                 tokens,
                 _clear_one,
-                max_concurrent=max_concurrent,
                 batch_size=batch_size,
                 on_item=on_item,
                 should_cancel=should_cancel,

app/services/grok/batch_services/nsfw.py

@@ -2,6 +2,7 @@
 Batch NSFW service.
 """
 
+import asyncio
 from typing import Callable, Awaitable, Dict, Any, Optional
 
 from curl_cffi.requests import AsyncSession
@@ -15,6 +16,19 @@ from app.services.reverse.set_birth import SetBirthReverse
 from app.core.batch import run_batch
 
 
+_NSFW_SEMAPHORE = None
+_NSFW_SEM_VALUE = None
+
+
+def _get_nsfw_semaphore() -> asyncio.Semaphore:
+    value = max(1, int(get_config("nsfw.concurrent")))
+    global _NSFW_SEMAPHORE, _NSFW_SEM_VALUE
+    if _NSFW_SEMAPHORE is None or value != _NSFW_SEM_VALUE:
+        _NSFW_SEM_VALUE = value
+        _NSFW_SEMAPHORE = asyncio.Semaphore(value)
+    return _NSFW_SEMAPHORE
+
+
 class NSFWService:
     """NSFW 模式服务"""
     @staticmethod
@@ -22,12 +36,11 @@ class NSFWService:
         tokens: list[str],
         mgr,
         *,
-        max_concurrent: int,
-        batch_size: int,
         on_item: Optional[Callable[[str, Dict[str, Any]], Awaitable[None]]] = None,
         should_cancel: Optional[Callable[[], bool]] = None,
     ) -> Dict[str, Dict[str, Any]]:
         """Batch enable NSFW."""
+        batch_size = get_config("nsfw.batch_size")
         async def _enable(token: str):
             try:
                 browser = get_config("security.browser")
@@ -43,7 +56,8 @@ class NSFWService:
                         return status or 0
 
                     try:
-                        await AcceptTosReverse.request(session, token)
+                        async with _get_nsfw_semaphore():
+                            await AcceptTosReverse.request(session, token)
                     except UpstreamException as e:
                         status = await _record_fail(e, "tos_auth_failed")
                         return {
@@ -53,7 +67,8 @@ class NSFWService:
                         }
 
                     try:
-                        await SetBirthReverse.request(session, token)
+                        async with _get_nsfw_semaphore():
+                            await SetBirthReverse.request(session, token)
                     except UpstreamException as e:
                         status = await _record_fail(e, "set_birth_auth_failed")
                         return {
@@ -63,7 +78,8 @@ class NSFWService:
                         }
 
                     try:
-                        grpc_status = await NsfwMgmtReverse.request(session, token)
+                        async with _get_nsfw_semaphore():
+                            grpc_status = await NsfwMgmtReverse.request(session, token)
                         success = grpc_status.code in (-1, 0)
                     except UpstreamException as e:
                         status = await _record_fail(e, "nsfw_mgmt_auth_failed")
@@ -88,7 +104,6 @@ class NSFWService:
         return await run_batch(
             tokens,
             _enable,
-            max_concurrent=max_concurrent,
             batch_size=batch_size,
             on_item=on_item,
             should_cancel=should_cancel,

app/services/grok/batch_services/usage.py

@@ -12,8 +12,17 @@ from app.core.config import get_config
 from app.services.reverse.rate_limits import RateLimitsReverse
 from app.core.batch import run_batch
 
-_USAGE_SEMAPHORE = asyncio.Semaphore(25)
-_USAGE_SEM_VALUE = 25
+_USAGE_SEMAPHORE = None
+_USAGE_SEM_VALUE = None
+
+
+def _get_usage_semaphore() -> asyncio.Semaphore:
+    value = max(1, int(get_config("usage.concurrent")))
+    global _USAGE_SEMAPHORE, _USAGE_SEM_VALUE
+    if _USAGE_SEMAPHORE is None or value != _USAGE_SEM_VALUE:
+        _USAGE_SEM_VALUE = value
+        _USAGE_SEMAPHORE = asyncio.Semaphore(value)
+    return _USAGE_SEMAPHORE
 
 
 class UsageService:
@@ -32,17 +41,7 @@ class UsageService:
         Raises:
             UpstreamException: 当获取失败且重试耗尽时
         """
-        value = get_config("usage.concurrent")
-        try:
-            value = int(value)
-        except Exception:
-            value = 25
-        value = max(1, value)
-        global _USAGE_SEMAPHORE, _USAGE_SEM_VALUE
-        if value != _USAGE_SEM_VALUE:
-            _USAGE_SEM_VALUE = value
-            _USAGE_SEMAPHORE = asyncio.Semaphore(value)
-        async with _USAGE_SEMAPHORE:
+        async with _get_usage_semaphore():
             try:
                 async with AsyncSession() as session:
                     response = await RateLimitsReverse.request(session, token)
@@ -63,18 +62,16 @@ class UsageService:
         tokens: List[str],
         mgr,
         *,
-        max_concurrent: int,
-        batch_size: int,
         on_item: Optional[Callable[[str, Dict[str, Any]], Awaitable[None]]] = None,
         should_cancel: Optional[Callable[[], bool]] = None,
     ) -> Dict[str, Dict[str, Any]]:
+        batch_size = get_config("usage.batch_size")
         async def _refresh_one(t: str):
             return await mgr.sync_usage(t, consume_on_fail=False, is_usage=False)
 
         return await run_batch(
             tokens,
             _refresh_one,
-            max_concurrent=max_concurrent,
             batch_size=batch_size,
             on_item=on_item,
             should_cancel=should_cancel,

app/services/reverse/rate_limits.py

@@ -50,8 +50,6 @@ class RateLimitsReverse:
 
             # Curl Config
             timeout = get_config("usage.timeout")
-            if timeout is None:
-                timeout = get_config("network.timeout")
             browser = get_config("security.browser")
 
             async def _do_request():