import os import sys import json import time import re import uuid import asyncio import logging from pathlib import Path from typing import Optional, Dict, Any, List, Tuple from dataclasses import dataclass, field, asdict from datetime import datetime, timedelta from config import settings from formatters import markdown_to_telegram_html, escape_html, split_message logger = logging.getLogger("AGYScheduler") # Convert Persian and Arabic digits to standard ASCII digits PERSIAN_ARABIC_DIGIT_MAP = str.maketrans("۰۱۲۳۴۵۶۷۸۹٠١٢٣٤٥٦٧٨٩", "01234567890123456789") def normalize_digits(text: str) -> str: if not text: return "" return text.translate(PERSIAN_ARABIC_DIGIT_MAP) def format_relative_time(seconds_diff: float, is_fa: bool = True) -> str: """Formats seconds difference into a human readable string.""" if seconds_diff <= 0: return "همین الان" if is_fa else "now" sec = int(seconds_diff) days = sec // 86400 hours = (sec % 86400) // 3600 minutes = (sec % 3600) // 60 seconds = sec % 60 parts = [] if days > 0: parts.append(f"{days} روز" if is_fa else f"{days}d") if hours > 0: parts.append(f"{hours} ساعت" if is_fa else f"{hours}h") if minutes > 0: parts.append(f"{minutes} دقیقه" if is_fa else f"{minutes}m") if not parts or (days == 0 and hours == 0 and minutes < 5 and seconds > 0): parts.append(f"{seconds} ثانیه" if is_fa else f"{seconds}s") res = " و ".join(parts) if is_fa else " ".join(parts) return f"{res} دیگر" if is_fa else f"in {res}" def format_timestamp(ts: Optional[float], is_fa: bool = True) -> str: if not ts: return "نامشخص" if is_fa else "N/A" dt = datetime.fromtimestamp(ts) return dt.strftime("%Y-%m-%d %H:%M:%S") # Cron Next Run Matcher (Zero-dependency 5-field standard cron calculator) def _parse_cron_field(field_str: str, min_val: int, max_val: int) -> set[int]: """Parses a single cron field (*, */5, 1-5, 1,2,3) into a set of valid integers.""" field_str = field_str.strip() result = set() for part in field_str.split(","): part = part.strip() if not part: continue if part == "*": result.update(range(min_val, max_val + 1)) elif "/" in part: subparts = part.split("/") step = int(subparts[1]) if subparts[0] == "*": start_v, end_v = min_val, max_val elif "-" in subparts[0]: r = subparts[0].split("-") start_v, end_v = int(r[0]), int(r[1]) else: start_v, end_v = int(subparts[0]), max_val result.update(range(start_v, end_v + 1, step)) elif "-" in part: r = part.split("-") start_v, end_v = int(r[0]), int(r[1]) result.update(range(start_v, end_v + 1)) else: val = int(part) if min_val <= val <= max_val: result.add(val) return result def get_next_cron_timestamp(cron_expr: str, base_timestamp: Optional[float] = None) -> Optional[float]: """Calculates the next Unix timestamp for a given 5-field cron expression.""" if not base_timestamp: base_timestamp = time.time() parts = cron_expr.strip().split() if len(parts) != 5: return None try: min_set = _parse_cron_field(parts[0], 0, 59) hour_set = _parse_cron_field(parts[1], 0, 23) day_set = _parse_cron_field(parts[2], 1, 31) month_set = _parse_cron_field(parts[3], 1, 12) # Cron dow: 0=Sun or 7=Sun in standard cron, or 0-6. Python weekday: 0=Mon..6=Sun. # Standard cron: 0=Sun, 1=Mon, 2=Tue, 3=Wed, 4=Thu, 5=Fri, 6=Sat, 7=Sun raw_dow = _parse_cron_field(parts[4], 0, 7) dow_set = set() for d in raw_dow: # Map cron dow (0=Sun..6=Sat, 7=Sun) to python weekday (0=Mon..6=Sun) if d == 0 or d == 7: dow_set.add(6) # Sunday else: dow_set.add(d - 1) # Mon=0, Tue=1, etc. except Exception as e: logger.error(f"Error parsing cron expression '{cron_expr}': {e}") return None # Step minute by minute starting from next minute curr = datetime.fromtimestamp(base_timestamp).replace(second=0, microsecond=0) + timedelta(minutes=1) # Search up to 5 years (max ~2.6 million minutes) for _ in range(525600 * 5): if curr.month not in month_set: # Skip to start of next month if curr.month == 12: curr = datetime(curr.year + 1, 1, 1, 0, 0) else: curr = datetime(curr.year, curr.month + 1, 1, 0, 0) continue if curr.day not in day_set or curr.weekday() not in dow_set: # Advance to start of next day curr = (curr + timedelta(days=1)).replace(hour=0, minute=0) continue if curr.hour not in hour_set: curr = (curr + timedelta(hours=1)).replace(minute=0) continue if curr.minute in min_set: return curr.timestamp() curr += timedelta(minutes=1) return None def extract_max_runs(timing_str: str) -> Tuple[str, Optional[int]]: """Extracts repeat count / max runs like '(3 times)', '[5 بار]', '(repeat: 4)', 'تکرار: 3'.""" raw = normalize_digits(timing_str).strip() # 1. Bracketed or parenthesized patterns: (3 times), [3 بار], (تکرار: 5), [max: 10] p1 = r"[\(\[\{]\s*(?:تکرار|تعداد\s*تکرار|repeat|max|max_runs|count|times|دفعه|بار)?\s*[:=]?\s*(\d+)\s*(?:times|bar|مرتبه|بار|دفعه|x)?\s*[\)\]\}]" m1 = re.search(p1, raw, flags=re.IGNORECASE) if m1: max_runs = int(m1.group(1)) cleaned = re.sub(p1, "", raw, flags=re.IGNORECASE).strip() return cleaned, max_runs # 2. Key-value style: repeat: 5, تکرار: 3 p2 = r"(?:تکرار|تعداد\s*تکرار|repeat|max_runs|count)\s*[:=]\s*(\d+)" m2 = re.search(p2, raw, flags=re.IGNORECASE) if m2: max_runs = int(m2.group(1)) cleaned = re.sub(p2, "", raw, flags=re.IGNORECASE).strip() return cleaned, max_runs # 3. Suffix style: 'every 10m 5 times' or 'هر ۱۰ دقیقه ۳ بار' p3 = r"\s+(\d+)\s*(?:بار|مرتبه|دفعه|times|x)\s*$" m3 = re.search(p3, raw, flags=re.IGNORECASE) if m3: max_runs = int(m3.group(1)) cleaned = re.sub(p3, "", raw, flags=re.IGNORECASE).strip() return cleaned, max_runs return raw, None def parse_timing_string(raw_timing: str) -> Tuple[str, float, Optional[float], Optional[str], Optional[int]]: """ Parses natural language timing into: (timing_type, next_run_timestamp, interval_seconds, cron_expression, max_runs) Supported formats: - Relative: 'in 10m', 'in 2h', '10m', '2h', '1d', '30s', '۱۰ دقیقه بعد', '۲ ساعت دیگر' - Interval: 'every 10m', 'every 1h', 'every 30s', 'هر ۱۰ دقیقه (۳ بار)', 'هر ۱ ساعت', 'هر روز' - Exact time: '14:30', '09:00', '2026-08-29 18:00', 'ساعت 14:30', 'فردا 10:00' - Cron: '0 9 * * *', '*/15 * * * *', 'cron: 0 9 * * *' """ cleaned_timing, max_runs = extract_max_runs(raw_timing) timing = normalize_digits(cleaned_timing).strip().lower() now = time.time() # 1. Cron Expression cron_match = re.match(r"^(?:cron:\s*)?((?:[\*\d/,-]+\s+){4}[\*\d/,-]+)$", timing) if cron_match: cron_expr = cron_match.group(1).strip() next_ts = get_next_cron_timestamp(cron_expr, now) if next_ts: return ("cron", next_ts, None, cron_expr, max_runs) else: raise ValueError(f"عبارت کرون نامعتبر است / Invalid cron expression: {cron_expr}") # 2. Interval: 'every ...' or 'هر ...' interval_match = re.match(r"^(?:every|هر|هریک|هر یک)\s+(.+)$", timing) if interval_match: sub = interval_match.group(1).strip() sec = parse_duration_seconds(sub) if sec and sec >= 5: return ("interval", now + sec, sec, None, max_runs) elif sub in ("روز", "day", "daily"): return ("interval", now + 86400, 86400.0, None, max_runs) elif sub in ("ساعت", "hour", "hourly"): return ("interval", now + 3600, 3600.0, None, max_runs) elif sub in ("هفته", "week", "weekly"): return ("interval", now + 604800, 604800.0, None, max_runs) else: raise ValueError(f"بازه زمانی نامعتبر است / Invalid interval: '{sub}'") # 3. Relative: 'in 10m', '10m', '10 دقیقه بعد', 'در ۱۰ دقیقه' rel_clean = re.sub(r"^(?:in|بعد\s+از|در|تا)\s+", "", timing).strip() rel_clean = re.sub(r"\s+(?:بعد|دیگر|آینده|بعدا)$", "", rel_clean).strip() sec = parse_duration_seconds(rel_clean) if sec and sec > 0: return ("once_relative", now + sec, None, None, 1 if max_runs is None else max_runs) # 4. Exact Date & Time: YYYY-MM-DD HH:MM[:SS] dt_match = re.match(r"^(\d{4}-\d{1,2}-\d{1,2})\s+(\d{1,2}:\d{2}(?::\d{2})?)$", timing) if dt_match: dt_str = f"{dt_match.group(1)} {dt_match.group(2)}" fmt = "%Y-%m-%d %H:%M:%S" if len(dt_match.group(2).split(":")) == 3 else "%Y-%m-%d %H:%M" try: target_dt = datetime.strptime(dt_str, fmt) target_ts = target_dt.timestamp() if target_ts <= now: raise ValueError("زمان مشخص شده در گذشته است / Specified time is in the past.") return ("once_date", target_ts, None, None, 1 if max_runs is None else max_runs) except ValueError as ve: if "past" in str(ve): raise pass # 5. Time only: HH:MM (e.g., 14:30 or ساعت 14:30) or Tomorrow HH:MM is_tomorrow = False if "فردا" in timing or "tomorrow" in timing: is_tomorrow = True timing = timing.replace("فردا", "").replace("tomorrow", "").strip() time_clean = re.sub(r"^(?:ساعت|at|time)\s+", "", timing).strip() time_match = re.match(r"^(\d{1,2}):(\d{2})(?::(\d{2}))?$", time_clean) if time_match: h = int(time_match.group(1)) m = int(time_match.group(2)) s = int(time_match.group(3)) if time_match.group(3) else 0 if 0 <= h <= 23 and 0 <= m <= 59 and 0 <= s <= 59: now_dt = datetime.now() target_dt = now_dt.replace(hour=h, minute=m, second=s, microsecond=0) if is_tomorrow or target_dt.timestamp() <= now: target_dt += timedelta(days=1) return ("once_date", target_dt.timestamp(), None, None, 1 if max_runs is None else max_runs) raise ValueError( f"فرمت زمان نامعتبر است: '{raw_timing}'.\n" "مثال‌های معتبر:\n" "• نسبی: in 10m یا ۱۰ دقیقه بعد یا 2h\n" "• بازه‌ای: every 1h یا هر ۳۰ دقیقه (۵ بار)\n" "• ساعت دقیق: 14:30 یا فردا 09:00\n" "• کرون: 0 9 * * *" ) def parse_duration_seconds(duration_str: str) -> Optional[float]: """Parses duration strings like '10m', '2h', '30s', '1d', '10 دقیقه', '۲ ساعت' into seconds.""" s = duration_str.strip().lower() # Special keywords if s in ("نیم ساعت", "half hour", "0.5h"): return 1800.0 if s in ("یک ربع", "15m", "15 min"): return 900.0 # Match number + unit pattern = r"^(\d+(?:\.\d+)?)\s*([a-zA-Z\u0600-\u06FF]+)?$" m = re.match(pattern, s) if not m: return None val = float(m.group(1)) unit = (m.group(2) or "m").lower() if unit in ("s", "sec", "second", "seconds", "ثانیه"): return val elif unit in ("m", "min", "mins", "minute", "minutes", "دقیقه"): return val * 60 elif unit in ("h", "hr", "hrs", "hour", "hours", "ساعت"): return val * 3600 elif unit in ("d", "day", "days", "روز"): return val * 86400 elif unit in ("w", "week", "weeks", "هفته"): return val * 604800 elif unit in ("mo", "month", "months", "ماه"): return val * 2592000 return None @dataclass class ScheduledTask: id: str chat_id: int creator_id: int project_name: str task_type: str # "prompt" | "command" | "reminder" content: str timing_type: str # "once_relative" | "once_date" | "interval" | "cron" timing_spec: str next_run_timestamp: float interval_seconds: Optional[float] = None cron_expression: Optional[str] = None max_runs: Optional[int] = None status: str = "active" # "active" | "paused" | "completed" | "failed" created_at: float = field(default_factory=time.time) last_run_timestamp: Optional[float] = None last_run_status: Optional[str] = None last_run_result: Optional[str] = None total_runs: int = 0 title: str = "" def is_recurring(self) -> bool: return self.timing_type in ("interval", "cron") def to_dict(self) -> Dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> "ScheduledTask": clean = {k: v for k, v in data.items() if k in cls.__dataclass_fields__} return cls(**clean) class TaskScheduler: def __init__(self, storage_file: Path = Path("/root/telegram-agy-bot/tasks.json")): self.storage_file = storage_file self.tasks: Dict[str, ScheduledTask] = {} self._lock = asyncio.Lock() self._running = False self._bg_task: Optional[asyncio.Task] = None self._app: Optional[Any] = None self._load() def _load(self): if self.storage_file.exists(): try: with open(self.storage_file, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict): for tid, tdata in data.items(): if isinstance(tdata, dict): self.tasks[tid] = ScheduledTask.from_dict(tdata) logger.info(f"Loaded {len(self.tasks)} scheduled tasks from {self.storage_file}") except Exception as e: logger.error(f"Failed to load tasks from {self.storage_file}: {e}") def save(self): try: temp_file = self.storage_file.with_suffix(".tmp") data = {tid: t.to_dict() for tid, t in self.tasks.items()} with open(temp_file, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) temp_file.replace(self.storage_file) except Exception as e: logger.error(f"Failed to save tasks to {self.storage_file}: {e}") def add_task( self, chat_id: int, creator_id: int, project_name: str, task_type: str, content: str, timing_spec: str, title: str = "", max_runs: Optional[int] = None, ) -> ScheduledTask: timing_type, next_ts, interval_sec, cron_expr, parsed_max_runs = parse_timing_string(timing_spec) effective_max_runs = max_runs if max_runs is not None else parsed_max_runs task_id = f"t-{uuid.uuid4().hex[:6]}" task = ScheduledTask( id=task_id, chat_id=chat_id, creator_id=creator_id, project_name=project_name, task_type=task_type, content=content, timing_type=timing_type, timing_spec=timing_spec, next_run_timestamp=next_ts, interval_seconds=interval_sec, cron_expression=cron_expr, max_runs=effective_max_runs, status="active", created_at=time.time(), title=title or (content[:30] + ("..." if len(content) > 30 else "")), ) self.tasks[task_id] = task self.save() logger.info(f"Scheduled task added: {task_id} ({timing_type} - {timing_spec} - max_runs={effective_max_runs}) for project {project_name}") return task def get_task(self, task_id: str) -> Optional[ScheduledTask]: return self.tasks.get(task_id) def get_user_tasks(self, chat_id: int) -> List[ScheduledTask]: return [t for t in self.tasks.values() if t.chat_id == chat_id or t.creator_id == chat_id] def get_all_tasks(self) -> List[ScheduledTask]: return list(self.tasks.values()) def pause_task(self, task_id: str) -> bool: task = self.tasks.get(task_id) if task and task.status == "active": task.status = "paused" self.save() return True return False def resume_task(self, task_id: str) -> bool: task = self.tasks.get(task_id) if task and task.status == "paused": now = time.time() if task.next_run_timestamp <= now: if task.timing_type == "interval" and task.interval_seconds: task.next_run_timestamp = now + task.interval_seconds elif task.timing_type == "cron" and task.cron_expression: task.next_run_timestamp = get_next_cron_timestamp(task.cron_expression, now) or (now + 60) else: task.next_run_timestamp = now + 10 # run soon if one-time in past task.status = "active" self.save() return True return False def delete_task(self, task_id: str) -> bool: if task_id in self.tasks: del self.tasks[task_id] self.save() return True return False def clear_completed_tasks(self, chat_id: int) -> int: to_del = [tid for tid, t in self.tasks.items() if (t.chat_id == chat_id or t.creator_id == chat_id) and t.status in ("completed", "failed")] for tid in to_del: del self.tasks[tid] if to_del: self.save() return len(to_del) async def execute_task_now(self, task_id: str, app: Any) -> Tuple[bool, str]: task = self.tasks.get(task_id) if not task: return False, "تسک یافت نشد / Task not found" return await self._run_single_task(task, app, is_manual=True) async def _run_single_task(self, task: ScheduledTask, app: Any, is_manual: bool = False) -> Tuple[bool, str]: from agy_engine import session_manager, AGYEngine, Session, Project logger.info(f"Executing scheduled task {task.id} (type: {task.task_type}, project: {task.project_name}, run: {task.total_runs + 1})...") start_t = time.time() output_text = "" success = False # Get project & session context session = session_manager.get_or_create(task.chat_id) is_fa = (session.language or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)") target_proj = session.projects.get(task.project_name) if not target_proj: # Fallback to current project or create dummy context target_proj = session.current_project workspace = target_proj.workspace if target_proj else settings.default_workspace model = target_proj.model if target_proj else settings.default_model effort = target_proj.effort if target_proj else settings.default_effort lang = target_proj.language if target_proj else session.language try: if task.task_type == "prompt": # Create dedicated execution context task_session = Session( chat_id=task.chat_id, active_project=task.project_name, projects=session.projects, language=lang, ) result = await AGYEngine.run_prompt( session=task_session, prompt=task.content, ) output_text = result.text or "✅ دستور با موفقیت اجرا شد (بدون خروجی متنی)." try: from bot_actions import process_all_ai_actions output_text, _, _, _ = await process_all_ai_actions( raw_text=output_text, chat_id=task.chat_id, project_name=task.project_name, is_fa=is_fa, app=app, ) except Exception as act_err: logger.warning(f"Error executing AI action tags in scheduled task: {act_err}") success = True elif task.task_type == "command": # Run bash command with timeout protection (5 minutes default) proc = await asyncio.create_subprocess_shell( task.content, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workspace, start_new_session=True, ) try: stdout_b, stderr_b = await asyncio.wait_for(proc.communicate(), timeout=300.0) stdout_str = stdout_b.decode("utf-8", errors="replace").strip() stderr_str = stderr_b.decode("utf-8", errors="replace").strip() success = (proc.returncode == 0) except asyncio.TimeoutError: try: import signal os.killpg(os.getpgid(proc.pid), signal.SIGKILL) except Exception: try: proc.kill() except Exception: pass stdout_str = "" stderr_str = "⏱️ مدت زمان اجرای دستور بیش از ۵ دقیقه طول کشید و به دلیل اتمام زمان مجاز (Timeout) متوقف شد." if is_fa else "⏱️ Command execution timed out (>5 minutes) and was automatically terminated." success = False out_parts = [] if stdout_str: out_parts.append(f"خروجی استاندارد (stdout):\n
{escape_html(stdout_str[:3000])}
") if stderr_str: out_parts.append(f"خروجی خطا / وضعیت:\n
{escape_html(stderr_str[:1500])}
") if not out_parts: out_parts.append("(بدون خروجی)") output_text = "\n\n".join(out_parts) elif task.task_type == "reminder": output_text = task.content success = True else: output_text = f"نوع تسک ناشناخته: {task.task_type}" success = False except Exception as e: logger.error(f"Error executing task {task.id}: {e}", exc_info=True) output_text = f"❌ خطا در اجرا: {str(e)}" success = False duration = time.time() - start_t task.last_run_timestamp = time.time() task.total_runs += 1 task.last_run_status = "success" if success else "failed" task.last_run_result = output_text[:500] # Calculate next execution time or mark completed if max_runs reached if task.max_runs and task.total_runs >= task.max_runs: task.status = "completed" elif task.is_recurring() and task.status == "active": now = time.time() if task.timing_type == "interval" and task.interval_seconds: task.next_run_timestamp = now + task.interval_seconds elif task.timing_type == "cron" and task.cron_expression: task.next_run_timestamp = get_next_cron_timestamp(task.cron_expression, now) or (now + 60) else: if not is_manual: task.status = "completed" if success else "failed" self.save() # Send Telegram notification to user if app and app.bot: await self._send_task_notification(task, output_text, duration, is_manual, is_fa, app) return success, output_text async def _send_task_notification( self, task: ScheduledTask, output_text: str, duration: float, is_manual: bool, is_fa: bool, app: Any, ): from telegram import InlineKeyboardButton, InlineKeyboardMarkup, constants type_icons = {"prompt": "🧠", "command": "⚡", "reminder": "🔔"} icon = type_icons.get(task.task_type, "⏰") manual_tag = " (اجرای دستی)" if (is_manual and is_fa) else " (Manual Run)" if is_manual else "" runs_info = "" if task.max_runs: if task.status == "completed": runs_info = f"{task.total_runs}/{task.max_runs} (پایان یافت)" if is_fa else f"{task.total_runs}/{task.max_runs} (Completed)" else: runs_info = f"{task.total_runs}/{task.max_runs}" else: runs_info = f"{task.total_runs} بار" if is_fa else f"{task.total_runs} runs" if is_fa: type_label = "پرامپت هوش مصنوعی" if task.task_type == "prompt" else "دستور شل" if task.task_type == "command" else "یادآور" status_str = "✅ با موفقیت انجام شد" if task.last_run_status == "success" else "❌ با خطا مواجه شد" header = ( f"{icon} گزارش اجرای تسک زمان‌بندی شده{manual_tag}\n\n" f"• 🏷️ شناسه تسک: {task.id}\n" f"• 📌 عنوان / دستور: {escape_html(task.title or task.content[:40])}\n" f"• 📁 پروژه: {escape_html(task.project_name)}\n" f"• 🛠 نوع: {type_label}\n" f"• ⏱️ زمان‌بندی: {escape_html(task.timing_spec)}\n" f"• 🔄 دفعات اجرا: {runs_info}\n" f"• ⌛ مدت زمان: {duration:.1f}s\n" f"• 📊 وضعیت: {status_str}\n" ) if task.is_recurring() and task.status == "active": next_rel = format_relative_time(task.next_run_timestamp - time.time(), is_fa=True) header += f"• ⏳ اجرای بعدی: {next_rel} ({format_timestamp(task.next_run_timestamp, is_fa=True)})\n" header += "──────────────\n\n" else: type_label = "AI Prompt" if task.task_type == "prompt" else "Shell Command" if task.task_type == "command" else "Reminder" status_str = "✅ Completed Successfully" if task.last_run_status == "success" else "❌ Execution Failed" header = ( f"{icon} Scheduled Task Execution Report{manual_tag}\n\n" f"• 🏷️ Task ID: {task.id}\n" f"• 📌 Title: {escape_html(task.title or task.content[:40])}\n" f"• 📁 Project: {escape_html(task.project_name)}\n" f"• 🛠 Type: {type_label}\n" f"• ⏱️ Timing: {escape_html(task.timing_spec)}\n" f"• 🔄 Runs: {runs_info}\n" f"• ⌛ Duration: {duration:.1f}s\n" f"• 📊 Status: {status_str}\n" ) if task.is_recurring() and task.status == "active": next_rel = format_relative_time(task.next_run_timestamp - time.time(), is_fa=False) header += f"• ⏳ Next Run: {next_rel} ({format_timestamp(task.next_run_timestamp, is_fa=False)})\n" header += "──────────────\n\n" if task.task_type == "prompt": formatted_body = markdown_to_telegram_html(output_text) elif task.task_type == "command": formatted_body = output_text else: formatted_body = f"🔔 یادآوری:\n{escape_html(output_text)}" if is_fa else f"🔔 Reminder:\n{escape_html(output_text)}" full_msg = header + formatted_body chunks = split_message(full_msg, max_length=settings.max_message_length) keyboard = [ [ InlineKeyboardButton("🔄 اجرای مجدد" if is_fa else "🔄 Run Again", callback_data=f"task_run:{task.id}"), InlineKeyboardButton("⚙️ مدیریت این تسک" if is_fa else "⚙️ Task Details", callback_data=f"task_detail:{task.id}"), ], [ InlineKeyboardButton("⏰ لیست همه تسک‌ها" if is_fa else "⏰ All Tasks", callback_data="btn_tasks_menu"), InlineKeyboardButton("🏠 منوی اصلی" if is_fa else "🏠 Main Dashboard", callback_data="btn_dashboard"), ] ] try: for i, chunk in enumerate(chunks): is_last = (i == len(chunks) - 1) await app.bot.send_message( chat_id=task.chat_id, text=chunk, parse_mode=constants.ParseMode.HTML, reply_markup=InlineKeyboardMarkup(keyboard) if is_last else None, disable_web_page_preview=True, ) except Exception as err: logger.error(f"Failed to send task {task.id} execution message to {task.chat_id}: {err}") async def _scheduler_loop(self): logger.info("Task Scheduler background loop started.") while self._running: try: now = time.time() due_tasks: List[ScheduledTask] = [] async with self._lock: for task in self.tasks.values(): if task.status == "active" and task.next_run_timestamp <= now: due_tasks.append(task) for task in due_tasks: # Spawn task execution asynchronously without blocking the loop asyncio.create_task(self._run_single_task(task, self._app, is_manual=False)) except Exception as e: logger.error(f"Error in task scheduler loop: {e}", exc_info=True) await asyncio.sleep(2.0) def start(self, app: Any): if not self._running: self._running = True self._app = app self._bg_task = asyncio.create_task(self._scheduler_loop()) logger.info("Task Scheduler started.") def stop(self): self._running = False if self._bg_task: self._bg_task.cancel() self._bg_task = None logger.info("Task Scheduler stopped.") # Singleton task scheduler instance task_scheduler = TaskScheduler()