import re
import html
import json
from pathlib import Path
from typing import List, Optional, Dict, Any
def escape_html(text: str) -> str:
"""Safely escapes text for Telegram HTML parse mode."""
if not text:
return ""
return html.escape(text, quote=False)
def format_token_count(n: int) -> str:
"""Formats an integer token count to a human-readable string (e.g. 360K, 1.2M)."""
if n >= 1_000_000:
val = n / 1_000_000
if val.is_integer() or f"{val:.1f}".endswith(".0"):
return f"{int(val)}M"
return f"{val:.1f}M"
elif n >= 1_000:
return f"{round(n / 1_000)}K"
return str(n)
def format_duration_compact(seconds: float) -> str:
"""Formats duration in seconds to M:SS or H:MM:SS format (e.g. 9:47, 1:02:15)."""
sec = int(round(seconds))
if sec >= 3600:
h, rem = divmod(sec, 3600)
m, s = divmod(rem, 60)
return f"{h}:{m:02d}:{s:02d}"
m, s = divmod(sec, 60)
return f"{m}:{s:02d}"
def get_model_display_name(model: Optional[str] = None, effort: Optional[str] = None) -> str:
"""Returns a clean user-friendly display name for the model and reasoning effort."""
if not model:
model = "gemini-3.7-flash-auto"
m_clean = model.lower().strip()
eff_clean = (effort or "").lower().strip()
if not eff_clean:
if "flash-high" in m_clean:
eff_clean = "high"
elif "flash-auto" in m_clean or "3.7-flash" in m_clean or "3.6-flash" in m_clean or "3.1-pro" in m_clean:
eff_clean = "medium"
eff_map = {"high": "High", "medium": "Med", "med": "Med", "low": "Low"}
eff_tag = f" [{eff_map.get(eff_clean, eff_clean.capitalize())}]" if eff_clean else ""
if "flash-auto" in m_clean or m_clean == "gemini-3.7-flash-auto":
return f"Gemini 3.7 Flash Auto{eff_tag}"
elif "flash-high" in m_clean:
return "Gemini 3.7 Flash [High]"
elif "3.7-flash" in m_clean:
return f"Gemini 3.7 Flash{eff_tag}"
elif "3.6-flash" in m_clean:
return f"Gemini 3.6 Flash{eff_tag}"
elif "3.1-pro" in m_clean or "pro" in m_clean:
return f"Gemini 3.1 Pro{eff_tag}"
elif "opus" in m_clean:
return f"Claude Opus 4.6 Thinking{eff_tag}"
elif "sonnet" in m_clean or "claude" in m_clean:
return f"Claude Sonnet 4.6 Thinking{eff_tag}"
elif "gpt" in m_clean:
return "GPT-120B"
else:
return f"{model}{eff_tag}"
def format_context_stats(
usage: Optional[Dict[str, Any]] = None,
duration: float = 0.0,
model: str = "",
effort: Optional[str] = None,
lang: str = "fa",
project_name: Optional[str] = None,
conversation_id: Optional[str] = None,
) -> str:
"""Formats token usage, project name, executed model/effort, and context length into an elegant 3-line footnote."""
p_name = project_name or "default"
is_fa = (lang or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)")
m_label = "Gemini 3.7 Flash"
if model:
m_clean = model.lower().strip()
eff_clean = (effort or "").lower().strip()
if "flash-auto" in m_clean or m_clean == "gemini-3.7-flash-auto":
if eff_clean == "high":
m_label = "3.7-Auto [High]"
elif eff_clean == "medium":
m_label = "3.7-Auto [Med]"
else:
m_label = "3.7-Auto [Low]"
elif "flash-high" in m_clean:
m_label = "3.7-Flash [High]"
elif "3.7-flash" in m_clean:
m_label = f"3.7-Flash [{eff_clean.capitalize()}]" if eff_clean else "3.7-Flash"
elif "3.6-flash" in m_clean:
m_label = f"3.6-Flash [{eff_clean.capitalize()}]" if eff_clean else "3.6-Flash"
elif "3.1-pro" in m_clean:
m_label = f"3.1-Pro [{eff_clean.capitalize()}]" if eff_clean else "3.1-Pro"
elif "opus" in m_clean:
m_label = "Opus 4.6"
elif "sonnet" in m_clean or "claude" in m_clean:
m_label = "Sonnet 4.6"
elif "gpt" in m_clean:
m_label = "GPT-120B"
else:
m_label = model
input_tokens = usage.get("input_tokens", 0) if usage else 0
output_tokens = usage.get("output_tokens", 0) if usage else 0
total_tokens = usage.get("total_tokens", input_tokens + output_tokens) if usage else 0
# Estimate model context window capacity
model_lower = (model or "").lower()
if "3.1-pro" in model_lower or "pro" in model_lower:
max_context = 2_000_000
elif "claude" in model_lower:
max_context = 200_000
elif "gpt" in model_lower:
max_context = 128_000
else:
max_context = 1_000_000
# Calculate actual active conversation tokens if conversation_id is provided
active_tokens = input_tokens
if conversation_id:
try:
from agy_engine import get_conversation_context_tokens
real_tok = get_conversation_context_tokens(conversation_id)
if real_tok > 0:
active_tokens = real_tok
except Exception:
pass
pct = min(100.0, max(0.0, (active_tokens / max_context) * 100)) if max_context else 0
pct_str = f"{int(round(pct))}%"
display_input = active_tokens if (active_tokens > 0) else input_tokens
ti_str = format_token_count(display_input) if display_input > 0 else "0"
to_str = format_token_count(output_tokens) if output_tokens > 0 else "0"
dur_str = format_duration_compact(duration) if duration > 0 else "<1s"
line1 = f"{escape_html(p_name)} {escape_html(m_label)}"
line2 = f"CL:{pct_str} Ti:{ti_str} To:{to_str} D:{dur_str}"
return f"{line1}\n{line2}"
def format_thought(
thought_text: str,
project_name: Optional[str] = None,
model_name: Optional[str] = None,
lang: str = "fa",
) -> str:
"""Formats model reasoning / thinking process with model context."""
if not thought_text.strip():
return ""
escaped = escape_html(thought_text.strip())
is_fa = (lang or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)")
label = model_name or project_name
if label:
badge = escape_html(label)
if model_name:
header = f"💭 تفکر و استدلال (مدل: {badge}):" if is_fa else f"💭 Reasoning / Thinking (Model: {badge}):"
else:
header = f"💭 تفکر و استدلال ({badge}):" if is_fa else f"💭 Reasoning / Thinking ({badge}):"
else:
header = "💭 تفکر و استدلال:" if is_fa else "💭 Reasoning / Thinking:"
return f"
{header}\n{escaped}\n" TOOL_EMOJI_MAP: Dict[str, str] = { "run_command": "💻", "manage_task": "⚙️", "view_file": "👁️", "view_image": "👁️", "replace_file_content": "✏️", "write_to_file": "📝", "list_dir": "📂", "grep_search": "🔍", "find_by_name": "📁", "search_web": "🌐", "read_url_content": "🔗", "generate_image": "🎨", "schedule": "⏰", "invoke_subagent": "🤖", "define_subagent": "🧠", "manage_subagents": "👥", "send_message": "💬", "ask_question": "❓", } def format_tool_call(tool_name: str, tool_args: Any, lang: str = "fa") -> str: """Formats an active tool execution notification into a clean, concise badge.""" is_fa = (lang or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)") label = "ابزار" if is_fa else "Tool" tool_emoji = TOOL_EMOJI_MAP.get(tool_name, "🔧") # Parse tool arguments if passed as JSON string or dict params = {} if isinstance(tool_args, dict): params = tool_args elif isinstance(tool_args, str): try: params = json.loads(tool_args) except Exception: params = {} # Extract friendly description/action action = "" target = "" if isinstance(params, dict) and params: action = params.get("toolAction") or params.get("toolSummary") or params.get("Description") or "" # Tool specific highlights if tool_name == "run_command": cmd = params.get("CommandLine", "") target = escape_html(cmd[:120]) if cmd else "" elif tool_name in ("write_to_file", "replace_file_content"): file_p = params.get("TargetFile", "") instr = params.get("Instruction", "") if file_p: target = escape_html(Path(file_p).name if "/" in file_p else file_p) if instr and not action: action = instr[:100] elif tool_name in ("view_file", "view_image"): file_p = params.get("AbsolutePath", "") if file_p: target = escape_html(Path(file_p).name if "/" in file_p else file_p) elif tool_name == "list_dir": dir_p = params.get("DirectoryPath", "") if dir_p: target = escape_html(dir_p[:80]) elif tool_name == "grep_search": q = params.get("Query", "") if q: target = escape_html(q[:80]) elif tool_name == "find_by_name": pat = params.get("Pattern", "") if pat: target = escape_html(pat[:80]) elif tool_name == "search_web": q = params.get("query", "") if q: target = escape_html(q[:80]) elif tool_name == "read_url_content": u = params.get("Url", "") if u: target = escape_html(u[:80]) elif tool_name == "generate_image": p = params.get("Prompt", "") if p: target = escape_html(p[:80]) elif tool_name == "schedule": timing = params.get("CronExpression") or (f"{params.get('DurationSeconds')}s" if params.get("DurationSeconds") else "") or params.get("Prompt", "") if timing: target = escape_html(str(timing)[:80]) elif tool_name == "manage_task": act = params.get("Action", "") tid = params.get("TaskId", "") target = f"{escape_html(act)} {escape_html(tid)}".strip() elif tool_name == "invoke_subagent": subagents = params.get("Subagents", []) roles = [s.get("Role", s.get("TypeName", "Agent")) for s in subagents if isinstance(s, dict)] if roles: target = escape_html(", ".join(roles)[:80]) elif tool_name == "define_subagent": name = params.get("name", "") if name: target = escape_html(name[:80]) elif tool_name == "manage_subagents": act = params.get("Action", "") target = escape_html(act) if act else "" elif tool_name == "send_message": msg = params.get("Message", "") target = escape_html(msg[:80]) if msg else "" elif tool_name == "ask_question": target = "درخواست تایید / سوال" if not action and not target and isinstance(tool_args, str) and tool_args.strip(): # Fallback short string preview short_args = tool_args.strip().replace("\n", " ") if len(short_args) > 100: short_args = short_args[:100] + "..." target = escape_html(short_args) action_text = f" ({escape_html(action)})" if action else "" out = f"{label}:
{escape_html(tool_name)} {tool_emoji}{action_text}"
if target:
out += f"\n ↳ {target}"
return out + "\n"
def markdown_to_telegram_html(text: str) -> str:
"""
Converts standard GitHub-flavored Markdown text to valid Telegram HTML.
Preserves fenced code blocks with syntax highlighting, inline code,
headings, blockquotes, bold, italics, strikethrough, links, and bullet lists.
"""
if not text:
return ""
text = text.replace("\r\n", "\n").replace("\r", "\n")
# If stream ends with unclosed code block, temporarily close it for formatting
fences = re.findall(r"^```", text, flags=re.MULTILINE)
if len(fences) % 2 != 0:
text += "\n```"
# 1. Protect fenced code blocks: ```lang ... ```
code_blocks = []
def code_block_sub(match):
lang = (match.group(1) or "").strip()
code = match.group(2)
idx = len(code_blocks)
code_blocks.append((lang, code))
return f"\x00CB{idx}\x00"
code_block_pattern = re.compile(r"```([a-zA-Z0-9_-]*)\n?(.*?)```", re.DOTALL)
processed = code_block_pattern.sub(code_block_sub, text)
# 2. Protect inline code: `code`
inline_codes = []
def inline_code_sub(match):
code = match.group(1)
idx = len(inline_codes)
inline_codes.append(code)
return f"\x00IC{idx}\x00"
processed = re.sub(r"`([^`\n]+)`", inline_code_sub, processed)
# 3. Escape general HTML characters
processed = escape_html(processed)
# 4. Markdown Headings: # Heading, ## Heading, ### Heading, etc.
processed = re.sub(r"^(?:#{1,6})\s+(.+)$", r"\1", processed, flags=re.MULTILINE)
# 5. Horizontal rules: ---, ***, ___ on a line by itself
processed = re.sub(r"^[ \t]*(?:-{3,}|\*{3,}|_{3,})[ \t]*$", "──────────────", processed, flags=re.MULTILINE)
# 6. Blockquotes: lines starting with > (since > was escaped)
processed = re.sub(r"^>\s*(.+)$", r"\1", processed, flags=re.MULTILINE) processed = re.sub(r"\n
", "\n", processed) # 7. Unordered lists: * item or - item processed = re.sub(r"^([ \t]*)[*\-]\s+(.+)$", r"\1• \2", processed, flags=re.MULTILINE) # 8. Bold: **text** or __text__ processed = re.sub(r"\*\*(.+?)\*\*", r"\1", processed) processed = re.sub(r"__(.+?)__", r"\1", processed) # 9. Italic: *text* or _text_ (ensure not touching word internals or placeholders) processed = re.sub(r"(?\1", processed) processed = re.sub(r"(?\1", processed) # 10. Strikethrough: ~~text~~ processed = re.sub(r"~~(.+?)~~", r"\1", processed) # 11. Links: [text](url) - Only http/https/tg/mailto supported in Telegram HTML def link_sub(match): label = match.group(1) url = match.group(2) if re.match(r"^(https?://|tg://|mailto:)", url, re.IGNORECASE): return f'{label}' else: return f"{label}" processed = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", link_sub, processed) # 12. Restore inline code for idx, code in enumerate(inline_codes): escaped_code = escape_html(code) processed = processed.replace(f"\x00IC{idx}\x00", f"{escaped_code}") # 13. Restore code blocks for idx, (lang, code) in enumerate(code_blocks): escaped_code = escape_html(code.strip("\n")) lang_attr = f' class="language-{escape_html(lang)}"' if lang else "" replacement = f"" processed = processed.replace(f"\x00CB{idx}\x00", replacement) return processed def balance_tags(chunk: str, open_tags: List[tuple]) -> List[tuple]: """Finds unclosed HTML tags in chunk.""" tag_pattern = re.compile(r"<\s*(/)?\s*([a-zA-Z0-9]+)(?:\s+[^>]*?)?>") active_tags = list(open_tags) for match in tag_pattern.finditer(chunk): is_closing = bool(match.group(1)) tag_name = match.group(2).lower() if tag_name in ("br", "hr", "img"): continue if is_closing: for i in range(len(active_tags) - 1, -1, -1): if active_tags[i][0] == tag_name: active_tags.pop(i) break else: active_tags.append((tag_name, match.group(0))) return active_tags def split_message(text: str, max_length: int = 4000) -> List[str]: """ Splits long messages into Telegram-compliant chunks (<4096 chars). Ensures that HTML tags remain balanced across split boundaries. """ if len(text) <= max_length: return [text] chunks = [] current_pos = 0 total_len = len(text) active_tags = [] while current_pos < total_len: prefix = "".join(full_tag for tag_name, full_tag in active_tags) effective_max = max_length - len(prefix) - 60 if effective_max < 100: effective_max = max_length // 2 if total_len - current_pos <= effective_max: chunk = prefix + text[current_pos:] chunks.append(chunk) break split_candidate = current_pos + effective_max # Find best split point (newline > space > hard cut) newline_pos = text.rfind("\n", current_pos, split_candidate) if newline_pos != -1 and newline_pos > current_pos + (effective_max // 3): split_at = newline_pos + 1 else: space_pos = text.rfind(" ", current_pos, split_candidate) if space_pos != -1 and space_pos > current_pos + (effective_max // 3): split_at = space_pos + 1 else: split_at = split_candidate raw_chunk = text[current_pos:split_at] chunk_active = balance_tags(raw_chunk, active_tags) suffix = "".join(f"{tag_name}>" for tag_name, _ in reversed(chunk_active)) final_chunk = prefix + raw_chunk + suffix chunks.append(final_chunk) active_tags = chunk_active current_pos = split_at return chunks def format_git_info( project_name: str, web_url: str, clone_url: str, branch: str = "main", last_commit: Optional[Dict[str, Any]] = None, dirty: bool = False, files_count: int = 0, lang: str = "fa", ) -> str: """Formats Gitea / Git repository status in Telegram HTML.""" is_fa = (lang or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)") clean_p = escape_html(project_name) commit_str = "—" if last_commit: c_hash = escape_html(last_commit.get("hash", "")) c_msg = escape_html(last_commit.get("message", "")) c_time = escape_html(last_commit.get("time", "")) commit_str = f"{escaped_code}{c_hash}- {c_msg} ({c_time})" dirty_str = f"⚠️ {files_count} فایل تغییریافته (منتظر کامیت)" if dirty else "✅ درخت کاری تمیز است (همگام)" dirty_str_en = f"⚠️ {files_count} modified files (uncommitted)" if dirty else "✅ Working tree clean (in sync)" if is_fa: return ( f"🐙 مدیریت مخزن گیت (Gitea Version Control)\n\n" f"• 📁 پروژه:{clean_p}\n" f"• 🌐 مخزن تحت وب (Gitea):\n{web_url}\n" f"• 📥 دستور کلون در سیستم شخصی:\ngit clone {clone_url}\n\n" f"• 🌿 شاخه فعال:{escape_html(branch)}\n" f"• 🔖 آخرین کامیت: {commit_str}\n" f"• 📊 وضعیت تغییرات محلی: {dirty_str}\n\n" f"💡 تمامی تغییرات اعمالشده توسط هوش مصنوعی به صورت خودکار کامیت و در مخزن Gitea پوش میشوند." ) else: return ( f"🐙 Git Repository Manager (Gitea)\n\n" f"• 📁 Project:{clean_p}\n" f"• 🌐 Web Repository (Gitea):\n{web_url}\n" f"• 📥 Clone Command:\ngit clone {clone_url}\n\n" f"• 🌿 Active Branch:{escape_html(branch)}\n" f"• 🔖 Last Commit: {commit_str}\n" f"• 📊 Local Status: {dirty_str_en}\n\n" f"💡 All changes made by AI are automatically committed and pushed to your Gitea repo." ) def format_git_commits_page( project_name: str, commits: List[Dict[str, Any]], current_page: int = 1, total_pages: int = 1, branch: str = "main", lang: str = "fa", ) -> str: """Formats a page of commit history list.""" is_fa = (lang or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)") clean_p = escape_html(project_name) if not commits: if is_fa: return f"📜 تاریخچه کامیتهای پروژه{clean_p}\n\n⚠️ هیچ کامیتی در این مخزن یافت نشد." else: return f"📜 Commit History for{clean_p}\n\n⚠️ No commits found in this repository." if is_fa: lines = [ f"📜 تاریخچه ۲۰ کامیت اخیر (پروژه:{clean_p})", f"🌿 شاخه/موقعیت فعلی:{escape_html(branch)}", f"📄 صفحه: {current_page} از {total_pages}\n", ] for c in commits: short_h = escape_html(c.get("short_hash", "")) author = escape_html(c.get("author", "")) time_rel = escape_html(c.get("relative_time", "")) subject = escape_html(c.get("subject", "")) is_curr = c.get("is_current", False) badge = "📍 [فعال] " if is_curr else "" lines.append(f"{badge}•{short_h}- {subject}\n 👤 {author} ({time_rel})") lines.append("\n💡 جهت مشاهده جزئیات، جابهجایی یا بازگشت به هر کامیت، دکمه مربوطه را انتخاب نمایید:") return "\n".join(lines) else: lines = [ f"📜 Recent Commits (Project:{clean_p})", f"🌿 Active Branch/Ref:{escape_html(branch)}", f"📄 Page: {current_page} of {total_pages}\n", ] for c in commits: short_h = escape_html(c.get("short_hash", "")) author = escape_html(c.get("author", "")) time_rel = escape_html(c.get("relative_time", "")) subject = escape_html(c.get("subject", "")) is_curr = c.get("is_current", False) badge = "📍 [HEAD] " if is_curr else "" lines.append(f"{badge}•{short_h}- {subject}\n 👤 {author} ({time_rel})") lines.append("\n💡 Select a commit button below to inspect, checkout, or reset:") return "\n".join(lines) def format_commit_detail_view( project_name: str, commit_detail: Dict[str, Any], web_url: str = "", lang: str = "fa", ) -> str: """Formats full details of a specific commit.""" is_fa = (lang or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)") clean_p = escape_html(project_name) full_h = escape_html(commit_detail.get("hash", "")) short_h = escape_html(commit_detail.get("short_hash", "")) author = escape_html(commit_detail.get("author", "")) email = escape_html(commit_detail.get("email", "")) time_rel = escape_html(commit_detail.get("relative_time", "")) date_str = escape_html(commit_detail.get("date", "")) subject = escape_html(commit_detail.get("subject", "")) stat = escape_html(commit_detail.get("stat", "")) is_curr = commit_detail.get("is_current", False) commit_link = f"{web_url}/commit/{full_h}" if web_url else "" link_html = f"\n• 🌐 مشاهده در وب: {commit_link}" if commit_link else "" stat_block = f"\n\n📊 فایلهای تغییریافته:\n{stat}" if stat else "" stat_block_en = f"\n\n📊 Changed Files:\n{stat}" if stat else "" status_tag = "📍 این کامیت هماکنون کامیت فعال (HEAD) پروژه است.\n" if is_curr else "" status_tag_en = "📍 This commit is currently the active HEAD of the project.\n" if is_curr else "" if is_fa: return ( f"🔖 مشخصات و جزئیات کامیت گیت\n\n" f"• 📁 پروژه:{clean_p}\n" f"• 🔑 شناسه کامیت:{full_h}\n" f"• 👤 نویسنده: {author} <{email}>\n" f"• 📅 زمان: {date_str} ({time_rel})\n" f"• 📝 پیام کامیت:\n{subject}" f"{link_html}" f"{stat_block}\n\n" f"{status_tag}" f"⚠️ توجه: با بازگردانی (Reset)، وضعیت پروژه به این کامیت بازمیگردد. با سوییچ (Checkout) میتوانید بدون تغییر شاخه اصلی، وضعیت این نسخه را مشاهده یا تست کنید." ) else: return ( f"🔖 Git Commit Details\n\n" f"• 📁 Project:{clean_p}\n" f"• 🔑 Hash:{full_h}\n" f"• 👤 Author: {author} <{email}>\n" f"• 📅 Date: {date_str} ({time_rel})\n" f"• 📝 Subject:\n{subject}" f"{link_html}" f"{stat_block_en}\n\n" f"{status_tag_en}" f"⚠️ Note: Resetting will restore project files to this state. Checkout lets you inspect this snapshot without affecting branch history." )