From 6257b68527f4e80975fa396bae8fa8b190ba988f Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Fri, 13 Feb 2026 14:21:11 +0800 Subject: [PATCH] refactor: remove unused files and streamline batch processing configuration in services --- app/__init__.py | 1 - app/api/v1/admin.py | 47 ----------------- app/core/batch.py | 54 ++++++++------------ app/services/__init__.py | 0 app/services/grok/__init__.py | 0 app/services/grok/batch_services/__init__.py | 5 -- app/services/grok/batch_services/assets.py | 8 +-- app/services/grok/batch_services/nsfw.py | 27 +++++++--- app/services/grok/batch_services/usage.py | 29 +++++------ app/services/reverse/rate_limits.py | 2 - 10 files changed, 58 insertions(+), 115 deletions(-) delete mode 100644 app/__init__.py delete mode 100644 app/services/__init__.py delete mode 100644 app/services/grok/__init__.py delete mode 100644 app/services/grok/batch_services/__init__.py diff --git a/app/__init__.py b/app/__init__.py deleted file mode 100644 index 9f7a423..0000000 --- a/app/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""App Package""" diff --git a/app/api/v1/admin.py b/app/api/v1/admin.py index 6daa3d9..a05db4a 100644 --- a/app/api/v1/admin.py +++ b/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, diff --git a/app/core/batch.py b/app/core/batch.py index 4857250..7c62c01 100644 --- a/app/core/batch.py +++ b/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] = {} diff --git a/app/services/__init__.py b/app/services/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/app/services/grok/__init__.py b/app/services/grok/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/app/services/grok/batch_services/__init__.py b/app/services/grok/batch_services/__init__.py deleted file mode 100644 index d71f450..0000000 --- a/app/services/grok/batch_services/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Batch services.""" - -from .usage import UsageService - -__all__ = ["UsageService"] diff --git a/app/services/grok/batch_services/assets.py b/app/services/grok/batch_services/assets.py index ca87b3c..7c3c31e 100644 --- a/app/services/grok/batch_services/assets.py +++ b/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, diff --git a/app/services/grok/batch_services/nsfw.py b/app/services/grok/batch_services/nsfw.py index facfbe5..e7761e2 100644 --- a/app/services/grok/batch_services/nsfw.py +++ b/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, diff --git a/app/services/grok/batch_services/usage.py b/app/services/grok/batch_services/usage.py index 9d10c98..66aab10 100644 --- a/app/services/grok/batch_services/usage.py +++ b/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, diff --git a/app/services/reverse/rate_limits.py b/app/services/reverse/rate_limits.py index 48b8c22..198164c 100644 --- a/app/services/reverse/rate_limits.py +++ b/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():