import os import sys import re import json import time import shutil import asyncio import logging from pathlib import Path from typing import Optional, Dict, Any, List, Tuple, Callable from datetime import datetime import uuid from telegram import InlineKeyboardButton, InlineKeyboardMarkup, WebAppInfo from config import settings from agy_engine import ( session_manager, AVAILABLE_MODELS, AVAILABLE_EFFORTS, AVAILABLE_LANGUAGES, get_valid_effort_for_model, Session, Project, ) from scheduler import ( task_scheduler, ScheduledTask, format_relative_time, format_timestamp, ) from backup_manager import backup_manager from invite_manager import invite_manager from formatters import markdown_to_telegram_html, escape_html, split_message logger = logging.getLogger("AGYBotActions") CADDYFILE_PATH = Path("/etc/caddy/Caddyfile") WILDCARD_DOMAIN = "msa.artacloud.ir" # ===================================================================== # Model & Setting Resolvers # ===================================================================== def resolve_model_name(req_model: str) -> Optional[str]: """Resolves exact or fuzzy model nicknames to standard model ID.""" if not req_model: return None raw = req_model.lower().strip() # Exact match for m in AVAILABLE_MODELS.keys(): if raw == m.lower(): return m # Fuzzy nicknames if any(w in raw for w in ("auto", "خودکار", "اتوماتیک", "اتو", "flash-auto", "3.7-auto")): return "gemini-3.7-flash-auto" if "opus" in raw: return "claude-opus-4-6-thinking" if "sonnet" in raw or "claude" in raw: return "claude-sonnet-4-6" if "3.1" in raw or "pro" in raw: return "gemini-3.1-pro" if "flash-high" in raw or ("high" in raw and "flash" in raw): return "gemini-3.7-flash-high" if "3.6" in raw: return "gemini-3.6-flash" if "3.7" in raw or "flash" in raw: return "gemini-3.7-flash" if "gpt" in raw or "120b" in raw or "oss" in raw: return "gpt-oss-120b" return None def resolve_effort_level(req_effort: str) -> Optional[str]: """Resolves effort level string (low, medium, high).""" if not req_effort: return None raw = req_effort.lower().strip() if raw in ("low", "medium", "high"): return raw if any(w in raw for w in ("high", "زیاد", "بالا", "حداکثر")): return "high" if any(w in raw for w in ("med", "medium", "متوسط", "معمولی")): return "medium" if any(w in raw for w in ("low", "کم", "پایین", "حداقل")): return "low" return None def resolve_language_code(req_lang: str) -> Optional[str]: """Resolves language code from alias or name.""" if not req_lang: return None raw = req_lang.lower().strip() try: from bot import LANG_ALIASES raw = LANG_ALIASES.get(raw, raw) except Exception: pass if raw in AVAILABLE_LANGUAGES: return raw if any(w in raw for w in ("fa", "farsi", "persian", "فارسی")): return "fa" if any(w in raw for w in ("en", "eng", "english", "انگلیسی")): return "en" if any(w in raw for w in ("auto", "خودکار", "detect")): return "auto" return None # ===================================================================== # Caddy Reverse Proxy Helpers # ===================================================================== def configure_caddy_reverse_proxy( subdomain: str, port_or_target: str, custom_domain: Optional[str] = None, ) -> Tuple[bool, str, str]: """ Configures Caddyfile to reverse-proxy a subdomain or domain to a local port or target. Auto-validates and reloads Caddy. Returns (success, message, full_url). """ sub = subdomain.strip().lower() sub = re.sub(r"[^\w\-.]", "", sub) if custom_domain: host = custom_domain.strip().lower() elif "." in sub: host = sub else: host = f"{sub}.{WILDCARD_DOMAIN}" target = str(port_or_target).strip() if target.isdigit(): target = f"localhost:{target}" elif not target.startswith("http://") and not target.startswith("https://") and not target.startswith("localhost:") and not target.startswith("127.0.0.1:"): target = f"localhost:{target}" if not CADDYFILE_PATH.exists(): try: CADDYFILE_PATH.parent.mkdir(parents=True, exist_ok=True) CADDYFILE_PATH.write_text("# Caddyfile\n:80 {\n\troot * /usr/share/caddy\n\tfile_server\n}\n", encoding="utf-8") except Exception as e: return False, f"Failed to initialize Caddyfile: {e}", "" try: content = CADDYFILE_PATH.read_text(encoding="utf-8") backup_content = content block_pattern = re.compile( rf"(?:^|\n)\s*{re.escape(host)}\s*\{{[^}}]*\}}", re.MULTILINE | re.DOTALL, ) new_block = f"\n{host} {{\n\treverse_proxy {target}\n}}\n" if block_pattern.search(content): new_content = block_pattern.sub(new_block.strip("\n"), content) else: new_content = content.rstrip() + "\n" + new_block # Save and validate CADDYFILE_PATH.write_text(new_content, encoding="utf-8") # Validate with caddy CLI proc = os.system(f"caddy validate --config {CADDYFILE_PATH} > /dev/null 2>&1") if proc != 0: CADDYFILE_PATH.write_text(backup_content, encoding="utf-8") return False, "Caddy configuration validation failed. Reverted to previous state.", "" # Reload Caddy service reload_res = os.system("systemctl reload caddy > /dev/null 2>&1 || caddy reload --config /etc/caddy/Caddyfile > /dev/null 2>&1") if reload_res != 0: logger.warning("Caddy reload returned non-zero, trying restart...") os.system("systemctl restart caddy > /dev/null 2>&1") full_url = f"https://{host}" return True, f"Reverse proxy for {host} configured successfully.", full_url except Exception as e: logger.error(f"Error configuring Caddy reverse proxy: {e}", exc_info=True) return False, f"Error configuring Caddy: {str(e)}", "" def remove_caddy_reverse_proxy(subdomain: str, custom_domain: Optional[str] = None) -> Tuple[bool, str]: """Removes a reverse proxy block from Caddyfile and reloads Caddy.""" sub = subdomain.strip().lower() sub = re.sub(r"[^\w\-.]", "", sub) if custom_domain: host = custom_domain.strip().lower() elif "." in sub: host = sub else: host = f"{sub}.{WILDCARD_DOMAIN}" if not CADDYFILE_PATH.exists(): return False, "Caddyfile does not exist." try: content = CADDYFILE_PATH.read_text(encoding="utf-8") backup_content = content block_pattern = re.compile( rf"(?:^|\n)\s*{re.escape(host)}\s*\{{[^}}]*\}}", re.MULTILINE | re.DOTALL, ) if not block_pattern.search(content): return False, f"ساب‌دامین {host} در Caddyfile یافت نشد." new_content = block_pattern.sub("", content).strip() + "\n" CADDYFILE_PATH.write_text(new_content, encoding="utf-8") proc = os.system(f"caddy validate --config {CADDYFILE_PATH} > /dev/null 2>&1") if proc != 0: CADDYFILE_PATH.write_text(backup_content, encoding="utf-8") return False, "Caddy validation failed after removal. Reverted." os.system("systemctl reload caddy > /dev/null 2>&1 || caddy reload --config /etc/caddy/Caddyfile > /dev/null 2>&1") return True, f"ساب‌دامین {host} با موفقیت از Caddy حذف گردید." except Exception as e: logger.error(f"Error removing Caddy subdomain: {e}", exc_info=True) return False, f"Error removing Caddy subdomain: {str(e)}" # ===================================================================== # Attribute & Tag Parser # ===================================================================== def parse_tag_attributes(attrs_str: str) -> Dict[str, str]: """ Parses key="value", key='value', or key=value pairs from a tag attribute string. If no key=value syntax is used (e.g. single positional argument), maps to '_default'. """ attrs: Dict[str, str] = {} if not attrs_str or not attrs_str.strip(): return attrs clean_str = attrs_str.strip() # Check if this is a single quoted string e.g. "claude-sonnet-4-6" or 'my-project' if (clean_str.startswith('"') and clean_str.endswith('"')) or (clean_str.startswith("'") and clean_str.endswith("'")): attrs["_default"] = clean_str[1:-1].strip() return attrs # Check if single unquoted word if "=" not in clean_str and " " not in clean_str: attrs["_default"] = clean_str return attrs kv_pairs = re.findall(r'(\w+)\s*=\s*(?:"([^"]*)"|\'([^\']*)\'|([^\s,\]]+))', clean_str) for k, v1, v2, v3 in kv_pairs: key = k.lower().strip() val = v1 if v1 != "" else (v2 if v2 != "" else v3) attrs[key] = val.strip() return attrs # ===================================================================== # AI Dynamic Button Storage & Helpers # ===================================================================== AI_BUTTON_PAYLOADS: Dict[str, Dict[str, Any]] = {} def register_ai_button_payload(btn_type: str, value: str, chat_id: int, project: str) -> str: """Registers an AI action payload and returns a compact 8-char ID for Telegram callback_data.""" now = time.time() # Clean up old payloads (older than 24h) if len(AI_BUTTON_PAYLOADS) > 2000: expired = [k for k, v in AI_BUTTON_PAYLOADS.items() if now - v.get("time", 0) > 86400] for k in expired: AI_BUTTON_PAYLOADS.pop(k, None) btn_id = uuid.uuid4().hex[:8] AI_BUTTON_PAYLOADS[btn_id] = { "type": btn_type, "value": value, "chat_id": chat_id, "project": project, "time": now, } return btn_id def get_ai_button_payload(btn_id: str) -> Optional[Dict[str, Any]]: """Retrieves registered button payload data by its ID.""" return AI_BUTTON_PAYLOADS.get(btn_id) def create_inline_button_from_spec( spec: Dict[str, Any], chat_id: int, project_name: str, ) -> Optional[InlineKeyboardButton]: """Creates a Telegram InlineKeyboardButton from a button specification dictionary.""" text = spec.get("text") or spec.get("title") or spec.get("label") or spec.get("name") or spec.get("_default", "") text = str(text).strip() if not text: return None # 1. URL button url = spec.get("url") or spec.get("link") or spec.get("href") if url: url_str = str(url).strip() if not (url_str.startswith("http://") or url_str.startswith("https://") or url_str.startswith("tg://")): url_str = f"https://{url_str}" return InlineKeyboardButton(text=text, url=url_str) # 2. Web App button web_app = spec.get("web_app") or spec.get("webapp") if web_app: return InlineKeyboardButton(text=text, web_app=WebAppInfo(url=str(web_app).strip())) # 3. Prompt button (Send prompt back to AI agent) prompt = spec.get("prompt") or spec.get("ask") or spec.get("query") if prompt: btn_id = register_ai_button_payload("prompt", str(prompt).strip(), chat_id, project_name) return InlineKeyboardButton(text=text, callback_data=f"aibtn:{btn_id}") # 4. Command button (Run slash command or bash) cmd = spec.get("cmd") or spec.get("command") or spec.get("run") if cmd: btn_id = register_ai_button_payload("cmd", str(cmd).strip(), chat_id, project_name) return InlineKeyboardButton(text=text, callback_data=f"aibtn:{btn_id}") # 5. Callback / action data cb_data = spec.get("data") or spec.get("callback") or spec.get("callback_data") or spec.get("action") if cb_data: cb_str = str(cb_data).strip() if len(cb_str.encode("utf-8")) <= 60 and not cb_str.startswith("aibtn:"): return InlineKeyboardButton(text=text, callback_data=cb_str) else: btn_id = register_ai_button_payload("action", cb_str, chat_id, project_name) return InlineKeyboardButton(text=text, callback_data=f"aibtn:{btn_id}") # Default fallback: Treat text as prompt btn_id = register_ai_button_payload("prompt", text, chat_id, project_name) return InlineKeyboardButton(text=text, callback_data=f"aibtn:{btn_id}") def _extract_json_from_text(raw_text: str) -> Optional[Any]: """Attempts to parse JSON list/dict from a raw string with high fault tolerance.""" raw = raw_text.strip() # Check if starts with layout= or buttons= m_kv = re.search(r'(?:layout|buttons|keyboard)\s*=\s*([\[\{][\s\S]*[\]\}])', raw, re.IGNORECASE) if m_kv: raw = m_kv.group(1).strip() elif not (raw.startswith("[") or raw.startswith("{")): # Try finding [ ... ] block m_arr = re.search(r'(\[[\s\S]*\])', raw) if m_arr: raw = m_arr.group(1).strip() # Attempt standard JSON parse try: return json.loads(raw) except Exception: pass # Attempt cleanup for Python style dicts / single quotes try: cleaned = raw.replace("'", '"') cleaned = re.sub(r',\s*([\]\}])', r'\1', cleaned) # Remove trailing commas return json.loads(cleaned) except Exception: pass return None def extract_action_tags(text: str) -> List[Tuple[str, str, str]]: """ Extracts all [[ACTION_NAME: ...]] or [[ACTION_NAME]] tags with bracket-depth balancing. Returns list of (full_tag_string, action_name, raw_attributes_string). """ tags = [] i = 0 n = len(text) while i < n - 1: if text[i:i+2] == "[[" and (i == 0 or text[i-1] != "\\"): start = i # Find action name (starts right after [[) j = i + 2 while j < n and text[j] not in (":", "]", " ", "\n", "\t"): j += 1 action_name = text[i+2:j].strip() # Start tracking inside tag k = j if k < n and text[k] == ":": k += 1 attrs_start = k depth = 1 # 1 for outer [[ ... ]] while k < n: if k < n - 1 and text[k:k+2] == "]]" and depth == 1: full_tag = text[start:k+2] attrs_raw = text[attrs_start:k] tags.append((full_tag, action_name, attrs_raw)) i = k + 2 break elif text[k] == "[": depth += 1 k += 1 elif text[k] == "]": depth -= 1 k += 1 else: k += 1 else: i += 1 else: i += 1 return tags def parse_buttons_from_text( raw_text: str, chat_id: int, project_name: str, ) -> Tuple[str, Optional[InlineKeyboardMarkup]]: """ Finds and extracts button tags ([[BUTTON:...]], [[BUTTONS:...]], [[INLINE_BUTTONS:...]], [[KEYBOARD:...]]) from the raw AI response text, builds the InlineKeyboardMarkup, and cleans the text. """ if not raw_text or "[[" not in raw_text: return raw_text, None cleaned_text = raw_text row_map: Dict[int, List[InlineKeyboardButton]] = {} unnumbered_buttons: List[InlineKeyboardButton] = [] matrix_rows: List[List[InlineKeyboardButton]] = [] tags = extract_action_tags(raw_text) if not tags: return raw_text, None for full_tag, tag_name, attrs_raw in tags: tag_type = tag_name.upper().strip() if tag_type not in ("BUTTON", "BUTTONS", "INLINE_BUTTONS", "KEYBOARD", "INLINE_KEYBOARD"): continue # Remove the tag from output text cleaned_text = cleaned_text.replace(full_tag, "") attrs_raw = attrs_raw.strip() if tag_type in ("BUTTONS", "INLINE_BUTTONS", "KEYBOARD", "INLINE_KEYBOARD"): parsed_json = _extract_json_from_text(attrs_raw) if isinstance(parsed_json, list): # Could be 2D array or 1D array if parsed_json and isinstance(parsed_json[0], list): # 2D layout: [[btn1, btn2], [btn3]] for r in parsed_json: row_items = [] if isinstance(r, list): for b_spec in r: if isinstance(b_spec, dict): btn = create_inline_button_from_spec(b_spec, chat_id, project_name) if btn: row_items.append(btn) if row_items: matrix_rows.append(row_items) else: # 1D layout: [btn1, btn2, btn3, ...] -> chunk into rows of 2 cur_row = [] for b_spec in parsed_json: if isinstance(b_spec, dict): btn = create_inline_button_from_spec(b_spec, chat_id, project_name) if btn: cur_row.append(btn) if len(cur_row) >= 2: matrix_rows.append(cur_row) cur_row = [] if cur_row: matrix_rows.append(cur_row) else: # Try fallback key-value parsing attrs = parse_tag_attributes(attrs_raw) btn = create_inline_button_from_spec(attrs, chat_id, project_name) if btn: unnumbered_buttons.append(btn) elif tag_type == "BUTTON": attrs = parse_tag_attributes(attrs_raw) row_idx = None if "row" in attrs: try: row_idx = int(attrs["row"]) except Exception: pass elif "line" in attrs: try: row_idx = int(attrs["line"]) except Exception: pass btn = create_inline_button_from_spec(attrs, chat_id, project_name) if btn: if row_idx is not None and row_idx > 0: if row_idx not in row_map: row_map[row_idx] = [] row_map[row_idx].append(btn) else: unnumbered_buttons.append(btn) # Assemble all buttons into final keyboard final_rows: List[List[InlineKeyboardButton]] = [] # 1. Add matrix rows first if matrix_rows: final_rows.extend(matrix_rows) # 2. Add numbered rows if row_map: for r_num in sorted(row_map.keys()): final_rows.append(row_map[r_num]) # 3. Add unnumbered buttons (chunked into pairs) if unnumbered_buttons: cur_row = [] for btn in unnumbered_buttons: cur_row.append(btn) if len(cur_row) >= 2: final_rows.append(cur_row) cur_row = [] if cur_row: final_rows.append(cur_row) # Clean up empty lines from text cleaned_text = re.sub(r'\n{3,}', '\n\n', cleaned_text).strip() if not final_rows: return cleaned_text, None return cleaned_text, InlineKeyboardMarkup(final_rows) # ===================================================================== # Core Action Processor # ===================================================================== ACTION_REGEX = re.compile( r"\[\[([A-Z0-9_]+)(?::\s*([\s\S]*?))?\]\]", re.MULTILINE, ) async def execute_action( action_name: str, attrs: Dict[str, str], chat_id: int, current_project_name: str, is_fa: bool, app: Any = None, ) -> Tuple[str, List[Callable], Optional[ScheduledTask]]: """ Executes a single bot action based on action_name and attributes. Returns (rendered_html_badge, list_of_async_side_effects, created_task_or_none). """ session = session_manager.get_or_create(chat_id) is_admin = settings.is_admin(chat_id) side_effects: List[Callable] = [] created_task: Optional[ScheduledTask] = None act = action_name.upper().strip() # ------------------------------------------------------------- # 1. SET_MODEL / CHANGE_MODEL / MODEL # ------------------------------------------------------------- if act in ("SET_MODEL", "CHANGE_MODEL", "MODEL", "SWITCH_MODEL"): raw_m = attrs.get("model") or attrs.get("name") or attrs.get("target") or attrs.get("_default", "") req_p = attrs.get("project") or attrs.get("proj") resolved_m = resolve_model_name(raw_m) if not resolved_m: err = ( f"\n⚠️ خطا در تغییر مدل: مدل «{escape_html(raw_m)}» معتبر نیست.\n" f"مدل‌های معتبر: {', '.join(AVAILABLE_MODELS.keys())}" if is_fa else f"\n⚠️ Model Change Failed: Model '{escape_html(raw_m)}' is invalid.\n" f"Valid models: {', '.join(AVAILABLE_MODELS.keys())}" ) return err, side_effects, created_task # Find target project dynamically target_proj = None if req_p: accessible = session_manager.get_all_accessible_projects(chat_id) for k, p in accessible.items(): if k.lower() == req_p.lower() or p.name.lower() == req_p.lower(): target_proj = p break if not target_proj: target_proj = session_manager.get_current_project(chat_id) if not target_proj: return "\n⚠️ پروژه‌ای برای تنظیم مدل یافت نشد." if is_fa else "\n⚠️ No active project found to set model.", side_effects, created_task target_proj.model = resolved_m # Check effort compatibility valid_effort = get_valid_effort_for_model(resolved_m, target_proj.effort) if valid_effort: target_proj.effort = valid_effort session_manager.save() model_display = AVAILABLE_MODELS.get(resolved_m, resolved_m) badge = ( f"\n\n🧠 مدل هوش مصنوعی تغییر یافت:\n" f"• 📁 پروژه: {escape_html(target_proj.name)}\n" f"• ⚡ مدل جدید: {escape_html(model_display)} ({resolved_m})" if is_fa else f"\n\n🧠 AI Model Switched:\n" f"• 📁 Project: {escape_html(target_proj.name)}\n" f"• ⚡ New Model: {escape_html(model_display)} ({resolved_m})" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 2. SET_EFFORT / EFFORT # ------------------------------------------------------------- elif act in ("SET_EFFORT", "EFFORT", "REASONING_EFFORT"): raw_eff = attrs.get("effort") or attrs.get("level") or attrs.get("_default", "") req_p = attrs.get("project") or attrs.get("proj") resolved_eff = resolve_effort_level(raw_eff) if not resolved_eff: return f"\n⚠️ سطح فکر «{raw_eff}» نامعتبر است (گزینه‌ها: low, medium, high)." if is_fa else f"\n⚠️ Invalid effort '{raw_eff}' (options: low, medium, high).", side_effects, created_task target_proj = None if req_p: accessible = session_manager.get_all_accessible_projects(chat_id) for k, p in accessible.items(): if k.lower() == req_p.lower() or p.name.lower() == req_p.lower(): target_proj = p break if not target_proj: target_proj = session_manager.get_current_project(chat_id) if not target_proj: return "\n⚠️ پروژه‌ای یافت نشد." if is_fa else "\n⚠️ No project found.", side_effects, created_task target_proj.effort = resolved_eff session_manager.save() badge = ( f"\n\n⚡ سطح تفکر مدل (Effort) تنظیم شد:\n" f"• 📁 پروژه: {escape_html(target_proj.name)}\n" f"• 🎯 میزان تفکر: {resolved_eff.upper()}" if is_fa else f"\n\n⚡ Reasoning Effort Updated:\n" f"• 📁 Project: {escape_html(target_proj.name)}\n" f"• 🎯 Effort Level: {resolved_eff.upper()}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 3. SET_LANG / SET_LANGUAGE / LANGUAGE # ------------------------------------------------------------- elif act in ("SET_LANG", "SET_LANGUAGE", "LANGUAGE", "LANG"): raw_l = attrs.get("lang") or attrs.get("language") or attrs.get("code") or attrs.get("_default", "") resolved_l = resolve_language_code(raw_l) if not resolved_l: return f"\n⚠️ کد زبان «{raw_l}» نامعتبر است." if is_fa else f"\n⚠️ Invalid language code '{raw_l}'.", side_effects, created_task await session_manager.set_language(chat_id, resolved_l) lang_display = AVAILABLE_LANGUAGES.get(resolved_l, resolved_l) badge = ( f"\n\n🌐 زبان پاسخ‌دهی تغییر یافت:\n" f"• 🗣️ زبان جدید: {escape_html(lang_display)}" if is_fa else f"\n\n🌐 Response Language Changed:\n" f"• 🗣️ New Language: {escape_html(lang_display)}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 4. NEW_PROJECT / CREATE_PROJECT # ------------------------------------------------------------- elif act in ("NEW_PROJECT", "CREATE_PROJECT", "ADD_PROJECT"): p_name = attrs.get("name") or attrs.get("project") or attrs.get("_default", "") if not p_name: return "\n⚠️ نام پروژه مشخص نشده است." if is_fa else "\n⚠️ Project name not specified.", side_effects, created_task ws = attrs.get("workspace") or attrs.get("path") req_model = resolve_model_name(attrs.get("model", "")) req_effort = resolve_effort_level(attrs.get("effort", "")) desc = attrs.get("description", "") proj = await session_manager.create_project( chat_id=chat_id, name=p_name, workspace=ws, model=req_model, effort=req_effort, description=desc, ) badge = ( f"\n\n📁 پروژه جدید با موفقیت ایجاد و فعال شد:\n" f"• 🏷️ نام پروژه: {escape_html(proj.name)}\n" f"• 📂 مسیر دایرکتوری: {escape_html(proj.workspace)}\n" f"• 🧠 مدل: {proj.model} | ⚡ تفکر: {proj.effort}" if is_fa else f"\n\n📁 New Project Created & Activated:\n" f"• 🏷️ Project Name: {escape_html(proj.name)}\n" f"• 📂 Workspace: {escape_html(proj.workspace)}\n" f"• 🧠 Model: {proj.model} | ⚡ Effort: {proj.effort}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 5. SWITCH_PROJECT / SELECT_PROJECT / USE_PROJECT # ------------------------------------------------------------- elif act in ("SWITCH_PROJECT", "SELECT_PROJECT", "USE_PROJECT", "CHANGE_PROJECT"): target_name = attrs.get("name") or attrs.get("project") or attrs.get("_default", "") if not target_name: return "\n⚠️ نام پروژه برای سوییچ مشخص نشده است." if is_fa else "\n⚠️ Project name for switch not specified.", side_effects, created_task proj = await session_manager.switch_project(chat_id, target_name) if not proj: return f"\n⚠️ پروژه با نام «{escape_html(target_name)}» یافت نشد." if is_fa else f"\n⚠️ Project '{escape_html(target_name)}' was not found.", side_effects, created_task badge = ( f"\n\n🔄 پروژه فعال با موفقیت تغییر کرد:\n" f"• 📁 پروژه جدید: {escape_html(proj.name)}\n" f"• 📂 مسیر دایرکتوری: {escape_html(proj.workspace)}\n" f"• 🧠 مدل: {proj.model}" if is_fa else f"\n\n🔄 Active Project Switched:\n" f"• 📁 Current Project: {escape_html(proj.name)}\n" f"• 📂 Workspace: {escape_html(proj.workspace)}\n" f"• 🧠 Model: {proj.model}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 6. DELETE_PROJECT / DEL_PROJECT # ------------------------------------------------------------- elif act in ("DELETE_PROJECT", "DEL_PROJECT", "REMOVE_PROJECT"): target_name = attrs.get("name") or attrs.get("project") or attrs.get("_default", "") delete_files_str = attrs.get("delete_files", "false").lower() del_files = delete_files_str in ("true", "1", "yes", "بله") if not target_name: return "\n⚠️ نام پروژه برای حذف مشخص نشده است." if is_fa else "\n⚠️ Project name not specified for deletion.", side_effects, created_task # Get workspace before deleting if files should be deleted proj_to_delete = session.projects.get(target_name) ws_to_clean = proj_to_delete.workspace if (proj_to_delete and del_files) else None ok, msg = await session_manager.delete_project(chat_id, target_name) if not ok: return f"\n⚠️ {msg}", side_effects, created_task if ws_to_clean and str(ws_to_clean).startswith("/root/projects"): try: shutil.rmtree(ws_to_clean, ignore_errors=True) msg += f"\n🗑️ پوشه فایل‌ها ({escape_html(ws_to_clean)}) نیز پاکسازی شد." if is_fa else f"\n🗑️ Files directory ({escape_html(ws_to_clean)}) deleted." except Exception as e: logger.warning(f"Error removing project files: {e}") badge = ( f"\n\n🗑️ عملیات حذف پروژه:\n{msg}" if is_fa else f"\n\n🗑️ Project Deletion:\n{msg}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 7. RENAME_PROJECT # ------------------------------------------------------------- elif act in ("RENAME_PROJECT", "RENAME"): old_n = attrs.get("old_name") or attrs.get("old") new_n = attrs.get("new_name") or attrs.get("new") or attrs.get("name") if not old_n or not new_n: return "\n⚠️ لطفاً نام قدیم و جدید پروژه را مشخص کنید." if is_fa else "\n⚠️ Please specify both old and new project names.", side_effects, created_task proj, msg = await session_manager.rename_project(chat_id, old_n, new_n) return f"\n\n✏️ {msg}", side_effects, created_task # ------------------------------------------------------------- # 8. SET_WORKSPACE / WORKSPACE # ------------------------------------------------------------- elif act in ("SET_WORKSPACE", "WORKSPACE", "CHANGE_WORKSPACE"): ws_path = attrs.get("path") or attrs.get("workspace") or attrs.get("dir") or attrs.get("_default", "") req_p = attrs.get("project") or attrs.get("proj") if not ws_path: return "\n⚠️ مسیر دایرکتوری مشخص نشده است." if is_fa else "\n⚠️ Workspace path not specified.", side_effects, created_task resolved_ws = str(Path(ws_path).expanduser().resolve()) os.makedirs(resolved_ws, exist_ok=True) target_proj = None if req_p: accessible = session_manager.get_all_accessible_projects(chat_id) for k, p in accessible.items(): if k.lower() == req_p.lower() or p.name.lower() == req_p.lower(): target_proj = p break if not target_proj: target_proj = session_manager.get_current_project(chat_id) if target_proj: target_proj.workspace = resolved_ws session_manager.save() badge = ( f"\n\n📂 مسیر کاری (Workspace) بروزرسانی شد:\n" f"• 📁 پروژه: {escape_html(target_proj.name if target_proj else current_project_name)}\n" f"• 📍 مسیر جدید: {escape_html(resolved_ws)}" if is_fa else f"\n\n📂 Workspace Updated:\n" f"• 📁 Project: {escape_html(target_proj.name if target_proj else current_project_name)}\n" f"• 📍 New Path: {escape_html(resolved_ws)}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 9. SHARE_PROJECT / UNSHARE_PROJECT # ------------------------------------------------------------- elif act in ("SHARE_PROJECT", "SHARE"): uid_str = attrs.get("user_id") or attrs.get("user") or attrs.get("target_id") or attrs.get("_default", "") target_p = attrs.get("project") try: target_uid = int(uid_str) ok, msg, _ = await session_manager.share_project(chat_id, target_uid, target_p) return f"\n\n👥 {msg}", side_effects, created_task except ValueError: return f"\n⚠️ شناسه کاربری «{uid_str}» نامعتبر است." if is_fa else f"\n⚠️ Invalid user ID '{uid_str}'.", side_effects, created_task elif act in ("UNSHARE_PROJECT", "UNSHARE", "REVOKE_SHARE"): uid_str = attrs.get("user_id") or attrs.get("user") or attrs.get("target_id") or attrs.get("_default", "") target_p = attrs.get("project") try: target_uid = int(uid_str) ok, msg, _ = await session_manager.unshare_project(chat_id, target_uid, target_p) return f"\n\n👥 {msg}", side_effects, created_task except ValueError: return f"\n⚠️ شناسه کاربری «{uid_str}» نامعتبر است." if is_fa else f"\n⚠️ Invalid user ID '{uid_str}'.", side_effects, created_task # ------------------------------------------------------------- # 10. RESET_CONVERSATION / NEW_CONV / NEW_CHAT / CLEAR_CONTEXT # ------------------------------------------------------------- elif act in ("RESET_CONVERSATION", "NEW_CONV", "NEW_CHAT", "CLEAR_CONTEXT", "RESET_CHAT"): p_name = attrs.get("project") await session_manager.reset_session(chat_id, p_name) badge = ( "\n\n🧹 نشست گفتگو با موفقیت پاکسازی شد و گفتگوی جدیدی آغاز گردید." if is_fa else "\n\n🧹 Conversation session reset. Starting a fresh new chat context." ) return badge, side_effects, created_task # ------------------------------------------------------------- # 11. SWITCH_CONVERSATION / SWITCH_CHAT # ------------------------------------------------------------- elif act in ("SWITCH_CONVERSATION", "SWITCH_CHAT", "OPEN_CONV"): conv_id = attrs.get("id") or attrs.get("conv_id") or attrs.get("_default", "") if not conv_id: ok, msg, cid, _ = session_manager.reopen_last_conversation(chat_id) return f"\n\n💬 {msg}", side_effects, created_task else: ok, msg, _ = session_manager.switch_conversation(chat_id, conv_id) return f"\n\n💬 {msg}", side_effects, created_task # ------------------------------------------------------------- # 12. DELETE_CONVERSATION / DEL_CONV # ------------------------------------------------------------- elif act in ("DELETE_CONVERSATION", "DEL_CONV", "DELETE_CHAT"): conv_id = attrs.get("id") or attrs.get("conv_id") or attrs.get("_default", "") if not conv_id: return "\n⚠️ شناسه گفتگو مشخص نشده است." if is_fa else "\n⚠️ Conversation ID not specified.", side_effects, created_task ok, msg = session_manager.delete_conversation(chat_id, conv_id) return f"\n\n🗑️ {msg}", side_effects, created_task # ------------------------------------------------------------- # 13. CLEAR_CONVERSATIONS / CLEAR_CHATS # ------------------------------------------------------------- elif act in ("CLEAR_CONVERSATIONS", "CLEAR_CHATS", "DELETE_ALL_CONVS"): ok, msg, count = session_manager.clear_project_conversations(chat_id) return f"\n\n🗑️ {msg}", side_effects, created_task # ------------------------------------------------------------- # 14. SCHEDULE_TASK / SCHEDULE / CREATE_TASK # ------------------------------------------------------------- elif act in ("SCHEDULE_TASK", "SCHEDULE", "CREATE_TASK", "ADD_TASK"): timing = attrs.get("timing") or attrs.get("time") or attrs.get("cron") or "" content = attrs.get("content") or attrs.get("prompt") or attrs.get("cmd") or attrs.get("command") or "" title = attrs.get("title") or "" task_type = (attrs.get("type") or "prompt").lower() req_p = attrs.get("project") or attrs.get("proj") if req_p: target_p = req_p else: curr = session_manager.get_current_project(chat_id) target_p = curr.name if curr else current_project_name max_runs = None runs_val = attrs.get("max_runs") or attrs.get("repeat") or attrs.get("count") or attrs.get("times") if runs_val and runs_val.lower() not in ("none", "null", "unlimited", "0"): try: max_runs = int(runs_val) except ValueError: pass if not timing or not content: return "\n⚠️ برای زمان‌بندی تسک، مشخص کردن timing و content الزامی است." if is_fa else "\n⚠️ Timing and content are required for task scheduling.", side_effects, created_task try: task = task_scheduler.add_task( chat_id=chat_id, creator_id=chat_id, project_name=target_p, task_type=task_type, content=content, timing_spec=timing, title=title, max_runs=max_runs, ) created_task = task type_name = "پرامپت هوش مصنوعی" if task_type == "prompt" else "دستور شل" if task_type == "command" else "یادآور" next_rel = format_relative_time(task.next_run_timestamp - time.time(), is_fa=is_fa) repeat_str = f"{task.max_runs} بار" if task.max_runs else "نامحدود (دائمی)" badge = ( f"\n\n⏰ تسک زمان‌بندی شده توسط هوش مصنوعی ثبت شد:\n" f"• 🏷️ شناسه تسک: {task.id}\n" f"• ⏱️ زمان‌بندی: {escape_html(task.timing_spec)}\n" f"• 🔄 حداکثر تکرار: {repeat_str}\n" f"• 🛠 نوع: {type_name}\n" f"• ⏳ اولین اجرا: {next_rel} ({format_timestamp(task.next_run_timestamp, is_fa=True)})\n" f"• 📝 محتوا: {escape_html(task.content[:120])}" if is_fa else f"\n\n⏰ Scheduled Task Registered:\n" f"• 🏷️ Task ID: {task.id}\n" f"• ⏱️ Timing: {escape_html(task.timing_spec)}\n" f"• 🔄 Max Runs: {task.max_runs if task.max_runs else 'Unlimited'}\n" f"• 🛠 Type: {type_name}\n" f"• ⏳ First Run: {next_rel} ({format_timestamp(task.next_run_timestamp, is_fa=False)})\n" f"• 📝 Content: {escape_html(task.content[:120])}" ) return badge, side_effects, created_task except Exception as e: return f"\n⚠️ خطا در ثبت تسک زمان‌بندی: {escape_html(str(e))}", side_effects, created_task # ------------------------------------------------------------- # 15. CANCEL_TASK / DELETE_TASK / PAUSE_TASK / RESUME_TASK # ------------------------------------------------------------- elif act in ("CANCEL_TASK", "DELETE_TASK", "REMOVE_TASK"): t_id = attrs.get("id") or attrs.get("task_id") or attrs.get("_default", "") if not t_id: return "\n⚠️ شناسه تسک مشخص نشده است." if is_fa else "\n⚠️ Task ID not specified.", side_effects, created_task deleted = False user_tasks = task_scheduler.get_user_tasks(chat_id) for t in user_tasks: if t.id == t_id or t.id.startswith(t_id) or (t.title and t.title.lower() == t_id.lower()): task_scheduler.delete_task(t.id) deleted = True break if deleted: return f"\n\n🗑️ تسک زمان‌بندی {escape_html(t_id)} با موفقیت لغو و حذف شد." if is_fa else f"\n\n🗑️ Scheduled task {escape_html(t_id)} was deleted.", side_effects, created_task else: return f"\n⚠️ تسک {escape_html(t_id)} یافت نشد." if is_fa else f"\n⚠️ Task {escape_html(t_id)} not found.", side_effects, created_task elif act in ("PAUSE_TASK", "STOP_TASK"): t_id = attrs.get("id") or attrs.get("task_id") or attrs.get("_default", "") ok = task_scheduler.pause_task(t_id) if ok: return f"\n\n⏸️ تسک زمان‌بندی {escape_html(t_id)} موقتاً متوقف (Pause) شد." if is_fa else f"\n\n⏸️ Task {escape_html(t_id)} paused.", side_effects, created_task return f"\n⚠️ امکان توقف تسک {escape_html(t_id)} وجود ندارد." if is_fa else f"\n⚠️ Could not pause task {escape_html(t_id)}.", side_effects, created_task elif act in ("RESUME_TASK", "START_TASK"): t_id = attrs.get("id") or attrs.get("task_id") or attrs.get("_default", "") ok = task_scheduler.resume_task(t_id) if ok: return f"\n\n▶️ تسک زمان‌بندی {escape_html(t_id)} مجدداً فعال شد." if is_fa else f"\n\n▶️ Task {escape_html(t_id)} resumed.", side_effects, created_task return f"\n⚠️ امکان فعال‌سازی تسک {escape_html(t_id)} وجود ندارد." if is_fa else f"\n⚠️ Could not resume task {escape_html(t_id)}.", side_effects, created_task # ------------------------------------------------------------- # 16. BACKUP_PROJECT / EXPORT_PROJECT # ------------------------------------------------------------- elif act in ("BACKUP_PROJECT", "EXPORT_PROJECT", "ZIP_PROJECT"): req_p = attrs.get("project") or attrs.get("proj") proj = None if req_p: accessible = session_manager.get_all_accessible_projects(chat_id) for k, p in accessible.items(): if k.lower() == req_p.lower() or p.name.lower() == req_p.lower(): proj = p break if not proj: proj = session_manager.get_current_project(chat_id) if not proj: return "\n⚠️ پروژه‌ای برای پشتیبان‌گیری یافت نشد." if is_fa else "\n⚠️ No project found for backup.", side_effects, created_task async def do_backup(): if app: from bot import handle_project_backup_flow await handle_project_backup_flow( application=app, chat_id=chat_id, proj_name=proj.name, workspace=proj.workspace, is_fa=is_fa, status_msg=None, ) side_effects.append(do_backup) badge = ( f"\n\n📦 درخواست پشتیبان‌گیری از پروژه «{escape_html(proj.name)}» ثبت شد و فایل بکاپ برای شما ارسال می‌گردد." if is_fa else f"\n\n📦 Backup requested for project '{escape_html(proj.name)}'. Zip file is being generated and sent." ) return badge, side_effects, created_task # ------------------------------------------------------------- # 17. SEND_FILE / SEND_DOCUMENT # ------------------------------------------------------------- elif act in ("SEND_FILE", "SEND_DOCUMENT", "ATTACH_FILE"): f_path = attrs.get("path") or attrs.get("file") or attrs.get("file_path") or attrs.get("_default", "") caption = attrs.get("caption") or "" resolved_p = Path(f_path).expanduser().resolve() if not resolved_p.exists() or not resolved_p.is_file(): return f"\n⚠️ فایل {escape_html(f_path)} در سرور یافت نشد." if is_fa else f"\n⚠️ File {escape_html(f_path)} not found.", side_effects, created_task async def send_doc(): if app and app.bot: from telegram import constants try: with open(resolved_p, "rb") as df: await app.bot.send_document( chat_id=chat_id, document=df, filename=resolved_p.name, caption=caption or f"📄 {resolved_p.name}", parse_mode=constants.ParseMode.HTML if caption else None, read_timeout=60, write_timeout=60, ) except Exception as ex: logger.error(f"Failed to send file {resolved_p}: {ex}") side_effects.append(send_doc) badge = ( f"\n\n📄 فایل {escape_html(resolved_p.name)} پیوست و ارسال گردید." if is_fa else f"\n\n📄 File {escape_html(resolved_p.name)} attached and sent." ) return badge, side_effects, created_task # ------------------------------------------------------------- # 18. SEND_PHOTO / SEND_IMAGE # ------------------------------------------------------------- elif act in ("SEND_PHOTO", "SEND_IMAGE"): img_path = attrs.get("path") or attrs.get("file") or attrs.get("_default", "") caption = attrs.get("caption") or "" resolved_p = Path(img_path).expanduser().resolve() if not resolved_p.exists() or not resolved_p.is_file(): return f"\n⚠️ تصویر {escape_html(img_path)} یافت نشد." if is_fa else f"\n⚠️ Image {escape_html(img_path)} not found.", side_effects, created_task async def send_img(): if app and app.bot: from telegram import constants try: with open(resolved_p, "rb") as pf: await app.bot.send_photo( chat_id=chat_id, photo=pf, caption=caption or None, parse_mode=constants.ParseMode.HTML if caption else None, ) except Exception as ex: logger.error(f"Failed to send photo {resolved_p}: {ex}") side_effects.append(send_img) badge = ( f"\n\n🖼️ تصویر {escape_html(resolved_p.name)} ارسال گردید." if is_fa else f"\n\n🖼️ Photo {escape_html(resolved_p.name)} sent." ) return badge, side_effects, created_task # ------------------------------------------------------------- # 19. SEND_AUDIO / SEND_VOICE # ------------------------------------------------------------- elif act in ("SEND_AUDIO", "SEND_VOICE"): a_path = attrs.get("path") or attrs.get("file") or attrs.get("_default", "") caption = attrs.get("caption") or "" resolved_p = Path(a_path).expanduser().resolve() if not resolved_p.exists() or not resolved_p.is_file(): return f"\n⚠️ فایل صوتی {escape_html(a_path)} یافت نشد." if is_fa else f"\n⚠️ Audio file {escape_html(a_path)} not found.", side_effects, created_task async def send_aud(): if app and app.bot: try: with open(resolved_p, "rb") as af: if act == "SEND_VOICE": await app.bot.send_voice(chat_id=chat_id, voice=af, caption=caption or None) else: await app.bot.send_audio(chat_id=chat_id, audio=af, caption=caption or None) except Exception as ex: logger.error(f"Failed to send audio {resolved_p}: {ex}") side_effects.append(send_aud) return f"\n\n🎵 فایل صوتی {escape_html(resolved_p.name)} ارسال شد.", side_effects, created_task # ------------------------------------------------------------- # 20. SETUP_SUBDOMAIN / ADD_SUBDOMAIN / CADDY_PROXY # ------------------------------------------------------------- elif act in ("SETUP_SUBDOMAIN", "ADD_SUBDOMAIN", "CADDY_PROXY", "EXPOSE_PORT"): sub = attrs.get("subdomain") or attrs.get("name") or attrs.get("host") or "" port = attrs.get("port") or attrs.get("target") or attrs.get("_default", "") custom_domain = attrs.get("custom_domain") if not sub or not port: return "\n⚠️ برای تنظیم ساب‌دامین، مشخص کردن subdomain و port الزامی است." if is_fa else "\n⚠️ Subdomain and port are required for proxy setup.", side_effects, created_task ok, msg, full_url = configure_caddy_reverse_proxy( subdomain=sub, port_or_target=port, custom_domain=custom_domain, ) if ok: badge = ( f"\n\n🌐 ساب‌دامین و Reverse Proxy با موفقیت راه‌اندازی شد:\n" f"• 🔗 آدرس وب‌سایت: {full_url}\n" f"• 🔌 متصل به پورت داخلی: {escape_html(str(port))}\n" f"• 🔒 گواهی امنیتی SSL (HTTPS): فعال و خودکار (Caddy)" if is_fa else f"\n\n🌐 Subdomain & Reverse Proxy Configured:\n" f"• 🔗 Website URL: {full_url}\n" f"• 🔌 Proxied Port: {escape_html(str(port))}\n" f"• 🔒 SSL Certificate (HTTPS): Active & Automatic (Caddy)" ) return badge, side_effects, created_task else: return f"\n⚠️ خطا در پیکربندی Caddy: {escape_html(msg)}", side_effects, created_task # ------------------------------------------------------------- # 21. REMOVE_SUBDOMAIN / DEL_SUBDOMAIN # ------------------------------------------------------------- elif act in ("REMOVE_SUBDOMAIN", "DEL_SUBDOMAIN", "DELETE_SUBDOMAIN"): sub = attrs.get("subdomain") or attrs.get("name") or attrs.get("_default", "") if not sub: return "\n⚠️ ساب‌دامین برای حذف مشخص نشده است." if is_fa else "\n⚠️ Subdomain not specified.", side_effects, created_task ok, msg = remove_caddy_reverse_proxy(sub) return f"\n\n🌐 {msg}", side_effects, created_task # ------------------------------------------------------------- # 22. INVITE_USER / ADD_USER (Admin Only) # ------------------------------------------------------------- elif act in ("INVITE_USER", "ADD_USER", "WHITELIST_USER"): if not is_admin: return "\n⚠️ فقط مدیران دسترسی به افزودن کاربر دارند." if is_fa else "\n⚠️ Only admins can whitelist users.", side_effects, created_task uid_str = attrs.get("user_id") or attrs.get("user") or attrs.get("_default", "") note = attrs.get("note", "") try: target_uid = int(uid_str) settings.add_authorized_user(target_uid) invite_manager.approve_request(target_uid) return f"\n\n✅ کاربر {target_uid} با موفقیت به لیست مجاز افزوده شد.", side_effects, created_task except ValueError: return f"\n⚠️ شناسه کاربری «{uid_str}» نامعتبر است.", side_effects, created_task # ------------------------------------------------------------- # 23. REMOVE_USER / BAN_USER (Admin Only) # ------------------------------------------------------------- elif act in ("REMOVE_USER", "BAN_USER", "REVOKE_USER"): if not is_admin: return "\n⚠️ فقط مدیران دسترسی به حذف کاربر دارند." if is_fa else "\n⚠️ Only admins can remove users.", side_effects, created_task uid_str = attrs.get("user_id") or attrs.get("user") or attrs.get("_default", "") try: target_uid = int(uid_str) settings.remove_authorized_user(target_uid) return f"\n\n🚫 دسترسی کاربر {target_uid} با موفقیت لغو شد.", side_effects, created_task except ValueError: return f"\n⚠️ شناسه کاربری «{uid_str}» نامعتبر است.", side_effects, created_task # ------------------------------------------------------------- # 24. RESTART_BOT / RELOAD_BOT # ------------------------------------------------------------- elif act in ("RESTART_BOT", "RELOAD_BOT"): async def do_restart(): start_w = time.time() while time.time() - start_w < 180.0: in_prog = any(s.turn_in_progress for s in session_manager.sessions.values()) if not in_prog and not session_manager.active_tasks and not session_manager.active_procs: break await asyncio.sleep(1.5) await asyncio.sleep(2.0) os.system("systemctl restart --no-block agy-telegram-bot.service &") side_effects.append(do_restart) badge = ( "\n\n🔄 دستور راه‌اندازی مجدد هوشمند ثبت شد (پس از اتمام کامل عملیات جاری، ربات ری‌استارت می‌گردد)." if is_fa else "\n\n🔄 Graceful restart scheduled (will execute automatically after all active tasks complete)." ) return badge, side_effects, created_task # ------------------------------------------------------------- # 25. GIT_SYNC / GIT_PULL_AND_PUSH # ------------------------------------------------------------- elif act in ("GIT_SYNC", "SYNC_GIT", "GITEA_SYNC"): p_name = attrs.get("project") or attrs.get("name") or current_project_name proj = session.projects.get(p_name) if p_name else curr_proj if not proj: return f"\n⚠️ پروژه {escape_html(p_name or '')} یافت نشد." if is_fa else f"\n⚠️ Project {escape_html(p_name or '')} not found.", side_effects, created_task from git_manager import git_manager msg_commit = attrs.get("message") or f"Sync from AI Bot for {proj.name}" ok, res_msg = await git_manager.git_sync(proj.workspace, message=msg_commit, repo_name=proj.name) urls = git_manager.get_repo_urls(proj.name) badge = ( f"\n\n🔄 عملیات همگام‌سازی مخزن گیت (Git Sync) با موفقیت انجام شد:\n" f"• 📁 پروژه: {escape_html(proj.name)}\n" f"• 🌐 مخزن تحت وب: {urls['web_url']}\n" f"• 📋 گزارش عملیات:\n{res_msg}" if is_fa else f"\n\n🔄 Git Repository Synced Successfully:\n" f"• 📁 Project: {escape_html(proj.name)}\n" f"• 🌐 Web Repo: {urls['web_url']}\n" f"• 📋 Details:\n{res_msg}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 26. GIT_COMMIT / GIT_PUSH # ------------------------------------------------------------- elif act in ("GIT_COMMIT", "GIT_PUSH", "COMMIT_PROJECT"): p_name = attrs.get("project") or attrs.get("name") or current_project_name proj = session.projects.get(p_name) if p_name else curr_proj if not proj: return f"\n⚠️ پروژه {escape_html(p_name or '')} یافت نشد." if is_fa else f"\n⚠️ Project {escape_html(p_name or '')} not found.", side_effects, created_task from git_manager import git_manager commit_msg = attrs.get("message") or attrs.get("_default") or f"Updates for {proj.name}" ok, res_msg = await git_manager.git_commit_and_push(proj.workspace, message=commit_msg, repo_name=proj.name) urls = git_manager.get_repo_urls(proj.name) badge = ( f"\n\n💾 عملیات کامیت و پوش گیت:\n" f"• 📁 پروژه: {escape_html(proj.name)}\n" f"• 🌐 مشاهده در گیت: {urls['web_url']}\n" f"• 📝 {res_msg}" if is_fa else f"\n\n💾 Git Commit & Push Result:\n" f"• 📁 Project: {escape_html(proj.name)}\n" f"• 🌐 View in Git: {urls['web_url']}\n" f"• 📝 {res_msg}" ) return badge, side_effects, created_task # ------------------------------------------------------------- # 27. GIT_INIT / CREATE_REPO # ------------------------------------------------------------- elif act in ("GIT_INIT", "INIT_REPO", "CREATE_REPO"): p_name = attrs.get("project") or attrs.get("name") or current_project_name proj = session.projects.get(p_name) if p_name else curr_proj if not proj: return f"\n⚠️ پروژه {escape_html(p_name or '')} یافت نشد." if is_fa else f"\n⚠️ Project {escape_html(p_name or '')} not found.", side_effects, created_task from git_manager import git_manager res = await git_manager.init_project_repo(proj.workspace, proj.name) urls = git_manager.get_repo_urls(proj.name) if res.get("success"): badge = ( f"\n\n🎉 مخزن گیت پروژه با موفقیت در Gitea ایجاد و متصل شد:\n" f"• 📁 پروژه: {escape_html(proj.name)}\n" f"• 🌐 آدرس وب: {urls['web_url']}\n" f"• 📥 دستور کلون: git clone {urls['clone_url']}" if is_fa else f"\n\n🎉 Git Repository Created & Linked on Gitea:\n" f"• 📁 Project: {escape_html(proj.name)}\n" f"• 🌐 Web URL: {urls['web_url']}\n" f"• 📥 Clone: git clone {urls['clone_url']}" ) else: badge = f"\n⚠️ خطا در راه‌اندازی مخزن گیت: {escape_html(str(res.get('error')))}" return badge, side_effects, created_task # ------------------------------------------------------------- # 28. GIT_CHECKOUT # ------------------------------------------------------------- elif act in ("GIT_CHECKOUT", "CHECKOUT_GIT", "GIT_SWITCH"): p_name = attrs.get("project") or attrs.get("name") or current_project_name proj = session.projects.get(p_name) if p_name else curr_proj if not proj: return f"\n⚠️ پروژه {escape_html(p_name or '')} یافت نشد." if is_fa else f"\n⚠️ Project {escape_html(p_name or '')} not found.", side_effects, created_task target = attrs.get("commit") or attrs.get("target") or attrs.get("branch") or attrs.get("_default") or "main" from git_manager import git_manager ok, msg = await git_manager.git_checkout(proj.workspace, target) if ok: badge = ( f"\n\n🔀 سوییچ شاخه/کامیت گیت انجام شد:\n" f"• 📁 پروژه: {escape_html(proj.name)}\n" f"• 🎯 مقصد: {escape_html(target)}\n" f"• 📋 {escape_html(msg)}" if is_fa else f"\n\n🔀 Git Checkout Succeeded:\n" f"• 📁 Project: {escape_html(proj.name)}\n" f"• 🎯 Target: {escape_html(target)}\n" f"• 📋 {escape_html(msg)}" ) else: badge = f"\n⚠️ خطا در Checkout گیت: {escape_html(msg)}" return badge, side_effects, created_task # ------------------------------------------------------------- # 29. GIT_RESET # ------------------------------------------------------------- elif act in ("GIT_RESET", "RESET_GIT", "GIT_ROLLBACK"): p_name = attrs.get("project") or attrs.get("name") or current_project_name proj = session.projects.get(p_name) if p_name else curr_proj if not proj: return f"\n⚠️ پروژه {escape_html(p_name or '')} یافت نشد." if is_fa else f"\n⚠️ Project {escape_html(p_name or '')} not found.", side_effects, created_task commit_h = attrs.get("commit") or attrs.get("hash") or attrs.get("_default") if not commit_h: return "\n⚠️ لطفاً شناسه کامیت مقصد را مشخص نمایید." if is_fa else "\n⚠️ Please specify the target commit hash.", side_effects, created_task from git_manager import git_manager ok, msg = await git_manager.git_reset_hard(proj.workspace, commit_h) if ok: badge = ( f"\n\n⏪ پروژه با موفقیت به کامیت مورد نظر بازگردانی شد:\n" f"• 📁 پروژه: {escape_html(proj.name)}\n" f"• 🔑 کامیت: {escape_html(commit_h)}\n" f"• 📝 {msg}" if is_fa else f"\n\n⏪ Git Reset Completed:\n" f"• 📁 Project: {escape_html(proj.name)}\n" f"• 🔑 Commit: {escape_html(commit_h)}\n" f"• 📝 {msg}" ) else: badge = f"\n⚠️ خطا در بازگردانی گیت: {escape_html(msg)}" return badge, side_effects, created_task # ------------------------------------------------------------- # 30. GIT_REVERT / GIT_UNDO # ------------------------------------------------------------- elif act in ("GIT_REVERT", "REVERT_GIT", "GIT_UNDO", "UNDO_GIT", "UNDO"): p_name = attrs.get("project") or attrs.get("name") or current_project_name proj = session.projects.get(p_name) if p_name else curr_proj if not proj: return f"\n⚠️ پروژه {escape_html(p_name or '')} یافت نشد." if is_fa else f"\n⚠️ Project {escape_html(p_name or '')} not found.", side_effects, created_task target = attrs.get("commit") or attrs.get("target") or attrs.get("_default") or "HEAD" from git_manager import git_manager ok, msg, extra = await git_manager.git_revert(proj.workspace, commit_target=target) if ok: badge = ( f"\n\n↩️ عملیات لغو تغییرات (Git Revert) با موفقیت انجام شد:\n" f"• 📁 پروژه: {escape_html(proj.name)}\n" f"• 🎯 کامیت هدف: {escape_html(target)}\n" f"• 📋 {msg}" if is_fa else f"\n\n↩️ Git Revert Succeeded:\n" f"• 📁 Project: {escape_html(proj.name)}\n" f"• 🎯 Target Commit: {escape_html(target)}\n" f"• 📋 {msg}" ) else: badge = f"\n⚠️ خطا در لغو کامیت: {msg}" return badge, side_effects, created_task # ------------------------------------------------------------- # 31. SAVE_MEMORY / STORE_MEMORY / SET_MEMORY # ------------------------------------------------------------- elif act in ("SAVE_MEMORY", "STORE_MEMORY", "SET_MEMORY", "ADD_MEMORY"): from memory_manager import memory_manager raw_type = (attrs.get("type") or "user").lower().strip() key = attrs.get("key") or attrs.get("slug") or attrs.get("name") content = attrs.get("content") or attrs.get("text") or attrs.get("value") or attrs.get("_default") category = (attrs.get("category") or "general").lower().strip() importance_raw = attrs.get("importance") or "1" req_project = attrs.get("project") or attrs.get("proj") or current_project_name try: importance = int(importance_raw) except Exception: importance = 1 if not key or not content: return ( "\n⚠️ خطا در ذخیره خاطره: کلید (key) و محتوا (content) الزامی هستند." if is_fa else "\n⚠️ Memory save error: 'key' and 'content' are required.", side_effects, created_task, ) # Map types: 'project' -> project tier, 'user'/'private' -> user tier, 'global' -> rejected/downgraded for AI if raw_type in ("project", "proj"): mem_type = "project" proj_name = req_project elif raw_type == "global": # Global memories are strictly bot-managed by Admin. AI cannot auto-save global memories. logger.info("AI attempted global memory save for user %s; safely converting to user memory.", chat_id) mem_type = "user" proj_name = None else: mem_type = "user" proj_name = None try: item, is_created = memory_manager.save_or_update( type_=mem_type, key=key, content=content, user_id=chat_id, project_name=proj_name, category=category, importance=importance, created_by=chat_id, ) if item.is_project: type_label = f"📁 پروژه اختصاصی ({escape_html(proj_name or '')})" elif item.is_user: type_label = "👤 کاربر (سراسری برای شما)" else: type_label = "🌐 عمومی (سیستم)" badge = ( f"\n\n🧠 خاطره هوش مصنوعی ثبت شد:\n" f"• 🏷️ بخش: {type_label}\n" f"• 🔑 کلید: {escape_html(item.key)}\n" f"• 📂 دسته: {escape_html(item.category)}\n" f"• 📝 محتوا: {escape_html(item.content)}" if is_fa else f"\n\n🧠 AI Memory Saved:\n" f"• 🏷️ Scope: {'Project (' + str(proj_name) + ')' if item.is_project else 'User'} \n" f"• 🔑 Key: {escape_html(item.key)}\n" f"• 📂 Category: {escape_html(item.category)}\n" f"• 📝 Content: {escape_html(item.content)}" ) except Exception as e: badge = f"\n⚠️ خطا در ذخیره خاطره: {escape_html(str(e))}" return badge, side_effects, created_task # ------------------------------------------------------------- # 31. UPDATE_MEMORY # ------------------------------------------------------------- elif act in ("UPDATE_MEMORY", "EDIT_MEMORY"): from memory_manager import memory_manager raw_type = (attrs.get("type") or "user").lower().strip() key = attrs.get("key") or attrs.get("slug") content = attrs.get("content") or attrs.get("text") or attrs.get("_default") category = attrs.get("category") importance_raw = attrs.get("importance") req_project = attrs.get("project") or attrs.get("proj") or current_project_name if not key or not content: return ( "\n⚠️ خطا در به‌روزرسانی خاطره: مشخص کردن key و content الزامی است." if is_fa else "\n⚠️ Memory update error: 'key' and 'content' required.", side_effects, created_task, ) if raw_type in ("project", "proj"): mem_type = "project" proj_name = req_project else: mem_type = "user" proj_name = None try: existing = memory_manager.get_by_key(type_=mem_type, key=key, user_id=chat_id, project_name=proj_name) cat = category or (existing.category if existing else "general") imp = int(importance_raw) if importance_raw and importance_raw.isdigit() else (existing.importance if existing else 1) item, is_created = memory_manager.save_or_update( type_=mem_type, key=key, content=content, user_id=chat_id, project_name=proj_name, category=cat, importance=imp, created_by=chat_id, ) type_label = f"📁 پروژه ({escape_html(proj_name or '')})" if item.is_project else "👤 کاربر (سراسری)" badge = ( f"\n\n🧠 خاطره هوش مصنوعی به‌روزرسانی شد:\n" f"• 🏷️ بخش: {type_label}\n" f"• 🔑 کلید: {escape_html(item.key)}\n" f"• 📝 محتوای جدید: {escape_html(item.content)}" if is_fa else f"\n\n🧠 AI Memory Updated:\n" f"• 🏷️ Scope: {type_label}\n" f"• 🔑 Key: {escape_html(item.key)}\n" f"• 📝 New Content: {escape_html(item.content)}" ) except Exception as e: badge = f"\n⚠️ خطا در ویرایش خاطره: {escape_html(str(e))}" return badge, side_effects, created_task # ------------------------------------------------------------- # 32. DELETE_MEMORY / FORGET_MEMORY / REMOVE_MEMORY # ------------------------------------------------------------- elif act in ("DELETE_MEMORY", "FORGET_MEMORY", "REMOVE_MEMORY"): from memory_manager import memory_manager mem_id_raw = attrs.get("id") key = attrs.get("key") or attrs.get("slug") or attrs.get("_default") raw_type = (attrs.get("type") or "user").lower().strip() req_project = attrs.get("project") or attrs.get("proj") or current_project_name mem_type = "project" if raw_type in ("project", "proj") else "user" proj_name = req_project if mem_type == "project" else None try: if mem_id_raw and mem_id_raw.isdigit(): ok = memory_manager.delete_by_id(int(mem_id_raw), user_id=chat_id, is_admin=is_admin) elif key: ok = memory_manager.delete_by_key( type_=mem_type, key=key, user_id=chat_id, project_name=proj_name, is_admin=is_admin ) else: return "\n⚠️ لطفاً شناسه یا کلید خاطره جهت حذف را مشخص کنید." if is_fa else "\n⚠️ Please specify memory id or key to delete.", side_effects, created_task if ok: badge = f"\n\n🗑️ خاطره مورد نظر از حافظه هوش مصنوعی حذف گردید." if is_fa else f"\n\n🗑️ Memory forgotten successfully." else: badge = f"\n⚠️ خاطره‌ای با این مشخصات یافت نشد." if is_fa else f"\n⚠️ Memory not found." except PermissionError: badge = "\n🚫 خطای دسترسی: فقط مالک یا مدیر می‌تواند این خاطره را حذف کند." if is_fa else "\n🚫 Access Denied." except Exception as e: badge = f"\n⚠️ خطا در حذف خاطره: {escape_html(str(e))}" return badge, side_effects, created_task # ------------------------------------------------------------- # 33. CLEAR_MEMORY / PURGE_MEMORY # ------------------------------------------------------------- elif act in ("CLEAR_MEMORY", "PURGE_MEMORY", "RESET_MEMORY"): from memory_manager import memory_manager raw_type = (attrs.get("type") or "user").lower().strip() req_project = attrs.get("project") or attrs.get("proj") or current_project_name if raw_type in ("project", "proj"): mem_type = "project" proj_name = req_project type_label = f"پروژه {proj_name}" elif raw_type == "global": if not is_admin: return "\n🚫 خطای دسترسی: پاک‌سازی خاطرات عمومی تنها توسط مدیر امکان‌پذیر است." if is_fa else "\n🚫 Access Denied: Only admins can clear global memories.", side_effects, created_task mem_type = "global" proj_name = None type_label = "عمومی سیستم" else: mem_type = "user" proj_name = None type_label = "شخصی کاربر" try: count = memory_manager.clear_memories( type_=mem_type, user_id=chat_id, project_name=proj_name, is_admin=is_admin ) badge = ( f"\n\n🧹 حافظه {type_label} پاک‌سازی شد ({count} مورد حذف گردید)." if is_fa else f"\n\n🧹 {type_label} memory cleared ({count} items purged)." ) except PermissionError: badge = "\n🚫 خطای دسترسی: عملیات مجاز نیست." if is_fa else "\n🚫 Access Denied." except Exception as e: badge = f"\n⚠️ خطا در پاک‌سازی حافظه: {escape_html(str(e))}" return badge, side_effects, created_task # ------------------------------------------------------------- # Auto-effort internal tag # ------------------------------------------------------------- elif act == "ESCALATE_EFFORT": return "", side_effects, created_task # ------------------------------------------------------------- # Dynamic Buttons internal tags (extracted during pre-pass) # ------------------------------------------------------------- elif act in ("BUTTON", "BUTTONS", "INLINE_BUTTONS", "KEYBOARD", "INLINE_KEYBOARD"): return "", side_effects, created_task # Unknown tag return f"\n[اکشن ناشناخته: {action_name}]" if is_fa else f"\n[Unknown action: {action_name}]", side_effects, created_task # ===================================================================== # Main AI Response Processor # ===================================================================== async def process_all_ai_actions( raw_text: str, chat_id: int, project_name: str, is_fa: bool, app: Any = None, ) -> Tuple[str, List[ScheduledTask], List[str], Optional[InlineKeyboardMarkup]]: """ Scans AI response text for ALL action tags: [[ACTION_NAME: key="value", ...]] Executes each action, gathers side-effects, replaces tags with formatted Telegram cards, extracts interactive inline buttons, and executes background side-effects. Returns (processed_text, created_tasks, executed_actions, inline_markup). """ if not raw_text or "[[" not in raw_text: return raw_text, [], [], None # 1. Parse and extract all interactive button tags first processed_text, inline_markup = parse_buttons_from_text( raw_text=raw_text, chat_id=chat_id, project_name=project_name, ) created_tasks: List[ScheduledTask] = [] executed_actions: List[str] = [] all_side_effects: List[Callable] = [] matches = list(ACTION_REGEX.finditer(processed_text)) if not matches: return processed_text, [], [], inline_markup for m in matches: full_tag = m.group(0) action_name = m.group(1).upper().strip() attrs_raw = m.group(2) or "" attrs = parse_tag_attributes(attrs_raw) try: rendered_badge, side_effects, task = await execute_action( action_name=action_name, attrs=attrs, chat_id=chat_id, current_project_name=project_name, is_fa=is_fa, app=app, ) if task: created_tasks.append(task) all_side_effects.extend(side_effects) executed_actions.append(action_name) # Replace tag with badge processed_text = processed_text.replace(full_tag, rendered_badge) except Exception as ex: logger.error(f"Error executing AI action tag {full_tag}: {ex}", exc_info=True) err_msg = f"\n⚠️ (خطا در اجرای عملکرد ربات: {escape_html(str(ex))})" if is_fa else f"\n⚠️ (Bot action error: {escape_html(str(ex))})" processed_text = processed_text.replace(full_tag, err_msg) # Execute all side-effects in background for side_effect in all_side_effects: asyncio.create_task(side_effect()) return processed_text, created_tasks, executed_actions, inline_markup # ===================================================================== # System Instructions for AGY Agent # ===================================================================== def get_ai_bot_actions_instruction(lang: str = "fa") -> str: """ Constructs the comprehensive system instruction empowering the AI to invoke ALL bot functions. """ return ( "[SYSTEM INSTRUCTION: TELEGRAM BOT ACTIONS & FUNCTION CALLING]\n" "You have complete access and authority to control and invoke ALL Telegram Bot features, project settings, model parameters, scheduled tasks, reverse proxies, and server operations directly from your conversation.\n" "Whenever the user asks (in Persian, English, or any language) to change settings, switch models, create or manage projects, schedule periodic tasks, configure subdomains on Caddy, export backups, restart the bot, or send files:\n" "You MUST output the corresponding action tag(s) anywhere in your reply. You can combine multiple action tags in a single reply if necessary.\n\n" "AVAILABLE ACTION TAGS & SYNTAX:\n" "1. Change AI Model for Project:\n" " [[SET_MODEL: model=\"\", project=\"\"]]\n" " Available models: 'gemini-3.7-flash-auto' (Gemini 3.7 Flash Auto Reasoning ⚡/🧠 - ارزیابی هوشمند خودکار), 'gemini-3.7-flash' (Gemini 3.7 Flash), 'gemini-3.7-flash-high' (Gemini 3.7 Flash High), 'gemini-3.1-pro' (Gemini 3.1 Pro), 'gemini-3.6-flash', 'claude-sonnet-4-6' (Claude Sonnet 4.6 Thinking), 'claude-opus-4-6-thinking' (Claude Opus 4.6 Thinking), 'gpt-oss-120b'.\n" " Example: [[SET_MODEL: model=\"gemini-3.7-flash-auto\"]]\n\n" "2. Change Reasoning Effort Level:\n" " [[SET_EFFORT: effort=\"low|medium|high\", project=\"\"]]\n" " Example: [[SET_EFFORT: effort=\"high\"]]\n\n" "3. Change Bot Language:\n" " [[SET_LANG: lang=\"fa|en|auto|...\"]]\n" " Example: [[SET_LANG: lang=\"fa\"]]\n\n" "4. Create New Project:\n" " [[NEW_PROJECT: name=\"\", workspace=\"\", model=\"\", effort=\"\"]]\n" " (Directory `/root/projects/{user_id}/{name}` is created automatically).\n" " Example: [[NEW_PROJECT: name=\"blog\"]]\n\n" "5. Switch Active Project:\n" " [[SWITCH_PROJECT: name=\"\"]]\n" " Example: [[SWITCH_PROJECT: name=\"blog\"]]\n\n" "6. Delete Project (Comprehensive Teardown & Resource Release):\n" " [[DELETE_PROJECT: name=\"\", delete_files=true|false]]\n" " (Automatically performs comprehensive teardown: releases Caddy subdomains/proxies, stops background processes/ports, removes systemd services, cancels scheduled cron tasks, deletes Gitea git repository, and purges physical files).\n" " Example: [[DELETE_PROJECT: name=\"old-project\", delete_files=true]]\n\n" "7. Rename Project:\n" " [[RENAME_PROJECT: old_name=\"\", new_name=\"\"]]\n\n" "8. Set Custom Workspace Directory:\n" " [[SET_WORKSPACE: path=\"/path/to/folder\", project=\"\"]]\n\n" "9. Share / Unshare Project:\n" " [[SHARE_PROJECT: user_id=, project=\"\"]]\n" " [[UNSHARE_PROJECT: user_id=, project=\"\"]]\n\n" "10. Reset / Clear Conversation Context:\n" " [[RESET_CONVERSATION: project=\"\"]]\n\n" "11. Switch / Delete / Clear Conversations:\n" " [[SWITCH_CONVERSATION: id=\"\"]]\n" " [[DELETE_CONVERSATION: id=\"\"]]\n" " [[CLEAR_CONVERSATIONS: project=\"\"]]\n\n" "12. Schedule Task / Timers / Cron:\n" " [[SCHEDULE_TASK: timing=\"\", max_runs=, type=\"prompt|command|reminder\", title=\"\", content=\"\"]]\n" " Timing examples: 'every 10m', 'in 15m', '14:30', 'cron: 0 23 * * *', 'every 1h (3 times)'.\n" " Types: 'prompt' (AI run), 'command' (bash script), 'reminder' (plain notification).\n" " [[CANCEL_TASK: id=\"\"]]\n" " [[PAUSE_TASK: id=\"\"]]\n" " [[RESUME_TASK: id=\"\"]]\n\n" "13. Project Backup & Export (Zip):\n" " [[BACKUP_PROJECT: project=\"\"]]\n\n" "14. Send File / Photo / Audio to Telegram:\n" " [[SEND_FILE: path=\"/path/to/file.ext\", caption=\"\"]]\n" " [[SEND_PHOTO: path=\"/path/to/image.png\", caption=\"\"]]\n" " [[SEND_AUDIO: path=\"/path/to/audio.mp3\", caption=\"\"]]\n\n" "15. Configure Subdomain & Caddy HTTPS Reverse Proxy (*.msa.artacloud.ir):\n" " [[SETUP_SUBDOMAIN: subdomain=\"\", port=]]\n" " Automatically links `{subdomain}.msa.artacloud.ir` to `localhost:{port}` with free SSL.\n" " [[REMOVE_SUBDOMAIN: subdomain=\"\"]]\n\n" "16. Git Version Control & Gitea Integration:\n" " [[GIT_SYNC: project=\"\", message=\"\"]]\n" " (Pulls remote changes from Gitea, commits any local modifications, and pushes to main branch).\n" " [[GIT_COMMIT: message=\"\", project=\"\"]]\n" " [[GIT_INIT: project=\"\"]]\n\n" "17. Restart Telegram Bot Service:\n" " [[RESTART_BOT]]\n\n" "18. Long-Term AI Memory Management (3-Tier: User & Project):\n" " [[SAVE_MEMORY: type=\"user|project\", key=\"\", category=\"\", content=\"\"]]\n" " (Use type='user' for personal user profile/habits, or type='project' for current project architecture/decisions. Global memory is read-only for AI).\n" " [[UPDATE_MEMORY: type=\"user|project\", key=\"\", content=\"\"]]\n" " [[DELETE_MEMORY: type=\"user|project\", key=\"\"]]\n" " [[CLEAR_MEMORY: type=\"user|project\"]]\n\n" "19. Interactive Inline Buttons (دکمه‌های شیشه‌ای تعاملی زیر پیام نهایی):\n" " You can attach interactive buttons with URLs, follow-up prompts, slash commands, or custom actions underneath your final response to give the user quick one-tap actions:\n" " • Single button tag syntax:\n" " [[BUTTON: text=\"\", url=\"\", prompt=\"\", cmd=\"\", row=]]\n" " Examples:\n" " [[BUTTON: text=\"🌐 مشاهده وب‌سایت\", url=\"https://app.msa.artacloud.ir\", row=1]]\n" " [[BUTTON: text=\"🐙 مخزن گیت\", url=\"https://git.msa.artacloud.ir/root/my-app\", row=1]]\n" " [[BUTTON: text=\"🔄 اجرای مجدد تست‌ها\", prompt=\"تست‌های برنامه را دوباره اجرا کن\", row=2]]\n" " [[BUTTON: text=\"📊 وضعیت مصرف\", cmd=\"/usage\", row=2]]\n\n" " • Grid / Matrix layout tag syntax (JSON layout):\n" " [[BUTTONS: layout=[\n" " [{\"text\": \"🚀 مشاهده سایت\", \"url\": \"https://app.msa.artacloud.ir\"}, {\"text\": \"📂 مخزن گیت\", \"url\": \"https://git.msa.artacloud.ir/root/my-app\"}],\n" " [{\"text\": \"🔄 تست مجدد\", \"prompt\": \"تست‌ها را دوباره اجرا کن\"}, {\"text\": \"📊 سهمیه مصرف\", \"cmd\": \"/usage\"}]\n" " ]]]\n\n" "Always explain to the user in a friendly and professional tone what actions have been performed." )