""" 调度器业务能力装配与生命周期入口。 """ import asyncio from typing import Awaitable, Callable, Optional from app.application.agenttask import AgentTaskRepository from app.application.image import WallpaperHelper from app.application.outbox import stop_outbox_recovery from app.application.scheduling import ( get_scheduler, register_scheduler_class, reset_scheduler_class, ) from app.chain.mediaserver import MediaServerChain from app.chain.recommend import RecommendChain from app.chain.site import SiteChain from app.chain.subscribe.facade import SubscribeChain from app.chain.transfer.facade import TransferChain from app.chain.workflow import WorkflowChain from app.runtime.execution import run_in_threadpool_to_completion from app.scheduler.chain import SchedulerChain from app.scheduler.facade import Scheduler from app.scheduler.services import SchedulerServices def configure_scheduler_runtime() -> None: """ 在显式装配阶段登记 concrete Scheduler。 """ register_scheduler_class(Scheduler) def reset_scheduler_runtime() -> None: """ 清除 concrete Scheduler 登记,支持重复 lifespan。 """ reset_scheduler_class() def reset_scheduler_bindings() -> None: """ 在 Scheduler 已停止后撤销当前 lifespan 的仓储、业务能力和类登记。 """ existing_scheduler: Callable[[], Optional[Scheduler]] = Scheduler.get_existing_instance scheduler = existing_scheduler() if scheduler is not None: scheduler.reset_runtime_bindings() reset_scheduler_runtime() def configure_scheduler_agent_tasks(repository: AgentTaskRepository) -> None: """ 在调度器启动前注入自主任务仓储。 """ configure_scheduler_runtime() get_scheduler().configure_agent_tasks(repository) def configure_scheduler_services() -> None: """ 构造一次调度任务能力,并在启动前注入 concrete Scheduler。 """ scheduler_chain = SchedulerChain() site_chain = SiteChain() mediaserver_chain = MediaServerChain() subscribe_chain = SubscribeChain() transfer_chain = TransferChain() recommend_chain = RecommendChain() workflow_chain = WorkflowChain() Scheduler().configure_services( SchedulerServices( sync_cookies=site_chain.sync_cookies, sync_mediaserver=mediaserver_chain.sync, check_subscribe=subscribe_chain.check_and_reconcile, search_subscribe=subscribe_chain.search, resume_subscribe_search=subscribe_chain.async_resume_search_queue, refresh_subscribe=subscribe_chain.refresh, follow_subscribe=subscribe_chain.follow, process_transfer=transfer_chain.process, clear_cache=scheduler_chain.clear_cache, cleanup_data=scheduler_chain.cleanup, run_modules=scheduler_chain.scheduler_job, get_wallpapers=WallpaperHelper().get_wallpapers, clear_wallpaper_cache=WallpaperHelper().clear_cache, refresh_site_data=site_chain.refresh_userdatas, refresh_recommend=recommend_chain.refresh_recommend, cache_subscribe_calendar=subscribe_chain.cache_calendar, list_workflows=workflow_chain.get_timer_workflows, process_workflow=workflow_chain.process, put_message=scheduler_chain.messagehelper.put, post_message=scheduler_chain.post_message, ) ) def init_scheduler() -> None: """ 初始化定时器 """ configure_scheduler_runtime() configure_scheduler_services() try: Scheduler().init() except Exception: reset_scheduler_runtime() raise def stop_scheduler() -> Optional[Awaitable[None]]: """ 停止定时器;生命周期事件循环中返回可等待的收口协程。 """ scheduler = Scheduler() try: asyncio.get_running_loop() except RuntimeError: scheduler.stop() if not stop_outbox_recovery(): raise RuntimeError("Outbox 后台处理尚未停止") reset_scheduler_bindings() return None async def stop_and_reset() -> None: """ 等待异步调度器收口后清除 provider。 """ await scheduler.stop_async() if not await run_in_threadpool_to_completion(stop_outbox_recovery): raise RuntimeError("Outbox 后台处理尚未停止") reset_scheduler_bindings() return stop_and_reset() def restart_scheduler() -> None: """ 重启定时器 """ configure_scheduler_runtime() Scheduler().init() def init_plugin_scheduler() -> None: """ 初始化插件定时器 """ Scheduler().init_plugin_jobs()