root/default
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AI Update: پیده سازی کن

Browse Source
This commit is contained in:
Antigravity Bot
2026-08-30 12:28:49 +03:30
parent b2ba9a17fd
commit 64f99b542a
5 changed files with 434 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram-agy-bot/bot.py
+103 -11
View File
@@ -4557,6 +4557,89 @@ async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
except Exception:
await query.message.reply_html(text, reply_markup=markup, disable_web_page_preview=True)
elif data.startswith("aibtn:"):
btn_id = data.split(":", 1)[1].strip()
from bot_actions import get_ai_button_payload
payload_data = get_ai_button_payload(btn_id)
if not payload_data:
await query.answer("⚠️ این دکمه منقضی شده یا نامعتبر است." if is_fa else "⚠️ This button has expired or is invalid.", show_alert=True)
return
act_type = payload_data.get("type", "prompt")
act_value = payload_data.get("value", "").strip()
btn_project = payload_data.get("project")
if act_type == "prompt":
await query.answer("🧠 در حال ارسال دستور به هوش مصنوعی..." if is_fa else "🧠 Processing prompt with AI...")
if btn_project:
accessible = session_manager.get_all_accessible_projects(chat_id)
for k, p in accessible.items():
if k.lower() == btn_project.lower() or p.name.lower() == btn_project.lower():
session.current_project = p.name
session_manager.save()
break
prompt_echo = (
f"💬 <b>دستور ارسالی:</b> <code>{escape_html(act_value)}</code>"
if is_fa else
f"💬 <b>Sent prompt:</b> <code>{escape_html(act_value)}</code>"
)
try:
await query.message.reply_html(prompt_echo)
except Exception:
pass
asyncio.create_task(process_agent_turn_by_chat_id(context.application, chat_id, act_value))
elif act_type == "cmd":
await query.answer("⚡ در حال اجرای دستور..." if is_fa else "⚡ Running command...")
cmd = act_value
if cmd.startswith("/"):
cmd_parts = cmd.split(maxsplit=1)
c_name = cmd_parts[0].lower()
if c_name in ("/usage", "/quota", "/credits", "/credit"):
text, markup = await build_usage_report(chat_id)
await query.message.reply_html(text, reply_markup=markup)
elif c_name in ("/status", "/stats", "/sys"):
text, markup = build_server_hardware_menu(chat_id)
await query.message.reply_html(text, reply_markup=markup)
elif c_name in ("/git", "/gitea"):
text, markup = await build_git_menu(chat_id)
await query.message.reply_html(text, reply_markup=markup)
elif c_name in ("/tasks", "/cron", "/schedule"):
text, markup = build_tasks_menu(chat_id)
await query.message.reply_html(text, reply_markup=markup)
elif c_name in ("/memory", "/mem"):
text, markup = build_memory_menu(chat_id)
await query.message.reply_html(text, reply_markup=markup)
elif c_name in ("/projects", "/proj"):
text, markup = build_projects_menu(chat_id)
await query.message.reply_html(text, reply_markup=markup)
elif c_name in ("/backup", "/export"):
if curr_proj:
await query.message.reply_html("📦 در حال تهیه فایل پشتیبان..." if is_fa else "📦 Creating backup...")
zip_p = await backup_manager.create_project_backup(curr_proj.workspace, curr_proj.name)
if zip_p and zip_p.exists():
await context.bot.send_document(
chat_id=chat_id,
document=zip_p.open("rb"),
filename=zip_p.name,
caption=f"📦 Backup of <b>{escape_html(curr_proj.name)}</b>",
parse_mode=constants.ParseMode.HTML,
)
else:
await query.message.reply_html("❌ خطا در ایجاد پشتیبان." if is_fa else "❌ Failed to create backup.")
else:
asyncio.create_task(process_agent_turn_by_chat_id(context.application, chat_id, cmd))
else:
asyncio.create_task(process_agent_turn_by_chat_id(context.application, chat_id, cmd))
elif act_type == "action":
query.data = act_value
await callback_handler(update, context)
return
elif data in ("btn_usage_menu", "btn_usage_refresh"):
await query.answer("⏳ استعلام مصرف..." if is_fa else "⏳ Refreshing...")
text, markup = await build_usage_report(chat_id)
@@ -6417,8 +6500,8 @@ async def process_agent_turn(
# 1. Convert Markdown to Telegram HTML first
formatted_html = markdown_to_telegram_html(raw_text)
# 2. Intercept and execute all AI Bot Actions (Model switching, Effort, Languages, Projects, Tasks, Caddy, Files, etc.)
formatted_html, created_tasks, executed_actions = await process_all_ai_actions(
# 2. Intercept and execute all AI Bot Actions (Model switching, Effort, Languages, Projects, Tasks, Caddy, Buttons, Files, etc.)
formatted_html, created_tasks, executed_actions, inline_markup = await process_all_ai_actions(
raw_text=formatted_html,
chat_id=chat_id,
project_name=curr_proj.name,
@@ -6428,17 +6511,26 @@ async def process_agent_turn(
chunks = split_message(formatted_html, max_length=settings.max_message_length)
first_chunk = chunks[0] if chunks else "✅ <i>Done</i>"
try:
await status_msg.edit_text(first_chunk, parse_mode=constants.ParseMode.HTML, disable_web_page_preview=True, reply_markup=None)
except Exception:
await status_msg.edit_text(escape_html(formatted_html[:settings.max_message_length]), parse_mode=constants.ParseMode.HTML, reply_markup=None)
for follow_up in chunks[1:]:
if len(chunks) <= 1:
first_chunk = chunks[0] if chunks else "✅ <i>Done</i>"
try:
await update.effective_message.reply_html(follow_up, disable_web_page_preview=True)
await status_msg.edit_text(first_chunk, parse_mode=constants.ParseMode.HTML, disable_web_page_preview=True, reply_markup=inline_markup)
except Exception:
await update.effective_message.reply_text(follow_up)
await status_msg.edit_text(escape_html(formatted_html[:settings.max_message_length]), parse_mode=constants.ParseMode.HTML, reply_markup=inline_markup)
else:
first_chunk = chunks[0] if chunks else "✅ <i>Done</i>"
try:
await status_msg.edit_text(first_chunk, parse_mode=constants.ParseMode.HTML, disable_web_page_preview=True, reply_markup=None)
except Exception:
await status_msg.edit_text(escape_html(formatted_html[:settings.max_message_length]), parse_mode=constants.ParseMode.HTML, reply_markup=None)
for i, follow_up in enumerate(chunks[1:]):
is_last = (i == len(chunks[1:]) - 1)
chunk_markup = inline_markup if is_last else None
try:
await update.effective_message.reply_html(follow_up, disable_web_page_preview=True, reply_markup=chunk_markup)
except Exception:
await update.effective_message.reply_text(follow_up, reply_markup=chunk_markup)
# Mark successfully completed and delivered
session.turn_in_progress = False
telegram-agy-bot/bot_actions.py
+314 -8
View File
@@ -9,6 +9,8 @@ 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 (
@@ -262,6 +264,284 @@ def parse_tag_attributes(attrs_str: str) -> Dict[str, str]:
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
# =====================================================================
@@ -1297,6 +1577,12 @@ async def execute_action(
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<i>[اکشن ناشناخته: {action_name}]</i>" if is_fa else f"\n<i>[Unknown action: {action_name}]</i>", side_effects, created_task
@@ -1312,24 +1598,30 @@ async def process_all_ai_actions(
project_name: str,
is_fa: bool,
app: Any = None,
) -> Tuple[str, List[ScheduledTask], List[str]]:
) -> 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,
and executes background side-effects.
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, [], []
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(raw_text))
matches = list(ACTION_REGEX.finditer(processed_text))
if not matches:
return raw_text, [], []
processed_text = raw_text
return processed_text, [], [], inline_markup
for m in matches:
full_tag = m.group(0)
@@ -1364,7 +1656,7 @@ async def process_all_ai_actions(
for side_effect in all_side_effects:
asyncio.create_task(side_effect())
return processed_text, created_tasks, executed_actions
return processed_text, created_tasks, executed_actions, inline_markup
# =====================================================================
@@ -1445,5 +1737,19 @@ def get_ai_bot_actions_instruction(lang: str = "fa") -> str:
" [[UPDATE_MEMORY: type=\"user|project\", key=\"<slug>\", content=\"<new_content>\"]]\n"
" [[DELETE_MEMORY: type=\"user|project\", key=\"<slug>\"]]\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=\"<button_text>\", url=\"<url>\", prompt=\"<follow_up_prompt>\", cmd=\"<bot_or_shell_cmd>\", row=<row_number_optional>]]\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."
)
telegram-agy-bot/scheduler.py
+1 -1
View File
@@ -495,7 +495,7 @@ class TaskScheduler:
output_text = result.text or "✅ دستور با موفقیت اجرا شد (بدون خروجی متنی)."
try:
from bot_actions import process_all_ai_actions
output_text, _, _ = await 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,
telegram-agy-bot/sessions.json
+8 -8
View File
File diff suppressed because one or more lines are too long