From 03ada7eb010a63b25946d813ed31d71f52ee94a5 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 19 Aug 2026 09:11:42 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=94=B6=E6=95=9B=20factory=20?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=E7=BA=A7=E5=89=AF=E4=BD=9C=E7=94=A8=E5=B9=B6?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=20async=20=E8=B7=AF=E5=BE=84=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6=E4=B8=BA=E5=BC=82=E6=AD=A5=E5=90=8E=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将 configure_token_codec/configure_plugin_routes 移入 create_app()，消除 import 期副作用 - runtime 层新增 AsyncCacheProxy/AsyncTTLCache/AsyncProgressHelper，共享 progress region - search/system/history/scheduler/dashboard 的事件循环路径切换异步进度后端 - 同步进度回调经事件循环提交或线程池执行，避免阻塞事件循环 --- app/api/endpoints/dashboard.py | 13 +- app/api/endpoints/history.py | 56 ++++--- app/api/endpoints/system.py | 6 +- app/chain/search.py | 75 +++++----- app/factory.py | 45 +++--- app/runtime/cache.py | 137 ++++++++++++++++++ app/runtime/progress.py | 117 ++++++++++++++- app/scheduler.py | 96 ++++++++++-- .../runtime-contract-baseline.json | 6 +- 9 files changed, 446 insertions(+), 105 deletions(-) diff --git a/app/api/endpoints/dashboard.py b/app/api/endpoints/dashboard.py index 20d9b18cd..8d2c15006 100644 --- a/app/api/endpoints/dashboard.py +++ b/app/api/endpoints/dashboard.py @@ -2,6 +2,7 @@ from pathlib import Path from typing import Any, List, Optional, Annotated from fastapi import Depends +from fastapi.concurrency import run_in_threadpool from app.schemas.dashboard import DashboardMemoryInfo as _SchemaDashboardMemoryInfo from app.schemas.dashboard import DashboardSystemInfo as _SchemaDashboardSystemInfo @@ -160,7 +161,8 @@ async def schedule(_: Any = Depends(get_current_active_superuser)) -> Any: """ 查询后台服务信息 """ - return Scheduler().list() + # 同步 list() 内含同步进度读取，放到线程池执行避免阻塞事件循环 + return await run_in_threadpool(Scheduler().list) @router.get( @@ -174,7 +176,8 @@ async def schedule_progress( """ 查询指定后台服务的执行进度。 """ - progress = Scheduler().get_progress(job_id) + # 异步进度后端读取，避免同步 Redis 调用阻塞事件循环 + progress = await Scheduler().aget_progress(job_id) if not progress: return _SchemaResponse(success=False, message="后台服务不存在") return _SchemaResponse(success=True, data=progress.model_dump()) @@ -189,7 +192,8 @@ async def schedule2(_: Annotated[str, Depends(verify_apitoken)]) -> Any: """ 查询下载器信息 API_TOKEN认证（?token=xxx） """ - return Scheduler().list() + # 同步 list() 内含同步进度读取，放到线程池执行避免阻塞事件循环 + return await run_in_threadpool(Scheduler().list) @router.get( @@ -203,7 +207,8 @@ async def schedule_progress2( """ 查询指定后台服务的执行进度 API_TOKEN认证（?token=xxx） """ - progress = Scheduler().get_progress(job_id) + # 异步进度后端读取，避免同步 Redis 调用阻塞事件循环 + progress = await Scheduler().aget_progress(job_id) if not progress: return _SchemaResponse(success=False, message="后台服务不存在") return _SchemaResponse(success=True, data=progress.model_dump()) diff --git a/app/api/endpoints/history.py b/app/api/endpoints/history.py index e6cada697..8753c3cbf 100644 --- a/app/api/endpoints/history.py +++ b/app/api/endpoints/history.py @@ -28,7 +28,7 @@ from app.api.deps import ( get_history_query_service, get_transfer_history_mutation_command, ) -from app.runtime.progress import ProgressHelper +from app.runtime.progress import AsyncProgressHelper from app.application.history import ( DownloadHistoryMutationCommand, HistoryQueryService, @@ -49,19 +49,24 @@ def normalize_history_ids(history_ids: list[int]) -> list[int]: def _start_ai_redo_task(history_id: int, prompt: str, progress_key: str): - """在后台线程中启动单条 AI 重新整理任务，并通过 ProgressHelper 实时更新进度。""" - progress = ProgressHelper(progress_key) - progress.start() - progress.update( - text=f"智能助手正在准备整理记录 #{history_id} ...", - data={"history_id": history_id, "success": True}, - ) + """在后台任务中启动单条 AI 重新整理任务，并通过异步进度辅助类实时更新进度。""" + progress = AsyncProgressHelper(progress_key) def update_output(text: str): - progress.update(text=text, data={"history_id": history_id}) + # 输出回调由 agent 在事件循环上同步调用，不能直接 await； + # 提交到全局事件循环非阻塞执行，避免同步缓存后端阻塞事件循环。 + asyncio.run_coroutine_threadsafe( + progress.update(text=text, data={"history_id": history_id}), + global_vars.loop, + ) async def runner(): try: + await progress.start() + await progress.update( + text=f"智能助手正在准备整理记录 #{history_id} ...", + data={"history_id": history_id, "success": True}, + ) manager = get_running_agent_manager() if manager is None: logger.warning("智能助手服务未运行，跳过单条整理历史 AI 重做") @@ -73,12 +78,12 @@ def _start_ai_redo_task(history_id: int, prompt: str, progress_key: str): reply_mode=ReplyMode.CAPTURE_ONLY, allow_message_tools=False, ) - progress.update( + await progress.update( text="智能助手整理完成", data={"history_id": history_id, "success": True, "completed": True}, ) except Exception as e: - progress.update( + await progress.update( text=f"智能助手整理失败：{str(e)}", data={ "history_id": history_id, @@ -88,7 +93,7 @@ def _start_ai_redo_task(history_id: int, prompt: str, progress_key: str): }, ) finally: - progress.end() + await progress.end() asyncio.run_coroutine_threadsafe(runner(), global_vars.loop) @@ -98,19 +103,24 @@ def _start_batch_ai_redo_task( prompt: str, progress_key: str, ): - """在后台线程中启动批量 AI 重新整理任务，并通过 ProgressHelper 实时更新进度。""" - progress = ProgressHelper(progress_key) - progress.start() - progress.update( - text=f"智能助手正在准备批量整理 {len(history_ids)} 条记录 ...", - data={"history_ids": history_ids, "success": True}, - ) + """在后台任务中启动批量 AI 重新整理任务，并通过异步进度辅助类实时更新进度。""" + progress = AsyncProgressHelper(progress_key) def update_output(text: str): - progress.update(text=text, data={"history_ids": history_ids}) + # 输出回调由 agent 在事件循环上同步调用，不能直接 await； + # 提交到全局事件循环非阻塞执行，避免同步缓存后端阻塞事件循环。 + asyncio.run_coroutine_threadsafe( + progress.update(text=text, data={"history_ids": history_ids}), + global_vars.loop, + ) async def runner(): try: + await progress.start() + await progress.update( + text=f"智能助手正在准备批量整理 {len(history_ids)} 条记录 ...", + data={"history_ids": history_ids, "success": True}, + ) manager = get_running_agent_manager() if manager is None: logger.warning("智能助手服务未运行，跳过批量整理历史 AI 重做") @@ -122,12 +132,12 @@ def _start_batch_ai_redo_task( reply_mode=ReplyMode.CAPTURE_ONLY, allow_message_tools=False, ) - progress.update( + await progress.update( text="智能助手批量整理完成", data={"history_ids": history_ids, "success": True, "completed": True}, ) except Exception as e: - progress.update( + await progress.update( text=f"智能助手批量整理失败：{str(e)}", data={ "history_ids": history_ids, @@ -137,7 +147,7 @@ def _start_batch_ai_redo_task( }, ) finally: - progress.end() + await progress.end() asyncio.run_coroutine_threadsafe(runner(), global_vars.loop) diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index 80be3ec20..c940c4175 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -52,7 +52,7 @@ from app.adapters.external.market import ( split_plugin_market_repo_urls, ) from app.application.messaging.message import MessageHelper -from app.runtime.progress import ProgressHelper +from app.runtime.progress import AsyncProgressHelper from app.application.rules import RuleHelper from app.adapters.external.server import MoviePilotServerHelper from app.runtime.state import SystemHelper @@ -847,7 +847,7 @@ async def get_progress( """ 实时获取处理进度，返回格式为SSE """ - progress = ProgressHelper(process_type) + progress = AsyncProgressHelper(process_type) locale = LocaleHelper.get_current_locale() async def event_generator(): @@ -855,7 +855,7 @@ async def get_progress( while not global_vars.is_system_stopped: if await request.is_disconnected(): break - detail = progress.get(locale=locale) + detail = await progress.get(locale=locale) yield f"data: {json.dumps(detail)}\n\n" await asyncio.sleep(0.5) except asyncio.CancelledError: diff --git a/app/chain/search.py b/app/chain/search.py index 019b903a5..00660eb44 100644 --- a/app/chain/search.py +++ b/app/chain/search.py @@ -22,7 +22,7 @@ from app.domain.meta.metamusic import MetaMusic from app.domain.metainfo import MetaInfo from app.domain.context import MusicInfo from app.application.configuration import get_configured_system_config -from app.runtime.progress import ProgressHelper +from app.runtime.progress import AsyncProgressHelper, ProgressHelper from app.application.site.sites import SitesHelper # pylint: disable=no-name-in-module from app.application.search.state import ( SearchStateService, @@ -2396,9 +2396,9 @@ class SearchChain(ChainBase): logger.warn('未开启任何有效站点，无法搜索资源') return [] - # 开始进度 - progress = ProgressHelper(ProgressKey.Search) - progress.start() + # 开始进度（异步后端，避免同步 Redis 在事件循环上阻塞） + progress = AsyncProgressHelper(ProgressKey.Search) + await progress.start() # 开始计时 start_time = datetime.now() search_pages = self._build_search_pages(page) @@ -2407,8 +2407,8 @@ class SearchChain(ChainBase): # 完成数 finish_count = 0 # 更新进度 - progress.update(value=0, - text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") + await progress.update(value=0, + text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") # 结果集 results = [] semaphore = asyncio.Semaphore(settings.CONF.threadpool or total_num) @@ -2470,8 +2470,8 @@ class SearchChain(ChainBase): f"{site.get('name')} 第 {search_page} 页返回 {len(result or [])} 条，停止继续翻页" ) logger.info(f"站点搜索进度：{finish_count} / {total_num}") - progress.update(value=finish_count / total_num * 100, - text=f"正在搜索{keyword or ''}，已完成 {finish_count} / {total_num} 个请求 ...") + await progress.update(value=finish_count / total_num * 100, + text=f"正在搜索{keyword or ''}，已完成 {finish_count} / {total_num} 个请求 ...") finally: for task in pending_tasks: if not task.done(): @@ -2482,11 +2482,11 @@ class SearchChain(ChainBase): # 计算耗时 end_time = datetime.now() # 更新进度 - progress.update(value=100, - text=f"站点搜索完成，有效资源数：{len(results)}，总耗时 {(end_time - start_time).seconds} 秒") + await progress.update(value=100, + text=f"站点搜索完成，有效资源数：{len(results)}，总耗时 {(end_time - start_time).seconds} 秒") logger.info(f"站点搜索完成，有效资源数：{len(results)}，总耗时 {(end_time - start_time).seconds} 秒") # 结束进度 - progress.end() + await progress.end() # 返回 return results @@ -2527,14 +2527,15 @@ class SearchChain(ChainBase): } return - progress = ProgressHelper(ProgressKey.Search) - progress.start() + # 开始进度（异步后端，避免同步 Redis 在事件循环上阻塞） + progress = AsyncProgressHelper(ProgressKey.Search) + await progress.start() start_time = datetime.now() search_pages = self._build_search_pages(page) total_num = len(indexer_sites) * len(search_pages) finish_count = 0 - progress.update(value=0, - text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") + await progress.update(value=0, + text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") yield { "type": "progress", "stage": "searching", @@ -2606,7 +2607,7 @@ class SearchChain(ChainBase): logger.info(f"站点搜索进度：{finish_count} / {total_num}") progress_value = finish_count / total_num * 100 progress_text = f"正在搜索{keyword or ''}，已完成 {finish_count} / {total_num} 个请求 ..." - progress.update(value=progress_value, text=progress_text) + await progress.update(value=progress_value, text=progress_text) yield { "type": "append", "stage": "searching", @@ -2628,10 +2629,10 @@ class SearchChain(ChainBase): await asyncio.gather(*tasks.keys(), return_exceptions=True) end_time = datetime.now() - progress.update(value=100, - text=f"站点搜索完成，有效资源数：{results_count}，总耗时 {(end_time - start_time).seconds} 秒") + await progress.update(value=100, + text=f"站点搜索完成，有效资源数：{results_count}，总耗时 {(end_time - start_time).seconds} 秒") logger.info(f"站点搜索完成，有效资源数：{results_count}，总耗时 {(end_time - start_time).seconds} 秒") - progress.end() + await progress.end() async def __async_search_subtitles_all_sites(self, keyword: str, sites: List[int] = None, @@ -2657,14 +2658,15 @@ class SearchChain(ChainBase): logger.warn('未开启任何支持字幕搜索的有效站点，无法搜索字幕') return [] - progress = ProgressHelper(ProgressKey.Search) - progress.start() + # 开始进度（异步后端，避免同步 Redis 在事件循环上阻塞） + progress = AsyncProgressHelper(ProgressKey.Search) + await progress.start() start_time = datetime.now() search_pages = self._build_search_pages(page) total_num = len(indexer_sites) * len(search_pages) finish_count = 0 - progress.update(value=0, - text=f"开始搜索字幕，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") + await progress.update(value=0, + text=f"开始搜索字幕，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") results = [] semaphore = asyncio.Semaphore(settings.CONF.threadpool or total_num) @@ -2714,8 +2716,8 @@ class SearchChain(ChainBase): f"{site.get('name')} 字幕第 {search_page} 页返回 {len(result or [])} 条，停止继续翻页" ) logger.info(f"站点字幕搜索进度：{finish_count} / {total_num}") - progress.update(value=finish_count / total_num * 100, - text=f"正在搜索字幕{keyword or ''}，已完成 {finish_count} / {total_num} 个请求 ...") + await progress.update(value=finish_count / total_num * 100, + text=f"正在搜索字幕{keyword or ''}，已完成 {finish_count} / {total_num} 个请求 ...") finally: for task in pending_tasks: if not task.done(): @@ -2724,10 +2726,10 @@ class SearchChain(ChainBase): await asyncio.gather(*pending_tasks.keys(), return_exceptions=True) end_time = datetime.now() - progress.update(value=100, - text=f"站点字幕搜索完成，有效字幕数：{len(results)}，总耗时 {(end_time - start_time).seconds} 秒") + await progress.update(value=100, + text=f"站点字幕搜索完成，有效字幕数：{len(results)}，总耗时 {(end_time - start_time).seconds} 秒") logger.info(f"站点字幕搜索完成，有效字幕数：{len(results)}，总耗时 {(end_time - start_time).seconds} 秒") - progress.end() + await progress.end() return results async def __async_search_subtitles_all_sites_stream(self, keyword: str, @@ -2762,14 +2764,15 @@ class SearchChain(ChainBase): } return - progress = ProgressHelper(ProgressKey.Search) - progress.start() + # 开始进度（异步后端，避免同步 Redis 在事件循环上阻塞） + progress = AsyncProgressHelper(ProgressKey.Search) + await progress.start() start_time = datetime.now() search_pages = self._build_search_pages(page) total_num = len(indexer_sites) * len(search_pages) finish_count = 0 - progress.update(value=0, - text=f"开始搜索字幕，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") + await progress.update(value=0, + text=f"开始搜索字幕，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") yield { "type": "progress", "stage": "searching", @@ -2831,7 +2834,7 @@ class SearchChain(ChainBase): logger.info(f"站点字幕搜索进度：{finish_count} / {total_num}") progress_value = finish_count / total_num * 100 progress_text = f"正在搜索字幕{keyword or ''}，已完成 {finish_count} / {total_num} 个请求 ..." - progress.update(value=progress_value, text=progress_text) + await progress.update(value=progress_value, text=progress_text) yield { "type": "append", "stage": "searching", @@ -2853,10 +2856,10 @@ class SearchChain(ChainBase): await asyncio.gather(*tasks.keys(), return_exceptions=True) end_time = datetime.now() - progress.update(value=100, - text=f"站点字幕搜索完成，有效字幕数：{results_count}，总耗时 {(end_time - start_time).seconds} 秒") + await progress.update(value=100, + text=f"站点字幕搜索完成，有效字幕数：{results_count}，总耗时 {(end_time - start_time).seconds} 秒") logger.info(f"站点字幕搜索完成，有效字幕数：{results_count}，总耗时 {(end_time - start_time).seconds} 秒") - progress.end() + await progress.end() @eventmanager.register(EventType.SiteDeleted) def remove_site(self, event: Event): diff --git a/app/factory.py b/app/factory.py index 44ae3cafe..d634311f2 100644 --- a/app/factory.py +++ b/app/factory.py @@ -314,6 +314,7 @@ def create_app() -> FastAPI: allow_methods=["*"], allow_headers=["*"], ) + @_app.middleware("http") async def locale_context_middleware( request: Request, @@ -330,29 +331,29 @@ def create_app() -> FastAPI: finally: LocaleHelper.reset_current_locale(token) + # HTTP 适配器只持有令牌编解码端口，具体实现由组合根在创建应用时连接。 + configure_token_codec(create_access_token, decode_access_token) + + # 向 application 层插件路由服务注入应用实例，插件 API 的动态注册/移除 + # 统一经服务完成，避免 api.endpoints 反向依赖本模块。 + configure_plugin_routes(FastAPIDynamicRouteRegistry( + app=_app, + plugin_ids=lambda: PluginManager().get_running_plugin_ids(), + plugin_apis=lambda plugin_id: PluginManager().get_plugin_apis(plugin_id), + verify_token=verify_token, + verify_apikey=verify_apikey, + prefix=f"{settings.API_V1_STR}/plugin", + protected_routes={ + f"{settings.API_V1_STR}/openapi.json", + "/docs", + "/docs/oauth2-redirect", + "/redoc", + }, + log=logger, + )) + return _app -# HTTP 适配器只持有令牌编解码端口，具体实现由组合根连接。 -configure_token_codec(create_access_token, decode_access_token) - -# 创建 FastAPI 应用实例 +# 创建 FastAPI 应用实例；所有组合根装配副作用都在 create_app() 内部完成 app = create_app() - -# 向 application 层插件路由服务注入应用实例，插件 API 的动态注册/移除 -# 统一经服务完成，避免 api.endpoints 反向依赖本模块。 -configure_plugin_routes(FastAPIDynamicRouteRegistry( - app=app, - plugin_ids=lambda: PluginManager().get_running_plugin_ids(), - plugin_apis=lambda plugin_id: PluginManager().get_plugin_apis(plugin_id), - verify_token=verify_token, - verify_apikey=verify_apikey, - prefix=f"{settings.API_V1_STR}/plugin", - protected_routes={ - f"{settings.API_V1_STR}/openapi.json", - "/docs", - "/docs/oauth2-redirect", - "/redoc", - }, - log=logger, -)) diff --git a/app/runtime/cache.py b/app/runtime/cache.py index c44dbd1a0..b4fee9163 100644 --- a/app/runtime/cache.py +++ b/app/runtime/cache.py @@ -1158,6 +1158,143 @@ class CacheProxy: self._cache_backend.close() +class AsyncCacheProxy: + """ + 异步缓存代理类，将异步缓存后端的方法直接代理到实例上 + + 与同步 CacheProxy 的唯一差异是方法均为 async，默认绑定构造时指定的 region。 + """ + + def __init__(self, cache_backend: AsyncCacheBackend, region: str): + """ + 初始化异步缓存代理 + + :param cache_backend: 异步缓存后端实例 + :param region: 缓存区域 + """ + self._cache_backend = cache_backend + self._region = region + + def is_redis(self) -> bool: + """ + 检查当前缓存后端是否为 Redis（纯状态判断，无需 await） + """ + return self._cache_backend.is_redis() + + async def get(self, key: str, **kwargs) -> Any: + """ + 获取缓存值 + """ + kwargs.setdefault('region', self._region) + return await self._cache_backend.get(key, **kwargs) + + async def set(self, key: str, value: Any, **kwargs) -> None: + """ + 设置缓存值 + """ + kwargs.setdefault('region', self._region) + await self._cache_backend.set(key, value, **kwargs) + + async def delete(self, key: str, **kwargs) -> None: + """ + 删除缓存值 + """ + kwargs.setdefault('region', self._region) + await self._cache_backend.delete(key, **kwargs) + + async def exists(self, key: str, **kwargs) -> bool: + """ + 检查缓存键是否存在 + """ + kwargs.setdefault('region', self._region) + return await self._cache_backend.exists(key, **kwargs) + + async def clear(self, **kwargs) -> None: + """ + 清除缓存 + """ + kwargs.setdefault('region', self._region) + await self._cache_backend.clear(**kwargs) + + async def items(self, **kwargs): + """ + 获取所有缓存项 + """ + kwargs.setdefault('region', self._region) + async for item in self._cache_backend.items(**kwargs): + yield item + + async def keys(self, **kwargs): + """ + 获取所有缓存键 + """ + kwargs.setdefault('region', self._region) + async for key in self._cache_backend.keys(**kwargs): + yield key + + async def values(self, **kwargs): + """ + 获取所有缓存值 + """ + kwargs.setdefault('region', self._region) + async for value in self._cache_backend.values(**kwargs): + yield value + + async def update(self, other: Dict[str, Any], **kwargs) -> None: + """ + 更新缓存 + """ + kwargs.setdefault('region', self._region) + await self._cache_backend.update(other, **kwargs) + + async def pop(self, key: str, default: Any = None, **kwargs) -> Any: + """ + 弹出缓存项 + """ + kwargs.setdefault('region', self._region) + return await self._cache_backend.pop(key, default, **kwargs) + + async def popitem(self, **kwargs) -> Tuple[str, Any]: + """ + 弹出最后一个缓存项 + """ + kwargs.setdefault('region', self._region) + return await self._cache_backend.popitem(**kwargs) + + async def setdefault(self, key: str, default: Any = None, **kwargs) -> Any: + """ + 设置默认值 + """ + kwargs.setdefault('region', self._region) + return await self._cache_backend.setdefault(key, default, **kwargs) + + async def close(self) -> None: + """ + 关闭缓存连接 + """ + await self._cache_backend.close() + + +class AsyncTTLCache(AsyncCacheProxy): + """ + 基于 TTL 的异步缓存类，与同步 TTLCache 使用同一 region 语义， + 内存后端共享进程内存储，Redis 后端共享同一键空间 + """ + + def __init__(self, + region: Optional[str] = DEFAULT_CACHE_REGION, + maxsize: Optional[int] = DEFAULT_CACHE_SIZE, + ttl: Optional[int] = DEFAULT_CACHE_TTL): + """ + 初始化异步 TTL 缓存 + + :param maxsize: 缓存的最大条目数 + :param ttl: 缓存的存活时间，单位秒 + :param region: 缓存的区，为 None 时使用默认区 + """ + super().__init__(AsyncCache(cache_type='ttl', maxsize=maxsize, ttl=ttl), region) + + class TTLCache(CacheProxy): """ 基于 TTL 的缓存类，兼容 cachetools.TTLCache 接口 diff --git a/app/runtime/progress.py b/app/runtime/progress.py index 7876ea6b1..b256c4a32 100644 --- a/app/runtime/progress.py +++ b/app/runtime/progress.py @@ -1,7 +1,7 @@ from enum import Enum from typing import Optional, Union -from app.runtime.cache import TTLCache +from app.runtime.cache import AsyncTTLCache, TTLCache from app.runtime.localization import LocaleHelper from app.schemas.types import ProgressKey @@ -115,3 +115,118 @@ class ProgressHelper: ) detail["data"] = localized_data return detail + + +class AsyncProgressHelper: + """ + 处理进度辅助类（异步） + + 与 ProgressHelper 共用同一个进度 region：内存后端共享进程内存储， + Redis 后端共享同一键空间，因此同步写入、异步读取（或反之）均互通。 + 供事件循环上的异步调用方使用，避免同步缓存后端阻塞事件循环。 + """ + + def __init__(self, key: Union[ProgressKey, str]) -> None: + """为指定业务键绑定独立的异步进度缓存区域。""" + if isinstance(key, Enum): + key = key.value + self._key = key + self._progress = AsyncTTLCache(region="progress", maxsize=1024, ttl=24 * 60 * 60) + + async def __reset(self) -> None: + """ + 重置进度 + """ + await self._progress.set(self._key, { + "enable": False, + "value": 0, + "text": "请稍候...", + "data": {} + }) + + async def start(self) -> None: + """ + 开始进度 + """ + await self.__reset() + current = await self._progress.get(self._key) + if not current: + return + current['enable'] = True + await self._progress.set(self._key, current) + + async def end( + self, + text: Optional[str] = "", + data: Optional[dict] = None, + value: Optional[Union[float, int]] = 100, + ) -> None: + """ + 结束进度 + """ + current = await self._progress.get(self._key) + if not current: + return + if data is not None: + if not current.get('data'): + current['data'] = {} + current['data'].update(data) + current["enable"] = False + if value is not None: + current["value"] = max(min(float(value), 100), 0) + current["text"] = text or "" + await self._progress.set(self._key, current) + + async def update( + self, + value: Optional[Union[float, int]] = None, + text: Optional[str] = None, + data: Optional[dict] = None, + ) -> None: + """ + 更新进度 + """ + current = await self._progress.get(self._key) + if not current or not current.get('enable'): + return + if value is not None: + current['value'] = max(min(float(value), 100), 0) + if text is not None: + current['text'] = text + if data is not None: + if not current.get('data'): + current['data'] = {} + current['data'].update(data) + await self._progress.set(self._key, current) + + async def get(self, locale: Optional[str] = None) -> Optional[dict]: + """ + 获取当前进度，并按语言补充前端展示字段。 + + :param locale: 目标语言，未传入时使用当前请求上下文语言 + :return: 当前进度字典 + """ + current = await self._progress.get(self._key) + if not current: + return current + + detail = current.copy() + text = detail.get("text") + if isinstance(text, str): + detail["text_i18n"] = LocaleHelper.translate_text(text, locale=locale) + + data = detail.get("data") + if isinstance(data, dict): + localized_data = data.copy() + error = localized_data.get("error") + message = localized_data.get("message") + if isinstance(error, str): + localized_data["error_i18n"] = LocaleHelper.translate_text( + error, locale=locale + ) + if isinstance(message, str): + localized_data["message_i18n"] = LocaleHelper.translate_text( + message, locale=locale + ) + detail["data"] = localized_data + return detail diff --git a/app/scheduler.py b/app/scheduler.py index 9b3b9d718..1c22dee31 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -31,7 +31,7 @@ from app.db.oper.systemconfig import SystemConfigOper from app.application.maintenance import build_cleanup_service from app.application.image import WallpaperHelper from app.application.messaging.message import MessageHelper -from app.runtime.progress import ProgressHelper +from app.runtime.progress import AsyncProgressHelper, ProgressHelper from app.adapters.external.server import MoviePilotServerHelper from app.runtime.extensions.service_config import ServiceConfigHelper from app.application.site.sites import SitesHelper # pylint: disable=no-name-in-module @@ -609,7 +609,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): ) return job - def __finish_job( + async def __finish_job( self, job_id: str, success: bool = True, @@ -627,10 +627,11 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job["last_finished_at"] = finished_at job["last_error"] = error job_name = job.get("name") if job else job_id - progress = ProgressHelper(self._get_progress_key(job_id)) - current_progress = progress.get() or {} + # 收尾可能发生在事件循环上（__run_coro_job），使用异步进度后端避免阻塞 + progress = AsyncProgressHelper(self._get_progress_key(job_id)) + current_progress = await progress.get() or {} progress_value = 100 if success else current_progress.get("value", 0) - progress.end( + await progress.end( text=f"{job_name} {'执行完成' if success else '执行失败'}", data={ "id": job_id, @@ -682,6 +683,45 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): data=data, ) + async def aget_progress(self, job_id: str) -> Optional[_SchemaScheduleProgress]: + """ + 查询指定定时服务的执行进度（异步版本，供事件循环上的端点使用）。 + """ + if not job_id: + return None + with self._lock: + job = self._jobs.get(job_id) + job_name = job.get("name") if job else job_id + provider_name = job.get("provider_name", "[系统]") if job else None + running = bool(job.get("running")) if job else False + last_started_at = job.get("last_started_at") if job else None + last_finished_at = job.get("last_finished_at") if job else None + last_error = job.get("last_error") if job else None + # 异步后端读取，避免在事件循环上阻塞 + detail = await AsyncProgressHelper(self._get_progress_key(job_id)).get() or {} + if not job and not detail: + return None + data = detail.get("data") or {} + value = detail.get("value", 0) + try: + value = float(value) + except (TypeError, ValueError): + value = 0.0 + return _SchemaScheduleProgress( + id=job_id, + name=data.get("name") or job_name, + provider=data.get("provider") or provider_name, + enable=bool(detail.get("enable", running)), + value=max(min(value, 100), 0), + text=detail.get("text"), + status=data.get("status") or ("running" if running else "waiting"), + success=data.get("success"), + started_at=data.get("started_at") or last_started_at, + finished_at=data.get("finished_at") or last_finished_at, + error=data.get("error") or last_error, + data=data, + ) + @staticmethod def __handle_job_error(job_id: str, job: dict, error: Exception) -> None: """ @@ -726,11 +766,19 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): } if data: progress_data.update(data) - ProgressHelper(self._get_progress_key(job_id)).update( - value=value, - text=text, - data=progress_data, - ) + key = self._get_progress_key(job_id) + + async def _update() -> None: + # 异步后端更新，避免任务函数在事件循环内调用回调时阻塞 + await AsyncProgressHelper(key).update( + value=value, + text=text, + data=progress_data, + ) + + # 回调可能在事件循环内（async 任务）或线程池中（sync 任务）被调用， + # 统一经事件循环提交；无运行中循环时同步执行兜底 + self._submit_to_loop(_update()) return update_progress @@ -778,7 +826,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): error = str(err) self.__handle_job_error(job_id=job_id, job=job, error=err) finally: - self.__finish_job(job_id=job_id, success=success, error=error) + # 协程收尾在事件循环上完成，同步路径（线程池/调用线程）提交到事件循环执行 + await self.__finish_job(job_id=job_id, success=success, error=error) def start(self, job_id: str, *args, **kwargs) -> None: """ @@ -845,8 +894,29 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): self.__handle_job_error(job_id=job_id, job=job, error=e) finally: if not deferred_finish: - # 运行结束 - self.__finish_job(job_id=job_id, success=success, error=error) + # 同步上下文执行异步收尾：优先提交到当前/全局事件循环，无循环时新建循环 + self._submit_to_loop(self.__finish_job( + job_id=job_id, success=success, error=error + )) + + @staticmethod + def _submit_to_loop(coro: Any) -> None: + """ + 把协程提交到事件循环执行，兼容以下调用环境： + - 已在事件循环内（async 任务内部）：排队为独立任务，避免阻塞 + - 外部线程且全局循环在运行：跨线程提交，非阻塞 + - 无运行中循环（测试/CLI）：新建循环同步执行，确保进度不丢失 + """ + try: + running_loop = asyncio.get_running_loop() + except RuntimeError: + running_loop = None + if running_loop: + asyncio.create_task(coro) + elif global_vars.loop and global_vars.loop.is_running(): + asyncio.run_coroutine_threadsafe(coro, global_vars.loop) + else: + asyncio.run(coro) @staticmethod def _get_agent_task_job_id(task_id: int) -> str: diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 8c31829e6..57696eb14 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1652,7 +1652,7 @@ "consumers": [ { "caller": "app.scheduler", - "line": 1046 + "line": 1116 } ], "producers": [] @@ -1665,7 +1665,7 @@ "consumers": [ { "caller": "app.chain.search", - "line": 2861 + "line": 2864 }, { "caller": "app.chain.subscribe", @@ -1813,7 +1813,7 @@ }, { "caller": "app.scheduler", - "line": 696 + "line": 736 } ] },