659 lines
28 KiB
Python
659 lines
28 KiB
Python
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"<code>{escape_html(p_name)} {escape_html(m_label)}</code>"
|
|
line2 = f"<code>CL:{pct_str} Ti:{ti_str} To:{to_str} D:{dur_str}</code>"
|
|
|
|
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"💭 <b>تفکر و استدلال (مدل: {badge}):</b>" if is_fa else f"💭 <b>Reasoning / Thinking (Model: {badge}):</b>"
|
|
else:
|
|
header = f"💭 <b>تفکر و استدلال ({badge}):</b>" if is_fa else f"💭 <b>Reasoning / Thinking ({badge}):</b>"
|
|
else:
|
|
header = "💭 <b>تفکر و استدلال:</b>" if is_fa else "💭 <b>Reasoning / Thinking:</b>"
|
|
return f"<blockquote expandable>{header}\n{escaped}</blockquote>\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"<b>{label}:</b> <code>{escape_html(tool_name)} {tool_emoji}</code>{action_text}"
|
|
if target:
|
|
out += f"\n ↳ <code>{target}</code>"
|
|
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"<b>\1</b>", 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"<blockquote>\1</blockquote>", processed, flags=re.MULTILINE)
|
|
processed = re.sub(r"</blockquote>\n<blockquote>", "\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"<b>\1</b>", processed)
|
|
processed = re.sub(r"__(.+?)__", r"<b>\1</b>", processed)
|
|
|
|
# 9. Italic: *text* or _text_ (ensure not touching word internals or placeholders)
|
|
processed = re.sub(r"(?<!\w)\*([^\*\n]+)\*(?!\w)", r"<i>\1</i>", processed)
|
|
processed = re.sub(r"(?<!\w)_([^_\n]+)_(?!\w)", r"<i>\1</i>", processed)
|
|
|
|
# 10. Strikethrough: ~~text~~
|
|
processed = re.sub(r"~~(.+?)~~", r"<s>\1</s>", 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'<a href="{url}">{label}</a>'
|
|
else:
|
|
return f"<code>{label}</code>"
|
|
|
|
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"<code>{escaped_code}</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"<pre><code{lang_attr}>{escaped_code}</code></pre>"
|
|
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"<code>{c_hash}</code> - {c_msg} (<i>{c_time}</i>)"
|
|
|
|
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"🐙 <b>مدیریت مخزن گیت (Gitea Version Control)</b>\n\n"
|
|
f"• 📁 <b>پروژه:</b> <code>{clean_p}</code>\n"
|
|
f"• 🌐 <b>مخزن تحت وب (Gitea):</b>\n<a href=\"{web_url}\">{web_url}</a>\n"
|
|
f"• 📥 <b>دستور کلون در سیستم شخصی:</b>\n<code>git clone {clone_url}</code>\n\n"
|
|
f"• 🌿 <b>شاخه فعال:</b> <code>{escape_html(branch)}</code>\n"
|
|
f"• 🔖 <b>آخرین کامیت:</b> {commit_str}\n"
|
|
f"• 📊 <b>وضعیت تغییرات محلی:</b> {dirty_str}\n\n"
|
|
f"💡 <i>تمامی تغییرات اعمالشده توسط هوش مصنوعی به صورت خودکار کامیت و در مخزن Gitea پوش میشوند.</i>"
|
|
)
|
|
else:
|
|
return (
|
|
f"🐙 <b>Git Repository Manager (Gitea)</b>\n\n"
|
|
f"• 📁 <b>Project:</b> <code>{clean_p}</code>\n"
|
|
f"• 🌐 <b>Web Repository (Gitea):</b>\n<a href=\"{web_url}\">{web_url}</a>\n"
|
|
f"• 📥 <b>Clone Command:</b>\n<code>git clone {clone_url}</code>\n\n"
|
|
f"• 🌿 <b>Active Branch:</b> <code>{escape_html(branch)}</code>\n"
|
|
f"• 🔖 <b>Last Commit:</b> {commit_str}\n"
|
|
f"• 📊 <b>Local Status:</b> {dirty_str_en}\n\n"
|
|
f"💡 <i>All changes made by AI are automatically committed and pushed to your Gitea repo.</i>"
|
|
)
|
|
|
|
|
|
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"📜 <b>تاریخچه کامیتهای پروژه <code>{clean_p}</code></b>\n\n⚠️ هیچ کامیتی در این مخزن یافت نشد."
|
|
else:
|
|
return f"📜 <b>Commit History for <code>{clean_p}</code></b>\n\n⚠️ No commits found in this repository."
|
|
|
|
if is_fa:
|
|
lines = [
|
|
f"📜 <b>تاریخچه ۲۰ کامیت اخیر (پروژه: <code>{clean_p}</code>)</b>",
|
|
f"🌿 <b>شاخه/موقعیت فعلی:</b> <code>{escape_html(branch)}</code>",
|
|
f"📄 <b>صفحه:</b> {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}• <code>{short_h}</code> - <b>{subject}</b>\n 👤 <i>{author}</i> ({time_rel})")
|
|
|
|
lines.append("\n💡 <i>جهت مشاهده جزئیات، جابهجایی یا بازگشت به هر کامیت، دکمه مربوطه را انتخاب نمایید:</i>")
|
|
return "\n".join(lines)
|
|
else:
|
|
lines = [
|
|
f"📜 <b>Recent Commits (Project: <code>{clean_p}</code>)</b>",
|
|
f"🌿 <b>Active Branch/Ref:</b> <code>{escape_html(branch)}</code>",
|
|
f"📄 <b>Page:</b> {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}• <code>{short_h}</code> - <b>{subject}</b>\n 👤 <i>{author}</i> ({time_rel})")
|
|
|
|
lines.append("\n💡 <i>Select a commit button below to inspect, checkout, or reset:</i>")
|
|
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• 🌐 <b>مشاهده در وب:</b> <a href=\"{commit_link}\">{commit_link}</a>" if commit_link else ""
|
|
|
|
stat_block = f"\n\n📊 <b>فایلهای تغییریافته:</b>\n<pre>{stat}</pre>" if stat else ""
|
|
stat_block_en = f"\n\n📊 <b>Changed Files:</b>\n<pre>{stat}</pre>" if stat else ""
|
|
|
|
status_tag = "📍 <b>این کامیت هماکنون کامیت فعال (HEAD) پروژه است.</b>\n" if is_curr else ""
|
|
status_tag_en = "📍 <b>This commit is currently the active HEAD of the project.</b>\n" if is_curr else ""
|
|
|
|
if is_fa:
|
|
return (
|
|
f"🔖 <b>مشخصات و جزئیات کامیت گیت</b>\n\n"
|
|
f"• 📁 <b>پروژه:</b> <code>{clean_p}</code>\n"
|
|
f"• 🔑 <b>شناسه کامیت:</b> <code>{full_h}</code>\n"
|
|
f"• 👤 <b>نویسنده:</b> {author} <{email}>\n"
|
|
f"• 📅 <b>زمان:</b> {date_str} (<i>{time_rel}</i>)\n"
|
|
f"• 📝 <b>پیام کامیت:</b>\n<blockquote><b>{subject}</b></blockquote>"
|
|
f"{link_html}"
|
|
f"{stat_block}\n\n"
|
|
f"{status_tag}"
|
|
f"⚠️ <i>توجه: با بازگردانی (Reset)، وضعیت پروژه به این کامیت بازمیگردد. با سوییچ (Checkout) میتوانید بدون تغییر شاخه اصلی، وضعیت این نسخه را مشاهده یا تست کنید.</i>"
|
|
)
|
|
else:
|
|
return (
|
|
f"🔖 <b>Git Commit Details</b>\n\n"
|
|
f"• 📁 <b>Project:</b> <code>{clean_p}</code>\n"
|
|
f"• 🔑 <b>Hash:</b> <code>{full_h}</code>\n"
|
|
f"• 👤 <b>Author:</b> {author} <{email}>\n"
|
|
f"• 📅 <b>Date:</b> {date_str} (<i>{time_rel}</i>)\n"
|
|
f"• 📝 <b>Subject:</b>\n<blockquote><b>{subject}</b></blockquote>"
|
|
f"{link_html}"
|
|
f"{stat_block_en}\n\n"
|
|
f"{status_tag_en}"
|
|
f"⚠️ <i>Note: Resetting will restore project files to this state. Checkout lets you inspect this snapshot without affecting branch history.</i>"
|
|
)
|
|
|
|
|
|
def format_ftp_info(
|
|
project_name: str,
|
|
host: Optional[str],
|
|
port: int = 21,
|
|
user: Optional[str] = None,
|
|
path: str = "/",
|
|
tls: bool = False,
|
|
lang: str = "fa",
|
|
) -> str:
|
|
"""Formats FTP connection and deployment status for Telegram HTML."""
|
|
is_fa = (lang or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)")
|
|
clean_p = escape_html(project_name)
|
|
host_str = f"<code>{escape_html(host)}:{port}</code>" if host else ("<i>(تنظیم نشده)</i>" if is_fa else "<i>(Not configured)</i>")
|
|
user_str = f"<code>{escape_html(user)}</code>" if user else ("<i>(ندارد)</i>" if is_fa else "<i>(None)</i>")
|
|
path_str = f"<code>{escape_html(path)}</code>"
|
|
tls_str = "بله (FTPS/TLS امن) 🔒" if tls else "خیر (FTP معمولی)"
|
|
tls_str_en = "Yes (Secure FTPS/TLS) 🔒" if tls else "No (Standard FTP)"
|
|
|
|
if is_fa:
|
|
return (
|
|
f"🚀 <b>تنظیمات استقرار و دیپلوی FTP پروژه</b>\n\n"
|
|
f"• 📁 <b>پروژه:</b> <code>{clean_p}</code>\n"
|
|
f"• 🌐 <b>سرور / هاست FTP:</b> {host_str}\n"
|
|
f"• 👤 <b>نام کاربری:</b> {user_str}\n"
|
|
f"• 📂 <b>مسیر مقصد روی هاست:</b> {path_str}\n"
|
|
f"• 🔒 <b>پروتکل امن (FTPS):</b> {tls_str}\n\n"
|
|
f"💡 <i>در زمان دیپلوی، هوش مصنوعی فایلهای زائد، کش، سشنها و مخزن گیت (.git) را به صورت خودکار فیلتر کرده و تنها سورس اصلی و تمیز به سرور منتقل میشود.</i>"
|
|
)
|
|
else:
|
|
return (
|
|
f"🚀 <b>Project FTP Deployment Settings</b>\n\n"
|
|
f"• 📁 <b>Project:</b> <code>{clean_p}</code>\n"
|
|
f"• 🌐 <b>FTP Host:</b> {host_str}\n"
|
|
f"• 👤 <b>Username:</b> {user_str}\n"
|
|
f"• 📂 <b>Remote Target Path:</b> {path_str}\n"
|
|
f"• 🔒 <b>Secure FTPS/TLS:</b> {tls_str_en}\n\n"
|
|
f"💡 <i>During deployment, unnecessary cache, temp, sessions, and .git files are automatically excluded for a clean production upload.</i>"
|
|
)
|
|
|
|
|
|
|