718 lines
31 KiB
Python
718 lines
31 KiB
Python
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"
|
||
"• نسبی: <code>in 10m</code> یا <code>۱۰ دقیقه بعد</code> یا <code>2h</code>\n"
|
||
"• بازهای: <code>every 1h</code> یا <code>هر ۳۰ دقیقه (۵ بار)</code>\n"
|
||
"• ساعت دقیق: <code>14:30</code> یا <code>فردا 09:00</code>\n"
|
||
"• کرون: <code>0 9 * * *</code>"
|
||
)
|
||
|
||
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"<b>خروجی استاندارد (stdout):</b>\n<pre>{escape_html(stdout_str[:3000])}</pre>")
|
||
if stderr_str:
|
||
out_parts.append(f"<b>خروجی خطا / وضعیت:</b>\n<pre>{escape_html(stderr_str[:1500])}</pre>")
|
||
if not out_parts:
|
||
out_parts.append("<i>(بدون خروجی)</i>")
|
||
|
||
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"<code>{task.total_runs}/{task.max_runs}</code> (پایان یافت)" if is_fa else f"<code>{task.total_runs}/{task.max_runs}</code> (Completed)"
|
||
else:
|
||
runs_info = f"<code>{task.total_runs}/{task.max_runs}</code>"
|
||
else:
|
||
runs_info = f"<code>{task.total_runs} بار</code>" if is_fa else f"<code>{task.total_runs} runs</code>"
|
||
|
||
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} <b>گزارش اجرای تسک زمانبندی شده{manual_tag}</b>\n\n"
|
||
f"• 🏷️ <b>شناسه تسک:</b> <code>{task.id}</code>\n"
|
||
f"• 📌 <b>عنوان / دستور:</b> <code>{escape_html(task.title or task.content[:40])}</code>\n"
|
||
f"• 📁 <b>پروژه:</b> <code>{escape_html(task.project_name)}</code>\n"
|
||
f"• 🛠 <b>نوع:</b> {type_label}\n"
|
||
f"• ⏱️ <b>زمانبندی:</b> <code>{escape_html(task.timing_spec)}</code>\n"
|
||
f"• 🔄 <b>دفعات اجرا:</b> {runs_info}\n"
|
||
f"• ⌛ <b>مدت زمان:</b> <code>{duration:.1f}s</code>\n"
|
||
f"• 📊 <b>وضعیت:</b> {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"• ⏳ <b>اجرای بعدی:</b> {next_rel} (<code>{format_timestamp(task.next_run_timestamp, is_fa=True)}</code>)\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} <b>Scheduled Task Execution Report{manual_tag}</b>\n\n"
|
||
f"• 🏷️ <b>Task ID:</b> <code>{task.id}</code>\n"
|
||
f"• 📌 <b>Title:</b> <code>{escape_html(task.title or task.content[:40])}</code>\n"
|
||
f"• 📁 <b>Project:</b> <code>{escape_html(task.project_name)}</code>\n"
|
||
f"• 🛠 <b>Type:</b> {type_label}\n"
|
||
f"• ⏱️ <b>Timing:</b> <code>{escape_html(task.timing_spec)}</code>\n"
|
||
f"• 🔄 <b>Runs:</b> {runs_info}\n"
|
||
f"• ⌛ <b>Duration:</b> <code>{duration:.1f}s</code>\n"
|
||
f"• 📊 <b>Status:</b> {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"• ⏳ <b>Next Run:</b> {next_rel} (<code>{format_timestamp(task.next_run_timestamp, is_fa=False)}</code>)\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"🔔 <b>یادآوری:</b>\n{escape_html(output_text)}" if is_fa else f"🔔 <b>Reminder:</b>\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()
|