# [DEF:backend/src/plugins/llm_analysis/scheduler.py:Module] # @TIER: STANDARD # @SEMANTICS: scheduler, task, automation # @PURPOSE: Provides helper functions to schedule LLM-based validation tasks. # @LAYER: Domain # @RELATION: DEPENDS_ON -> backend.src.core.scheduler from typing import Dict, Any from ...dependencies import get_task_manager, get_scheduler_service from ...core.logger import belief_scope, logger # [DEF:schedule_dashboard_validation:Function] # @PURPOSE: Schedules a recurring dashboard validation task. # @PARAM: dashboard_id (str) - ID of the dashboard to validate. # @PARAM: cron_expression (str) - Standard cron expression for scheduling. # @PARAM: params (Dict[str, Any]) - Task parameters (environment_id, provider_id). # @SIDE_EFFECT: Adds a job to the scheduler service. def schedule_dashboard_validation(dashboard_id: str, cron_expression: str, params: Dict[str, Any]): with belief_scope("schedule_dashboard_validation", f"dashboard_id={dashboard_id}"): scheduler = get_scheduler_service() task_manager = get_task_manager() job_id = f"llm_val_{dashboard_id}" async def job_func(): await task_manager.create_task( plugin_id="llm_dashboard_validation", params={ "dashboard_id": dashboard_id, **params } ) scheduler.add_job( job_func, "cron", id=job_id, replace_existing=True, **_parse_cron(cron_expression) ) logger.info(f"Scheduled validation for dashboard {dashboard_id} with cron {cron_expression}") # [/DEF:schedule_dashboard_validation:Function] # [DEF:_parse_cron:Function] # @PURPOSE: Basic cron parser placeholder. # @PARAM: cron (str) - Cron expression. # @RETURN: Dict[str, str] - Parsed cron parts. def _parse_cron(cron: str) -> Dict[str, str]: # Basic cron parser placeholder parts = cron.split() if len(parts) != 5: return {} return { "minute": parts[0], "hour": parts[1], "day": parts[2], "month": parts[3], "day_of_week": parts[4] } # [/DEF:_parse_cron:Function] # [/DEF:backend/src/plugins/llm_analysis/scheduler.py:Module]