From b11aace053485ca8cef28e14a6ee9f59a46c7d75 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Tue, 3 Feb 2026 23:15:13 +0800 Subject: [PATCH] feat: optimize frontend batch processing logic --- app/api/v1/admin.py | 774 +++++++++++++++++++++++++++++++-- app/core/batch_tasks.py | 151 +++++++ app/services/grok/batch.py | 24 +- app/static/cache/cache.html | 13 +- app/static/cache/cache.js | 731 ++++++++++++++++++------------- app/static/common/batch-sse.js | 46 ++ app/static/common/common.css | 12 + app/static/common/footer.html | 2 +- app/static/common/header.html | 2 +- app/static/common/header.js | 2 +- app/static/config/config.html | 8 +- app/static/config/config.js | 258 ++++++----- app/static/login/login.html | 2 +- app/static/login/login.js | 39 +- app/static/token/token.css | 47 +- app/static/token/token.html | 31 +- app/static/token/token.js | 660 ++++++++++++++++++---------- config.defaults.toml | 7 +- data/config.toml | 8 +- docs/README.en.md | 14 +- readme.md | 19 +- 21 files changed, 2094 insertions(+), 756 deletions(-) create mode 100644 app/core/batch_tasks.py create mode 100644 app/static/common/batch-sse.js diff --git a/app/api/v1/admin.py b/app/api/v1/admin.py index 7ea3bba..2e8651d 100644 --- a/app/api/v1/admin.py +++ b/app/api/v1/admin.py @@ -1,12 +1,14 @@ from fastapi import APIRouter, Depends, HTTPException, Request, Query -from fastapi.responses import HTMLResponse +from fastapi.responses import HTMLResponse, StreamingResponse from app.core.auth import verify_api_key, verify_app_key from app.core.config import config, get_config +from app.core.batch_tasks import create_task, get_task, expire_task from app.core.storage import get_storage, LocalStorage, RedisStorage, SQLStorage import os from pathlib import Path import aiofiles import asyncio +import orjson from app.core.logger import logger @@ -26,6 +28,65 @@ async def render_template(filename: str): return HTMLResponse(content) +def _sse_event(payload: dict) -> str: + return f"data: {orjson.dumps(payload).decode()}\n\n" + + +def _verify_stream_api_key(request: Request) -> None: + api_key = get_config("app.api_key", "") + if not api_key: + return + key = request.query_params.get("api_key") + if key != api_key: + raise HTTPException(status_code=401, detail="Invalid authentication token") + + +@router.get("/api/v1/admin/batch/{task_id}/stream") +async def stream_batch(task_id: str, request: Request): + _verify_stream_api_key(request) + task = get_task(task_id) + if not task: + raise HTTPException(status_code=404, detail="Task not found") + + async def event_stream(): + queue = task.attach() + try: + yield _sse_event({"type": "snapshot", **task.snapshot()}) + + final = task.final_event() + if final: + yield _sse_event(final) + return + + while True: + try: + event = await asyncio.wait_for(queue.get(), timeout=15) + except asyncio.TimeoutError: + yield ": ping\n\n" + final = task.final_event() + if final: + yield _sse_event(final) + return + continue + + yield _sse_event(event) + if event.get("type") in ("done", "error", "cancelled"): + return + finally: + task.detach(queue) + + return StreamingResponse(event_stream(), media_type="text/event-stream") + + +@router.post("/api/v1/admin/batch/{task_id}/cancel", dependencies=[Depends(verify_api_key)]) +async def cancel_batch(task_id: str): + task = get_task(task_id) + if not task: + raise HTTPException(status_code=404, detail="Task not found") + task.cancel() + return {"status": "success"} + + @router.get("/admin", response_class=HTMLResponse, include_in_schema=False) async def admin_login_page(): """管理后台登录页""" @@ -118,6 +179,7 @@ async def update_tokens_api(data: dict): async def refresh_tokens_api(data: dict): """刷新 Token 状态""" from app.services.token.manager import get_token_manager + from app.services.grok.batch import run_in_batches try: mgr = await get_token_manager() @@ -130,22 +192,158 @@ async def refresh_tokens_api(data: dict): if not tokens: raise HTTPException(status_code=400, detail="No tokens provided") - unique_tokens = list(set(tokens)) + # 去重并保持顺序 + unique_tokens = list(dict.fromkeys(tokens)) - sem = asyncio.Semaphore(10) + # 最大数量限制 + max_tokens = get_config("performance.usage_max_tokens", 1000) + try: + max_tokens = int(max_tokens) + except Exception: + max_tokens = 1000 + + truncated = False + original_count = len(unique_tokens) + if len(unique_tokens) > max_tokens: + unique_tokens = unique_tokens[:max_tokens] + truncated = True + logger.warning( + f"Usage refresh: truncated from {original_count} to {max_tokens} tokens" + ) + + # 批量执行配置 + max_concurrent = get_config("performance.usage_max_concurrent", 25) + batch_size = get_config("performance.usage_batch_size", 50) async def _refresh_one(t): - async with sem: - return t, await mgr.sync_usage( + return await mgr.sync_usage( + t, "grok-3", consume_on_fail=False, is_usage=False + ) + + raw_results = await run_in_batches( + unique_tokens, _refresh_one, max_concurrent=max_concurrent, batch_size=batch_size + ) + + results = {} + for token, res in raw_results.items(): + if res.get("ok"): + results[token] = res.get("data", False) + else: + results[token] = False + + response = {"status": "success", "results": results} + if truncated: + response["warning"] = ( + f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）" + ) + return response + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post( + "/api/v1/admin/tokens/refresh/async", dependencies=[Depends(verify_api_key)] +) +async def refresh_tokens_api_async(data: dict): + """刷新 Token 状态（异步批量 + SSE 进度）""" + from app.services.token.manager import get_token_manager + from app.services.grok.batch import run_in_batches + + mgr = await get_token_manager() + tokens: list[str] = [] + if isinstance(data.get("token"), str) and data["token"].strip(): + tokens.append(data["token"].strip()) + if isinstance(data.get("tokens"), list): + tokens.extend([str(t).strip() for t in data["tokens"] if str(t).strip()]) + + if not tokens: + raise HTTPException(status_code=400, detail="No tokens provided") + + unique_tokens = list(dict.fromkeys(tokens)) + + max_tokens = get_config("performance.usage_max_tokens", 1000) + try: + max_tokens = int(max_tokens) + except Exception: + max_tokens = 1000 + + truncated = False + original_count = len(unique_tokens) + if len(unique_tokens) > max_tokens: + unique_tokens = unique_tokens[:max_tokens] + truncated = True + logger.warning( + f"Usage refresh: truncated from {original_count} to {max_tokens} tokens" + ) + + max_concurrent = get_config("performance.usage_max_concurrent", 25) + batch_size = get_config("performance.usage_batch_size", 50) + + task = create_task(len(unique_tokens)) + + async def _run(): + try: + async def _refresh_one(t: str): + return await mgr.sync_usage( t, "grok-3", consume_on_fail=False, is_usage=False ) - results_list = await asyncio.gather(*[_refresh_one(t) for t in unique_tokens]) - results = dict(results_list) + async def _on_item(item: str, res: dict): + task.record(bool(res.get("ok"))) - return {"status": "success", "results": results} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + raw_results = await run_in_batches( + unique_tokens, + _refresh_one, + max_concurrent=max_concurrent, + batch_size=batch_size, + on_item=_on_item, + should_cancel=lambda: task.cancelled, + ) + + if task.cancelled: + task.finish_cancelled() + return + + results: dict[str, bool] = {} + ok_count = 0 + fail_count = 0 + for token, res in raw_results.items(): + if res.get("ok") and res.get("data") is True: + ok_count += 1 + results[token] = True + else: + fail_count += 1 + results[token] = False + + await mgr._save() + + result = { + "status": "success", + "summary": { + "total": len(unique_tokens), + "ok": ok_count, + "fail": fail_count, + }, + "results": results, + } + warning = None + if truncated: + warning = ( + f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）" + ) + task.finish(result, warning=warning) + except Exception as e: + task.fail_task(str(e)) + finally: + asyncio.create_task(expire_task(task.id, 300)) + + asyncio.create_task(_run()) + + return { + "status": "success", + "task_id": task.id, + "total": len(unique_tokens), + } @router.post("/api/v1/admin/tokens/nsfw/enable", dependencies=[Depends(verify_api_key)]) @@ -199,7 +397,7 @@ async def enable_nsfw_api(data: dict): # 批量执行配置 max_concurrent = get_config("performance.nsfw_max_concurrent", 10) - batch_size = get_config("performance.admin_nsfw_batch_size", 50) + batch_size = get_config("performance.nsfw_batch_size", 50) # 定义 worker async def _enable(token: str): @@ -257,6 +455,129 @@ async def enable_nsfw_api(data: dict): raise HTTPException(status_code=500, detail=str(e)) +@router.post( + "/api/v1/admin/tokens/nsfw/enable/async", dependencies=[Depends(verify_api_key)] +) +async def enable_nsfw_api_async(data: dict): + """批量开启 NSFW (Unhinged) 模式（异步批量 + SSE 进度）""" + from app.services.grok.nsfw import NSFWService + from app.services.grok.batch import run_in_batches + from app.services.token.manager import get_token_manager + + mgr = await get_token_manager() + nsfw_service = NSFWService() + + tokens: list[str] = [] + if isinstance(data.get("token"), str) and data["token"].strip(): + tokens.append(data["token"].strip()) + if isinstance(data.get("tokens"), list): + tokens.extend([str(t).strip() for t in data["tokens"] if str(t).strip()]) + + if not tokens: + for pool_name, pool in mgr.pools.items(): + for info in pool.list(): + raw = info.token[4:] if info.token.startswith("sso=") else info.token + tokens.append(raw) + + if not tokens: + raise HTTPException(status_code=400, detail="No tokens available") + + unique_tokens = list(dict.fromkeys(tokens)) + + max_tokens = get_config("performance.nsfw_max_tokens", 1000) + try: + max_tokens = int(max_tokens) + except Exception: + max_tokens = 1000 + + truncated = False + original_count = len(unique_tokens) + if len(unique_tokens) > max_tokens: + unique_tokens = unique_tokens[:max_tokens] + truncated = True + logger.warning( + f"NSFW enable: truncated from {original_count} to {max_tokens} tokens" + ) + + max_concurrent = get_config("performance.nsfw_max_concurrent", 10) + batch_size = get_config("performance.nsfw_batch_size", 50) + + task = create_task(len(unique_tokens)) + + async def _run(): + try: + async def _enable(token: str): + result = await nsfw_service.enable(token) + if result.success: + await mgr.add_tag(token, "nsfw") + return { + "success": result.success, + "http_status": result.http_status, + "grpc_status": result.grpc_status, + "grpc_message": result.grpc_message, + "error": result.error, + } + + async def _on_item(item: str, res: dict): + ok = bool(res.get("ok") and res.get("data", {}).get("success")) + task.record(ok) + + raw_results = await run_in_batches( + unique_tokens, + _enable, + max_concurrent=max_concurrent, + batch_size=batch_size, + on_item=_on_item, + should_cancel=lambda: task.cancelled, + ) + + if task.cancelled: + task.finish_cancelled() + return + + results = {} + ok_count = 0 + fail_count = 0 + for token, res in raw_results.items(): + masked = f"{token[:8]}...{token[-8:]}" if len(token) > 20 else token + if res.get("ok") and res.get("data", {}).get("success"): + ok_count += 1 + results[masked] = res.get("data", {}) + else: + fail_count += 1 + results[masked] = res.get("data") or {"error": res.get("error")} + + await mgr._save() + + result = { + "status": "success", + "summary": { + "total": len(unique_tokens), + "ok": ok_count, + "fail": fail_count, + }, + "results": results, + } + warning = None + if truncated: + warning = ( + f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）" + ) + task.finish(result, warning=warning) + except Exception as e: + task.fail_task(str(e)) + finally: + asyncio.create_task(expire_task(task.id, 300)) + + asyncio.create_task(_run()) + + return { + "status": "success", + "task_id": task.id, + "total": len(unique_tokens), + } + + @router.get("/admin/cache", response_class=HTMLResponse, include_in_schema=False) async def admin_cache_page(): """缓存管理页""" @@ -268,6 +589,7 @@ async def get_cache_stats_api(request: Request): """获取缓存统计""" from app.services.grok.assets import DownloadService, ListService from app.services.token.manager import get_token_manager + from app.services.grok.batch import run_in_batches try: dl_service = DownloadService() @@ -312,13 +634,28 @@ async def get_cache_stats_api(request: Request): } online_details = [] account_map = {a["token"]: a for a in accounts} - batch_size = get_config("performance.admin_assets_batch_size", 10) + max_concurrent = get_config("performance.assets_max_concurrent", 25) + batch_size = get_config("performance.assets_batch_size", 10) + try: + max_concurrent = int(max_concurrent) + except Exception: + max_concurrent = 25 try: batch_size = int(batch_size) except Exception: batch_size = 10 + max_concurrent = max(1, max_concurrent) batch_size = max(1, batch_size) + max_tokens = get_config("performance.assets_max_tokens", 1000) + try: + max_tokens = int(max_tokens) + except Exception: + max_tokens = 1000 + + truncated = False + original_count = 0 + async def _fetch_assets(token: str): list_service = ListService() try: @@ -330,8 +667,8 @@ async def get_cache_stats_api(request: Request): account = account_map.get(token) try: count = await _fetch_assets(token) - return ( - { + return { + "detail": { "token": token, "token_masked": account["token_masked"] if account else token, "count": count, @@ -340,11 +677,11 @@ async def get_cache_stats_api(request: Request): if account else None, }, - count, - ) + "count": count, + } except Exception as e: - return ( - { + return { + "detail": { "token": token, "token_masked": account["token_masked"] if account else token, "count": 0, @@ -353,19 +690,40 @@ async def get_cache_stats_api(request: Request): if account else None, }, - 0, - ) + "count": 0, + } if selected_tokens: + selected_tokens = list(dict.fromkeys(selected_tokens)) + original_count = len(selected_tokens) + if len(selected_tokens) > max_tokens: + selected_tokens = selected_tokens[:max_tokens] + truncated = True total = 0 - for i in range(0, len(selected_tokens), batch_size): - chunk = selected_tokens[i : i + batch_size] - results = await asyncio.gather( - *[_fetch_detail(token) for token in chunk] - ) - for detail, count in results: + raw_results = await run_in_batches( + selected_tokens, + _fetch_detail, + max_concurrent=max_concurrent, + batch_size=batch_size, + ) + for token, res in raw_results.items(): + if res.get("ok"): + data = res.get("data", {}) + detail = data.get("detail") + total += data.get("count", 0) + else: + account = account_map.get(token) + detail = { + "token": token, + "token_masked": account["token_masked"] if account else token, + "count": 0, + "status": f"error: {res.get('error')}", + "last_asset_clear_at": account["last_asset_clear_at"] + if account + else None, + } + if detail: online_details.append(detail) - total += count online_stats = { "count": total, "status": "ok" if selected_tokens else "no_token", @@ -375,15 +733,35 @@ async def get_cache_stats_api(request: Request): scope = "selected" elif scope == "all": total = 0 - tokens = [account["token"] for account in accounts] - for i in range(0, len(tokens), batch_size): - chunk = tokens[i : i + batch_size] - results = await asyncio.gather( - *[_fetch_detail(token) for token in chunk] - ) - for detail, count in results: + tokens = list(dict.fromkeys([account["token"] for account in accounts])) + original_count = len(tokens) + if len(tokens) > max_tokens: + tokens = tokens[:max_tokens] + truncated = True + raw_results = await run_in_batches( + tokens, + _fetch_detail, + max_concurrent=max_concurrent, + batch_size=batch_size, + ) + for token, res in raw_results.items(): + if res.get("ok"): + data = res.get("data", {}) + detail = data.get("detail") + total += data.get("count", 0) + else: + account = account_map.get(token) + detail = { + "token": token, + "token_masked": account["token_masked"] if account else token, + "count": 0, + "status": f"error: {res.get('error')}", + "last_asset_clear_at": account["last_asset_clear_at"] + if account + else None, + } + if detail: online_details.append(detail) - total += count online_stats = { "count": total, "status": "ok" if accounts else "no_token", @@ -424,7 +802,7 @@ async def get_cache_stats_api(request: Request): "last_asset_clear_at": None, } - return { + response = { "local_image": image_stats, "local_video": video_stats, "online": online_stats, @@ -432,10 +810,177 @@ async def get_cache_stats_api(request: Request): "online_scope": scope or "none", "online_details": online_details, } + if truncated: + response["warning"] = ( + f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）" + ) + return response except Exception as e: raise HTTPException(status_code=500, detail=str(e)) +@router.post( + "/api/v1/admin/cache/online/load/async", dependencies=[Depends(verify_api_key)] +) +async def load_online_cache_api_async(data: dict): + """在线资产统计（异步批量 + SSE 进度）""" + from app.services.grok.assets import DownloadService, ListService + from app.services.token.manager import get_token_manager + from app.services.grok.batch import run_in_batches + + mgr = await get_token_manager() + + # 账号列表 + accounts = [] + for pool_name, pool in mgr.pools.items(): + for info in pool.list(): + raw_token = info.token[4:] if info.token.startswith("sso=") else info.token + masked = ( + f"{raw_token[:8]}...{raw_token[-16:]}" + if len(raw_token) > 24 + else raw_token + ) + accounts.append( + { + "token": raw_token, + "token_masked": masked, + "pool": pool_name, + "status": info.status, + "last_asset_clear_at": info.last_asset_clear_at, + } + ) + + account_map = {a["token"]: a for a in accounts} + + tokens = data.get("tokens") + scope = data.get("scope") + selected_tokens: list[str] = [] + if isinstance(tokens, list): + selected_tokens = [str(t).strip() for t in tokens if str(t).strip()] + + if not selected_tokens and scope == "all": + selected_tokens = [account["token"] for account in accounts] + scope = "all" + elif selected_tokens: + scope = "selected" + else: + raise HTTPException(status_code=400, detail="No tokens provided") + + selected_tokens = list(dict.fromkeys(selected_tokens)) + + max_tokens = get_config("performance.assets_max_tokens", 1000) + try: + max_tokens = int(max_tokens) + except Exception: + max_tokens = 1000 + + truncated = False + original_count = len(selected_tokens) + if len(selected_tokens) > max_tokens: + selected_tokens = selected_tokens[:max_tokens] + truncated = True + + max_concurrent = get_config("performance.assets_max_concurrent", 25) + batch_size = get_config("performance.assets_batch_size", 10) + + task = create_task(len(selected_tokens)) + + async def _run(): + try: + dl_service = DownloadService() + image_stats = dl_service.get_stats("image") + video_stats = dl_service.get_stats("video") + + async def _fetch_detail(token: str): + account = account_map.get(token) + list_service = ListService() + try: + count = await list_service.count(token) + detail = { + "token": token, + "token_masked": account["token_masked"] if account else token, + "count": count, + "status": "ok", + "last_asset_clear_at": account["last_asset_clear_at"] + if account + else None, + } + return {"ok": True, "detail": detail, "count": count} + except Exception as e: + detail = { + "token": token, + "token_masked": account["token_masked"] if account else token, + "count": 0, + "status": f"error: {str(e)}", + "last_asset_clear_at": account["last_asset_clear_at"] + if account + else None, + } + return {"ok": False, "detail": detail, "count": 0} + finally: + await list_service.close() + + async def _on_item(item: str, res: dict): + ok = bool(res.get("data", {}).get("ok")) + task.record(ok) + + raw_results = await run_in_batches( + selected_tokens, + _fetch_detail, + max_concurrent=max_concurrent, + batch_size=batch_size, + on_item=_on_item, + should_cancel=lambda: task.cancelled, + ) + + if task.cancelled: + task.finish_cancelled() + return + + online_details = [] + total = 0 + for token, res in raw_results.items(): + data = res.get("data", {}) + detail = data.get("detail") + if detail: + online_details.append(detail) + total += data.get("count", 0) + + online_stats = { + "count": total, + "status": "ok" if selected_tokens else "no_token", + "token": None, + "last_asset_clear_at": None, + } + + result = { + "local_image": image_stats, + "local_video": video_stats, + "online": online_stats, + "online_accounts": accounts, + "online_scope": scope or "none", + "online_details": online_details, + } + warning = None + if truncated: + warning = ( + f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）" + ) + task.finish(result, warning=warning) + except Exception as e: + task.fail_task(str(e)) + finally: + asyncio.create_task(expire_task(task.id, 300)) + + asyncio.create_task(_run()) + + return { + "status": "success", + "task_id": task.id, + "total": len(selected_tokens), + } + + @router.post("/api/v1/admin/cache/clear", dependencies=[Depends(verify_api_key)]) async def clear_local_cache_api(data: dict): """清理本地缓存""" @@ -493,6 +1038,7 @@ async def clear_online_cache_api(data: dict): """清理在线缓存""" from app.services.grok.assets import DeleteService from app.services.token.manager import get_token_manager + from app.services.grok.batch import run_in_batches delete_service = None try: @@ -505,29 +1051,61 @@ async def clear_online_cache_api(data: dict): if not token_list: raise HTTPException(status_code=400, detail="No tokens provided") + # 去重并保持顺序 + token_list = list(dict.fromkeys(token_list)) + + # 最大数量限制 + max_tokens = get_config("performance.assets_max_tokens", 1000) + try: + max_tokens = int(max_tokens) + except Exception: + max_tokens = 1000 + truncated = False + original_count = len(token_list) + if len(token_list) > max_tokens: + token_list = token_list[:max_tokens] + truncated = True + results = {} - batch_size = get_config("performance.admin_assets_batch_size", 10) + max_concurrent = get_config("performance.assets_max_concurrent", 25) + batch_size = get_config("performance.assets_batch_size", 10) + try: + max_concurrent = int(max_concurrent) + except Exception: + max_concurrent = 25 try: batch_size = int(batch_size) except Exception: batch_size = 10 + max_concurrent = max(1, max_concurrent) batch_size = max(1, batch_size) async def _clear_one(t: str): try: result = await delete_service.delete_all(t) await mgr.mark_asset_clear(t) - return t, {"status": "success", "result": result} + return {"status": "success", "result": result} except Exception as e: - return t, {"status": "error", "error": str(e)} + return {"status": "error", "error": str(e)} - for i in range(0, len(token_list), batch_size): - chunk = token_list[i : i + batch_size] - res_list = await asyncio.gather(*[_clear_one(t) for t in chunk]) - for t, res in res_list: - results[t] = res + raw_results = await run_in_batches( + token_list, + _clear_one, + max_concurrent=max_concurrent, + batch_size=batch_size, + ) + for token, res in raw_results.items(): + if res.get("ok"): + results[token] = res.get("data", {}) + else: + results[token] = {"status": "error", "error": res.get("error")} - return {"status": "success", "results": results} + response = {"status": "success", "results": results} + if truncated: + response["warning"] = ( + f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）" + ) + return response token = data.get("token") or mgr.get_token() if not token: @@ -543,3 +1121,109 @@ async def clear_online_cache_api(data: dict): finally: if delete_service: await delete_service.close() + + +@router.post( + "/api/v1/admin/cache/online/clear/async", dependencies=[Depends(verify_api_key)] +) +async def clear_online_cache_api_async(data: dict): + """清理在线缓存（异步批量 + SSE 进度）""" + from app.services.grok.assets import DeleteService + from app.services.token.manager import get_token_manager + from app.services.grok.batch import run_in_batches + + mgr = await get_token_manager() + tokens = data.get("tokens") + if not isinstance(tokens, list): + raise HTTPException(status_code=400, detail="No tokens provided") + + token_list = [t.strip() for t in tokens if isinstance(t, str) and t.strip()] + if not token_list: + raise HTTPException(status_code=400, detail="No tokens provided") + + token_list = list(dict.fromkeys(token_list)) + + max_tokens = get_config("performance.assets_max_tokens", 1000) + try: + max_tokens = int(max_tokens) + except Exception: + max_tokens = 1000 + truncated = False + original_count = len(token_list) + if len(token_list) > max_tokens: + token_list = token_list[:max_tokens] + truncated = True + + max_concurrent = get_config("performance.assets_max_concurrent", 25) + batch_size = get_config("performance.assets_batch_size", 10) + + task = create_task(len(token_list)) + + async def _run(): + delete_service = DeleteService() + try: + async def _clear_one(t: str): + try: + result = await delete_service.delete_all(t) + await mgr.mark_asset_clear(t) + return {"ok": True, "result": result} + except Exception as e: + return {"ok": False, "error": str(e)} + + async def _on_item(item: str, res: dict): + ok = bool(res.get("data", {}).get("ok")) + task.record(ok) + + raw_results = await run_in_batches( + token_list, + _clear_one, + max_concurrent=max_concurrent, + batch_size=batch_size, + on_item=_on_item, + should_cancel=lambda: task.cancelled, + ) + + if task.cancelled: + task.finish_cancelled() + return + + results = {} + ok_count = 0 + fail_count = 0 + for token, res in raw_results.items(): + data = res.get("data", {}) + if data.get("ok"): + ok_count += 1 + results[token] = {"status": "success", "result": data.get("result")} + else: + fail_count += 1 + results[token] = {"status": "error", "error": data.get("error")} + + result = { + "status": "success", + "summary": { + "total": len(token_list), + "ok": ok_count, + "fail": fail_count, + }, + "results": results, + } + warning = None + if truncated: + warning = ( + f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）" + ) + task.finish(result, warning=warning) + except Exception as e: + task.fail_task(str(e)) + finally: + await delete_service.close() + asyncio.create_task(expire_task(task.id, 300)) + + asyncio.create_task(_run()) + + return { + "status": "success", + "task_id": task.id, + "total": len(token_list), + } diff --git a/app/core/batch_tasks.py b/app/core/batch_tasks.py new file mode 100644 index 0000000..07038ff --- /dev/null +++ b/app/core/batch_tasks.py @@ -0,0 +1,151 @@ +""" +Batch task manager for admin batch operations (SSE progress). +""" + +from __future__ import annotations + +import asyncio +import time +import uuid +from typing import Any, Dict, List, Optional + + +class BatchTask: + def __init__(self, total: int): + self.id = uuid.uuid4().hex + self.total = int(total) + self.processed = 0 + self.ok = 0 + self.fail = 0 + self.status = "running" + self.warning: Optional[str] = None + self.result: Optional[Dict[str, Any]] = None + self.error: Optional[str] = None + self.created_at = time.time() + self._queues: List[asyncio.Queue] = [] + self._final_event: Optional[Dict[str, Any]] = None + self.cancelled = False + + def snapshot(self) -> Dict[str, Any]: + return { + "task_id": self.id, + "status": self.status, + "total": self.total, + "processed": self.processed, + "ok": self.ok, + "fail": self.fail, + "warning": self.warning, + } + + def attach(self) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=200) + self._queues.append(q) + return q + + def detach(self, q: asyncio.Queue) -> None: + if q in self._queues: + self._queues.remove(q) + + def _publish(self, event: Dict[str, Any]) -> None: + for q in list(self._queues): + try: + q.put_nowait(event) + except Exception: + # Drop if queue is full or closed + pass + + def record(self, ok: bool, *, item: Any = None, detail: Any = None, error: str = "") -> None: + self.processed += 1 + if ok: + self.ok += 1 + else: + self.fail += 1 + event: Dict[str, Any] = { + "type": "progress", + "task_id": self.id, + "total": self.total, + "processed": self.processed, + "ok": self.ok, + "fail": self.fail, + } + if item is not None: + event["item"] = item + if detail is not None: + event["detail"] = detail + if error: + event["error"] = error + self._publish(event) + + def finish(self, result: Dict[str, Any], *, warning: Optional[str] = None) -> None: + self.status = "done" + self.result = result + self.warning = warning + event = { + "type": "done", + "task_id": self.id, + "total": self.total, + "processed": self.processed, + "ok": self.ok, + "fail": self.fail, + "warning": self.warning, + "result": result, + } + self._final_event = event + self._publish(event) + + def fail_task(self, error: str) -> None: + self.status = "error" + self.error = error + event = { + "type": "error", + "task_id": self.id, + "total": self.total, + "processed": self.processed, + "ok": self.ok, + "fail": self.fail, + "error": error, + } + self._final_event = event + self._publish(event) + + def cancel(self) -> None: + self.cancelled = True + + def finish_cancelled(self) -> None: + self.status = "cancelled" + event = { + "type": "cancelled", + "task_id": self.id, + "total": self.total, + "processed": self.processed, + "ok": self.ok, + "fail": self.fail, + } + self._final_event = event + self._publish(event) + + def final_event(self) -> Optional[Dict[str, Any]]: + return self._final_event + + +_TASKS: Dict[str, BatchTask] = {} + + +def create_task(total: int) -> BatchTask: + task = BatchTask(total) + _TASKS[task.id] = task + return task + + +def get_task(task_id: str) -> Optional[BatchTask]: + return _TASKS.get(task_id) + + +def delete_task(task_id: str) -> None: + _TASKS.pop(task_id, None) + + +async def expire_task(task_id: str, delay: int = 300) -> None: + await asyncio.sleep(delay) + delete_task(task_id) + diff --git a/app/services/grok/batch.py b/app/services/grok/batch.py index 899d6f2..adb64ea 100644 --- a/app/services/grok/batch.py +++ b/app/services/grok/batch.py @@ -5,7 +5,7 @@ """ import asyncio -from typing import Any, Awaitable, Callable, Dict, List, TypeVar +from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar from app.core.logger import logger @@ -18,6 +18,8 @@ async def run_in_batches( *, max_concurrent: int = 10, batch_size: int = 50, + on_item: Optional[Callable[[str, Dict[str, Any]], Awaitable[None]]] = None, + should_cancel: Optional[Callable[[], bool]] = None, ) -> Dict[str, Dict[str, Any]]: """ 分批并发执行，单项失败不影响整体 @@ -46,18 +48,34 @@ async def run_in_batches( sem = asyncio.Semaphore(max_concurrent) async def _one(item: str) -> tuple[str, dict]: + if should_cancel and should_cancel(): + return item, {"ok": False, "error": "cancelled", "cancelled": True} async with sem: try: data = await worker(item) - return item, {"ok": True, "data": data} + result = {"ok": True, "data": data} + 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}") - return item, {"ok": False, "error": str(e)} + result = {"ok": False, "error": str(e)} + if on_item: + try: + await on_item(item, result) + except Exception: + pass + return item, result results: Dict[str, dict] = {} # 分批执行，避免一次性创建所有 task for i in range(0, len(items), batch_size): + if should_cancel and should_cancel(): + break chunk = items[i : i + batch_size] pairs = await asyncio.gather(*(_one(x) for x in chunk)) results.update(dict(pairs)) diff --git a/app/static/cache/cache.html b/app/static/cache/cache.html index af31c87..b66e14f 100644 --- a/app/static/cache/cache.html +++ b/app/static/cache/cache.html @@ -9,15 +9,15 @@ - - + +
-
+
@@ -141,7 +141,7 @@ - +
@@ -197,10 +197,11 @@ - + + - + diff --git a/app/static/cache/cache.js b/app/static/cache/cache.js index 7a04c5c..40b99d6 100644 --- a/app/static/cache/cache.js +++ b/app/static/cache/cache.js @@ -9,6 +9,7 @@ const selectedLocal = { video: new Set() }; const ui = {}; +const byId = (id) => document.getElementById(id); const loadFailed = new Map(); const deleteFailed = new Map(); let currentBatchAction = null; @@ -18,6 +19,69 @@ const cacheListState = { image: { loaded: false, visible: false, items: [] }, video: { loaded: false, visible: false, items: [] } }; +const UI_MAP = { + imgCount: 'img-count', + imgSize: 'img-size', + videoCount: 'video-count', + videoSize: 'video-size', + onlineCount: 'online-count', + onlineStatus: 'online-status', + onlineLastClear: 'online-last-clear', + accountTableBody: 'account-table-body', + accountEmpty: 'account-empty', + selectAll: 'select-all', + localImageSelectAll: 'local-image-select-all', + localVideoSelectAll: 'local-video-select-all', + selectedCount: 'selected-count', + batchActions: 'batch-actions', + loadBtn: 'btn-load-stats', + deleteBtn: 'btn-delete-assets', + localCacheLists: 'local-cache-lists', + localImageList: 'local-image-list', + localVideoList: 'local-video-list', + localImageBody: 'local-image-body', + localVideoBody: 'local-video-body', + onlineAssetsTable: 'online-assets-table', + batchProgress: 'batch-progress', + batchProgressText: 'batch-progress-text', + pauseActionBtn: 'btn-pause-action', + stopActionBtn: 'btn-stop-action', + failureDetailsBtn: 'btn-failure-details', + confirmDialog: 'confirm-dialog', + confirmMessage: 'confirm-message', + confirmOk: 'confirm-ok', + confirmCancel: 'confirm-cancel', + failureDialog: 'failure-dialog', + failureList: 'failure-list', + failureClose: 'failure-close', + failureRetry: 'failure-retry' +}; + +function setText(el, text) { + if (el) el.textContent = text; +} + +function resolveOnlineStatus(status) { + if (status === 'ok') { + return { text: '连接正常', className: 'text-xs text-green-600 mt-1' }; + } + if (status === 'no_token') { + return { text: '无可用 Token', className: 'text-xs text-orange-500 mt-1' }; + } + if (status === 'not_loaded') { + return { text: '未加载', className: 'text-xs text-[var(--accents-4)] mt-1' }; + } + return { text: '无法连接', className: 'text-xs text-red-500 mt-1' }; +} + +function createIconButton(title, svg, onClick) { + const btn = document.createElement('button'); + btn.className = 'cache-icon-button'; + btn.title = title; + btn.innerHTML = svg; + btn.addEventListener('click', onClick); + return btn; +} async function init() { apiKey = await ensureApiKey(); @@ -42,42 +106,10 @@ function setupCacheCards() { } function cacheUI() { - ui.imgCount = document.getElementById('img-count'); - ui.imgSize = document.getElementById('img-size'); - ui.videoCount = document.getElementById('video-count'); - ui.videoSize = document.getElementById('video-size'); - ui.onlineCount = document.getElementById('online-count'); - ui.onlineStatus = document.getElementById('online-status'); - ui.onlineLastClear = document.getElementById('online-last-clear'); - ui.accountTableBody = document.getElementById('account-table-body'); - ui.accountEmpty = document.getElementById('account-empty'); - ui.selectAll = document.getElementById('select-all'); - ui.localImageSelectAll = document.getElementById('local-image-select-all'); - ui.localVideoSelectAll = document.getElementById('local-video-select-all'); - ui.selectedCount = document.getElementById('selected-count'); - ui.batchActions = document.getElementById('batch-actions'); - ui.loadBtn = document.getElementById('btn-load-stats'); - ui.deleteBtn = document.getElementById('btn-delete-assets'); - ui.localCacheLists = document.getElementById('local-cache-lists'); - ui.localImageList = document.getElementById('local-image-list'); - ui.localVideoList = document.getElementById('local-video-list'); - ui.localImageBody = document.getElementById('local-image-body'); - ui.localVideoBody = document.getElementById('local-video-body'); + Object.entries(UI_MAP).forEach(([key, id]) => { + ui[key] = byId(id); + }); ui.cacheCards = document.querySelectorAll('.cache-card'); - ui.onlineAssetsTable = document.getElementById('online-assets-table'); - ui.batchProgress = document.getElementById('batch-progress'); - ui.batchProgressText = document.getElementById('batch-progress-text'); - ui.pauseActionBtn = document.getElementById('btn-pause-action'); - ui.stopActionBtn = document.getElementById('btn-stop-action'); - ui.failureDetailsBtn = document.getElementById('btn-failure-details'); - ui.confirmDialog = document.getElementById('confirm-dialog'); - ui.confirmMessage = document.getElementById('confirm-message'); - ui.confirmOk = document.getElementById('confirm-ok'); - ui.confirmCancel = document.getElementById('confirm-cancel'); - ui.failureDialog = document.getElementById('failure-dialog'); - ui.failureList = document.getElementById('failure-list'); - ui.failureClose = document.getElementById('failure-close'); - ui.failureRetry = document.getElementById('failure-retry'); } function ensureUI() { @@ -180,7 +212,8 @@ let isBatchDeleting = false; let isDeletePaused = false; let deleteTotal = 0; let deleteProcessed = 0; -const BATCH_SIZE = 10; +let currentBatchTaskId = null; +let batchEventSource = null; async function loadStats(options = {}) { try { @@ -210,70 +243,7 @@ async function loadStats(options = {}) { return; } const data = await res.json(); - if (!merge) { - accountStates.clear(); - } - - if (ui.imgCount) ui.imgCount.textContent = data.local_image.count; - if (ui.imgSize) ui.imgSize.textContent = `${data.local_image.size_mb} MB`; - if (ui.videoCount) ui.videoCount.textContent = data.local_video.count; - if (ui.videoSize) ui.videoSize.textContent = `${data.local_video.size_mb} MB`; - if (ui.onlineCount) ui.onlineCount.textContent = data.online.count; - - const statusEl = ui.onlineStatus; - const lastClearEl = ui.onlineLastClear; - if (data.online.status === 'ok') { - statusEl.textContent = '连接正常'; - statusEl.className = 'text-xs text-green-600 mt-1'; - } else if (data.online.status === 'no_token') { - statusEl.textContent = '无可用 Token'; - statusEl.className = 'text-xs text-orange-500 mt-1'; - } else if (data.online.status === 'not_loaded') { - statusEl.textContent = '未加载'; - statusEl.className = 'text-xs text-[var(--accents-4)] mt-1'; - } else { - statusEl.textContent = '无法连接'; - statusEl.className = 'text-xs text-red-500 mt-1'; - } - - // Update master accounts list - updateAccountSelect(data.online_accounts || []); - - // Update dynamic states - const details = Array.isArray(data.online_details) ? data.online_details : []; - details.forEach(detail => { - accountStates.set(detail.token, { - count: detail.count, - status: detail.status, - last_asset_clear_at: detail.last_asset_clear_at - }); - }); - if (data.online?.token) { - accountStates.set(data.online.token, { - count: data.online.count, - status: data.online.status, - last_asset_clear_at: data.online.last_asset_clear_at - }); - } - - if (data.online_scope === 'all') { - currentScope = 'all'; - currentToken = ''; - } else if (data.online_scope === 'selected') { - currentScope = 'selected'; - } else if (data.online.token) { - currentScope = 'single'; - currentToken = data.online.token; - } else { - currentScope = 'none'; - } - - if (lastClearEl) { - const timeText = formatTime(data.online.last_asset_clear_at); - lastClearEl.textContent = timeText ? `上次清空：${timeText}` : ''; - } - - renderAccountTable(data); + applyStatsData(data, merge); return data; } catch (e) { if (!silent) showToast('加载统计失败', 'error'); @@ -281,6 +251,59 @@ async function loadStats(options = {}) { } } +function applyStatsData(data, merge = false) { + if (!merge) { + accountStates.clear(); + } + + setText(ui.imgCount, data.local_image.count); + setText(ui.imgSize, `${data.local_image.size_mb} MB`); + setText(ui.videoCount, data.local_video.count); + setText(ui.videoSize, `${data.local_video.size_mb} MB`); + setText(ui.onlineCount, data.online.count); + + const online = data.online || {}; + const status = resolveOnlineStatus(online.status); + setOnlineStatus(status.text, status.className); + + // Update master accounts list + updateAccountSelect(data.online_accounts || []); + + // Update dynamic states + const details = Array.isArray(data.online_details) ? data.online_details : []; + details.forEach(detail => { + accountStates.set(detail.token, { + count: detail.count, + status: detail.status, + last_asset_clear_at: detail.last_asset_clear_at + }); + }); + if (online?.token) { + accountStates.set(online.token, { + count: online.count, + status: online.status, + last_asset_clear_at: online.last_asset_clear_at + }); + } + + if (data.online_scope === 'all') { + currentScope = 'all'; + currentToken = ''; + } else if (data.online_scope === 'selected') { + currentScope = 'selected'; + } else if (online.token) { + currentScope = 'single'; + currentToken = online.token; + } else { + currentScope = 'none'; + } + + const timeText = formatTime(online.last_asset_clear_at); + setText(ui.onlineLastClear, timeText ? `上次清空：${timeText}` : ''); + + renderAccountTable(data); +} + function updateAccountSelect(accounts) { accountMap.clear(); accounts.forEach(account => { @@ -341,60 +364,78 @@ function renderAccountTable(data) { } if (rows.length === 0) { - tbody.innerHTML = ''; + tbody.replaceChildren(); empty.classList.remove('hidden'); return; } empty.classList.add('hidden'); - tbody.innerHTML = rows.map(row => { - let statusClass = 'badge-gray'; - let statusText = '未加载'; + const selected = selectedTokens; + const fragment = document.createDocumentFragment(); + rows.forEach(row => { + const tr = document.createElement('tr'); + const isSelected = selected.has(row.token); + if (isSelected) tr.classList.add('row-selected'); - if (row.status === 'ok') { - statusClass = 'badge-green'; - statusText = '正常'; - } else if (row.status === 'not_loaded') { - statusClass = 'badge-gray'; - statusText = '未加载'; - } else { - statusClass = 'badge-red'; - statusText = '异常'; - } + const tdCheck = document.createElement('td'); + tdCheck.className = 'text-center'; + const checkbox = document.createElement('input'); + checkbox.type = 'checkbox'; + checkbox.className = 'checkbox'; + checkbox.checked = isSelected; + checkbox.setAttribute('data-token', row.token); + checkbox.addEventListener('change', () => toggleSelect(row.token, checkbox)); + tdCheck.appendChild(checkbox); - const lastClear = formatTime(row.last_asset_clear_at) || '-'; - // Use token as identifier, but we might want index for simpler toggling if we switch to array-based state like token.js - // For now, keep Set-based logic but update UI - const checked = selectedTokens.has(row.token) ? 'checked' : ''; + const tdToken = document.createElement('td'); + tdToken.className = 'text-left'; + const tokenWrap = document.createElement('div'); + tokenWrap.className = 'flex items-center gap-2'; + const tokenText = document.createElement('span'); + tokenText.className = 'font-mono text-xs text-gray-500'; + tokenText.title = row.token; + tokenText.textContent = row.token_masked || row.token; + tokenWrap.appendChild(tokenText); + tdToken.appendChild(tokenWrap); - // Shorten token for display if not already masked, though backend gives masked - // We'll use the masked version from backend + const tdPool = document.createElement('td'); + tdPool.className = 'text-center'; + const poolBadge = document.createElement('span'); + poolBadge.className = 'badge badge-gray'; + poolBadge.textContent = row.pool || '-'; + tdPool.appendChild(poolBadge); - const rowClass = selectedTokens.has(row.token) ? 'row-selected' : ''; - const countText = row.count === '-' ? '未加载' : row.count; - return ` - - - - - -
- ${row.token_masked} -
- - ${row.pool || '-'} - ${countText} - ${lastClear} - -
- -
- - - `; - }).join(''); + const tdCount = document.createElement('td'); + tdCount.className = 'text-center'; + const countBadge = document.createElement('span'); + countBadge.className = 'badge badge-gray'; + countBadge.textContent = row.count === '-' ? '未加载' : row.count; + tdCount.appendChild(countBadge); + + const tdLast = document.createElement('td'); + tdLast.className = 'text-left text-xs text-gray-500'; + tdLast.textContent = formatTime(row.last_asset_clear_at) || '-'; + + const tdActions = document.createElement('td'); + tdActions.className = 'text-center'; + const actionsWrap = document.createElement('div'); + actionsWrap.className = 'flex items-center justify-center gap-2'; + actionsWrap.appendChild(createIconButton( + '清空', + ``, + () => clearOnlineCache(row.token) + )); + tdActions.appendChild(actionsWrap); + + tr.appendChild(tdCheck); + tr.appendChild(tdToken); + tr.appendChild(tdPool); + tr.appendChild(tdCount); + tr.appendChild(tdLast); + tr.appendChild(tdActions); + fragment.appendChild(tr); + }); + tbody.replaceChildren(fragment); syncSelectAllState(); updateSelectedCount(); updateBatchActionsVisibility(); @@ -417,6 +458,18 @@ async function clearCache(type) { const data = await res.json(); if (data.status === 'success') { showToast(`清理成功，释放 ${data.result.size_mb} MB`, 'success'); + const state = cacheListState[type]; + if (state) { + state.items = []; + state.loaded = true; + } + if (selectedLocal[type]) selectedLocal[type].clear(); + if (state && state.visible) { + renderLocalCacheList(type, []); + } else { + syncLocalSelectAllState(type); + updateSelectedCount(); + } loadStats(); } else { showToast('清理失败', 'error'); @@ -611,15 +664,19 @@ function updateBatchProgress() { container.classList.remove('hidden'); if (ui.pauseActionBtn) { - const paused = isLoading ? isLoadPaused : isDeletePaused; - ui.pauseActionBtn.textContent = paused ? '继续' : '暂停'; - ui.pauseActionBtn.classList.remove('hidden'); + ui.pauseActionBtn.classList.add('hidden'); } if (ui.stopActionBtn) { ui.stopActionBtn.classList.remove('hidden'); } } +function refreshBatchUI() { + setActionButtonsState(); + updateBatchActionsVisibility(); + updateBatchProgress(); +} + function setOnlineStatus(text, className) { const statusEl = ui.onlineStatus; if (!statusEl) return; @@ -640,8 +697,6 @@ function updateToolbarForSection() { } function updateOnlineCountFromTokens(tokens) { - const el = ui.onlineCount; - if (!el) return; let total = 0; tokens.forEach(token => { const state = accountStates.get(token); @@ -649,7 +704,7 @@ function updateOnlineCountFromTokens(tokens) { total += state.count; } }); - el.textContent = String(total); + setText(ui.onlineCount, String(total)); } function formatSize(bytes) { @@ -741,48 +796,79 @@ function renderLocalCacheList(type, items) { const body = type === 'image' ? ui.localImageBody : ui.localVideoBody; if (!body) return; if (!items || items.length === 0) { - body.innerHTML = `暂无文件`; + body.innerHTML = `暂无文件`; syncLocalSelectAllState(type); return; } const selected = selectedLocal[type]; - body.innerHTML = items.map(item => { - const timeText = formatTime(item.mtime_ms); - const preview = item.preview_url ? `` : ''; - const checked = selected.has(item.name) ? 'checked' : ''; - const rowClass = selected.has(item.name) ? 'row-selected' : ''; - return ` - - - - - -
- ${preview} - ${item.name} -
- - ${formatSize(item.size_bytes)} - ${timeText} - -
- - -
- - + const fragment = document.createDocumentFragment(); + items.forEach(item => { + const tr = document.createElement('tr'); + const isSelected = selected.has(item.name); + if (isSelected) tr.classList.add('row-selected'); + + const tdCheck = document.createElement('td'); + tdCheck.className = 'text-center'; + const checkbox = document.createElement('input'); + checkbox.type = 'checkbox'; + checkbox.className = 'checkbox'; + checkbox.checked = isSelected; + checkbox.setAttribute('data-name', item.name); + checkbox.onchange = () => toggleLocalSelect(type, item.name, checkbox); + tdCheck.appendChild(checkbox); + + const tdName = document.createElement('td'); + tdName.className = 'text-left'; + const nameWrap = document.createElement('div'); + nameWrap.className = 'flex items-center gap-2'; + if (item.preview_url) { + const img = document.createElement('img'); + img.src = item.preview_url; + img.alt = ''; + img.className = 'cache-preview'; + nameWrap.appendChild(img); + } + const nameText = document.createElement('span'); + nameText.className = 'font-mono text-xs text-gray-500'; + nameText.textContent = item.name; + nameWrap.appendChild(nameText); + tdName.appendChild(nameWrap); + + const tdSize = document.createElement('td'); + tdSize.className = 'text-left'; + tdSize.textContent = formatSize(item.size_bytes); + + const tdTime = document.createElement('td'); + tdTime.className = 'text-left text-xs text-gray-500'; + tdTime.textContent = formatTime(item.mtime_ms); + + const tdActions = document.createElement('td'); + tdActions.className = 'text-center'; + tdActions.innerHTML = ` +
+ + +
`; - }).join(''); + + tr.appendChild(tdCheck); + tr.appendChild(tdName); + tr.appendChild(tdSize); + tr.appendChild(tdTime); + tr.appendChild(tdActions); + fragment.appendChild(tr); + }); + body.replaceChildren(fragment); syncLocalSelectAllState(type); updateSelectedCount(); } @@ -894,55 +980,47 @@ function handleDeleteClick() { } } -function stopBatchLoad() { +function stopBatchLoad(options = {}) { if (!isBatchLoading) return; isBatchLoading = false; isLoadPaused = false; currentBatchAction = null; batchQueue = []; + BatchSSE.close(batchEventSource); + batchEventSource = null; + currentBatchTaskId = null; setOnlineStatus('已终止', 'text-xs text-[var(--accents-4)] mt-1'); updateLoadButton(); - setActionButtonsState(); - updateBatchActionsVisibility(); - updateBatchProgress(); - showToast('已终止剩余加载请求', 'info'); + refreshBatchUI(); + if (!options.silent) showToast('已终止剩余加载请求', 'info'); } -function stopBatchDelete() { +function stopBatchDelete(options = {}) { if (!isBatchDeleting) return; isBatchDeleting = false; isDeletePaused = false; currentBatchAction = null; batchQueue = []; + BatchSSE.close(batchEventSource); + batchEventSource = null; + currentBatchTaskId = null; updateDeleteButton(); - setActionButtonsState(); - updateBatchActionsVisibility(); - updateBatchProgress(); - showToast('已终止剩余清理请求', 'info'); + refreshBatchUI(); + if (!options.silent) showToast('已终止剩余清理请求', 'info'); } function togglePause() { - if (isBatchLoading) { - isLoadPaused = !isLoadPaused; - if (isLoadPaused) { - setOnlineStatus('已暂停', 'text-xs text-[var(--accents-4)] mt-1'); - } else { - setOnlineStatus('加载中', 'text-xs text-blue-600 mt-1'); - processBatchQueue(); - } - } else if (isBatchDeleting) { - isDeletePaused = !isDeletePaused; - if (!isDeletePaused) { - processDeleteQueue(); - } + if (isBatchLoading || isBatchDeleting) { + showToast('当前批量任务不支持暂停', 'info'); } - updateBatchProgress(); } function stopActiveBatch() { if (isBatchLoading) { + BatchSSE.cancel(currentBatchTaskId, apiKey); stopBatchLoad(); } else if (isBatchDeleting) { + BatchSSE.cancel(currentBatchTaskId, apiKey); stopBatchDelete(); } } @@ -996,7 +1074,7 @@ function retryFailed() { } } -function startBatchLoad(tokens) { +async function startBatchLoad(tokens) { if (isBatchLoading) { showToast('正在加载中，请稍候', 'info'); return; @@ -1020,52 +1098,79 @@ function startBatchLoad(tokens) { updateOnlineCountFromTokens(batchTokens); setOnlineStatus('加载中', 'text-xs text-blue-600 mt-1'); updateLoadButton(); - setActionButtonsState(); if (accountMap.size > 0) { renderAccountTable({ online_accounts: Array.from(accountMap.values()), online_details: [], online: {} }); } - updateBatchActionsVisibility(); - updateBatchProgress(); + refreshBatchUI(); - processBatchQueue(); -} + try { + const res = await fetch('/api/v1/admin/cache/online/load/async', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + ...buildAuthHeaders(apiKey) + }, + body: JSON.stringify({ tokens }) + }); + const data = await res.json(); + if (!res.ok || data.status !== 'success') { + throw new Error(data.detail || '请求失败'); + } -async function processBatchQueue() { - if (!isBatchLoading || isLoadPaused) return; - if (batchQueue.length === 0) { - finishBatchLoad(); - return; - } - - const chunk = batchQueue.splice(0, BATCH_SIZE); - const data = await loadStats({ tokens: chunk, merge: true, silent: true }); - if (!data) { - chunk.forEach(token => loadFailed.set(token, '请求失败')); - } else { - const details = Array.isArray(data.online_details) ? data.online_details : []; - const detailMap = new Map(details.map(item => [item.token, item])); - chunk.forEach(token => { - const detail = detailMap.get(token); - if (!detail) { - loadFailed.set(token, '返回为空'); - return; - } - if (detail.status !== 'ok') { - loadFailed.set(token, detail.status); - } else { - loadFailed.delete(token); + currentBatchTaskId = data.task_id; + BatchSSE.close(batchEventSource); + batchEventSource = BatchSSE.open(currentBatchTaskId, apiKey, { + onMessage: (msg) => { + if (msg.type === 'snapshot' || msg.type === 'progress') { + if (typeof msg.total === 'number') batchTotal = msg.total; + if (typeof msg.processed === 'number') batchProcessed = msg.processed; + updateBatchProgress(); + } else if (msg.type === 'done') { + if (typeof msg.total === 'number') batchTotal = msg.total; + batchProcessed = batchTotal; + updateBatchProgress(); + const result = msg.result; + if (result) { + applyStatsData(result, true); + const details = Array.isArray(result.online_details) ? result.online_details : []; + loadFailed.clear(); + details.forEach(detail => { + if (detail.status !== 'ok') loadFailed.set(detail.token, detail.status); + }); + } + finishBatchLoad(); + if (msg.warning) { + showToast(`加载完成\n⚠️ ${msg.warning}`, 'warning'); + } + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; + } else if (msg.type === 'cancelled') { + stopBatchLoad({ silent: true }); + showToast('已终止加载', 'info'); + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; + } else if (msg.type === 'error') { + stopBatchLoad({ silent: true }); + showToast('加载失败: ' + (msg.error || '未知错误'), 'error'); + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; + } + }, + onError: () => { + stopBatchLoad({ silent: true }); + showToast('连接中断', 'error'); + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; } }); + } catch (e) { + stopBatchLoad({ silent: true }); + showToast(e.message || '请求失败', 'error'); } - batchProcessed += chunk.length; - updateOnlineCountFromTokens(batchTokens); - updateLoadButton(); - setOnlineStatus('加载中', 'text-xs text-blue-600 mt-1'); - updateBatchProgress(); - - setTimeout(() => { - processBatchQueue(); - }, 300); } function finishBatchLoad() { @@ -1085,9 +1190,7 @@ function finishBatchLoad() { setOnlineStatus('连接正常', 'text-xs text-green-600 mt-1'); } updateLoadButton(); - setActionButtonsState(); - updateBatchActionsVisibility(); - updateBatchProgress(); + refreshBatchUI(); } async function loadSelectedAccounts() { @@ -1125,7 +1228,7 @@ async function clearSelectedAccounts() { startBatchDelete(Array.from(selectedTokens)); } -function startBatchDelete(tokens) { +async function startBatchDelete(tokens) { if (!tokens || tokens.length === 0) return; isBatchDeleting = true; isDeletePaused = false; @@ -1137,55 +1240,9 @@ function startBatchDelete(tokens) { batchQueue = tokens.slice(); showToast('正在批量清理在线资产，请稍候...', 'info'); updateDeleteButton(); - setActionButtonsState(); - updateBatchActionsVisibility(); - updateBatchProgress(); - processDeleteQueue(); -} - -async function processDeleteQueue() { - if (!isBatchDeleting || isDeletePaused) return; - if (batchQueue.length === 0) { - finishBatchDelete(); - return; - } - const chunk = batchQueue.splice(0, BATCH_SIZE); - const results = await clearOnlineCacheBatch(chunk); - if (results && results.status === 'success' && results.results) { - Object.entries(results.results).forEach(([token, result]) => { - if (result.status !== 'success') { - deleteFailed.set(token, result.error || '清理失败'); - } else { - deleteFailed.delete(token); - } - }); - } else { - chunk.forEach(token => deleteFailed.set(token, '请求失败')); - } - deleteProcessed += chunk.length; - updateDeleteButton(); - updateBatchProgress(); - setTimeout(() => { - processDeleteQueue(); - }, 300); -} - -function finishBatchDelete() { - isBatchDeleting = false; - isDeletePaused = false; - currentBatchAction = null; - updateDeleteButton(); - setActionButtonsState(); - updateBatchActionsVisibility(); - updateBatchProgress(); - showToast('批量清理完成', 'success'); - loadStats(); -} - -async function clearOnlineCacheBatch(tokens = []) { - if (!tokens || tokens.length === 0) return; + refreshBatchUI(); try { - const res = await fetch('/api/v1/admin/cache/online/clear', { + const res = await fetch('/api/v1/admin/cache/online/clear/async', { method: 'POST', headers: { 'Content-Type': 'application/json', @@ -1195,15 +1252,75 @@ async function clearOnlineCacheBatch(tokens = []) { }); const data = await res.json(); if (!res.ok || data.status !== 'success') { - showToast('批量清理失败', 'error'); + throw new Error(data.detail || '请求失败'); } - return data; + + currentBatchTaskId = data.task_id; + BatchSSE.close(batchEventSource); + batchEventSource = BatchSSE.open(currentBatchTaskId, apiKey, { + onMessage: (msg) => { + if (msg.type === 'snapshot' || msg.type === 'progress') { + if (typeof msg.total === 'number') deleteTotal = msg.total; + if (typeof msg.processed === 'number') deleteProcessed = msg.processed; + updateBatchProgress(); + } else if (msg.type === 'done') { + if (typeof msg.total === 'number') deleteTotal = msg.total; + deleteProcessed = deleteTotal; + updateBatchProgress(); + const result = msg.result; + deleteFailed.clear(); + if (result && result.results) { + Object.entries(result.results).forEach(([token, res]) => { + if (res.status !== 'success') { + deleteFailed.set(token, res.error || '清理失败'); + } + }); + } + finishBatchDelete(); + if (msg.warning) { + showToast(`清理完成\n⚠️ ${msg.warning}`, 'warning'); + } + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; + } else if (msg.type === 'cancelled') { + stopBatchDelete({ silent: true }); + showToast('已终止清理', 'info'); + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; + } else if (msg.type === 'error') { + stopBatchDelete({ silent: true }); + showToast('清理失败: ' + (msg.error || '未知错误'), 'error'); + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; + } + }, + onError: () => { + stopBatchDelete({ silent: true }); + showToast('连接中断', 'error'); + currentBatchTaskId = null; + BatchSSE.close(batchEventSource); + batchEventSource = null; + } + }); } catch (e) { - showToast('请求超时或失败', 'error'); - return null; + stopBatchDelete({ silent: true }); + showToast(e.message || '请求失败', 'error'); } } +function finishBatchDelete() { + isBatchDeleting = false; + isDeletePaused = false; + currentBatchAction = null; + updateDeleteButton(); + refreshBatchUI(); + showToast('批量清理完成', 'success'); + loadStats(); +} + async function clearOnlineCache(targetToken = '', skipConfirm = false) { const tokenToClear = targetToken || (currentScope === 'all' ? '' : currentToken); if (!tokenToClear) { diff --git a/app/static/common/batch-sse.js b/app/static/common/batch-sse.js new file mode 100644 index 0000000..036a6a9 --- /dev/null +++ b/app/static/common/batch-sse.js @@ -0,0 +1,46 @@ +(function (global) { + function openBatchStream(taskId, apiKey, handlers = {}) { + if (!taskId) return null; + const url = `/api/v1/admin/batch/${taskId}/stream?api_key=${encodeURIComponent(apiKey || '')}`; + const es = new EventSource(url); + + es.onmessage = (e) => { + if (!e.data) return; + let msg; + try { + msg = JSON.parse(e.data); + } catch { + return; + } + if (handlers.onMessage) handlers.onMessage(msg); + }; + + es.onerror = () => { + if (handlers.onError) handlers.onError(); + }; + + return es; + } + + function closeBatchStream(es) { + if (es) es.close(); + } + + async function cancelBatchTask(taskId, apiKey) { + if (!taskId) return; + try { + await fetch(`/api/v1/admin/batch/${taskId}/cancel`, { + method: 'POST', + headers: apiKey ? { Authorization: `Bearer ${apiKey}` } : undefined + }); + } catch { + // ignore + } + } + + global.BatchSSE = { + open: openBatchStream, + close: closeBatchStream, + cancel: cancelBatchTask + }; +})(window); diff --git a/app/static/common/common.css b/app/static/common/common.css index a9a11b9..099a4bc 100644 --- a/app/static/common/common.css +++ b/app/static/common/common.css @@ -30,6 +30,18 @@ body { font-family: 'Geist Mono', monospace; } +.table-empty { + color: var(--accents-4); + font-size: 12px; + font-weight: 400; + padding: 48px 0; + text-align: center; +} + +td.table-empty { + height: auto; +} + .fade-in { animation: fadeIn 0.5s ease-out forwards; } diff --git a/app/static/common/footer.html b/app/static/common/footer.html index 5eabca4..1026540 100644 --- a/app/static/common/footer.html +++ b/app/static/common/footer.html @@ -1,5 +1,5 @@