kobalt/gotelegram_pro
1
0
Fork 0
mirror of https://github.com/anten-ka/gotelegram_pro.git synced 2026-05-19 16:46:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

v2.3.0: Lite/Pro rebrand, submenu system, traffic stats, bot stats

Browse Source
This commit is contained in:
anten-ka
2026-04-08 21:49:03 +03:00
parent 364501d66d
commit 6ec2123f83
11 changed files with 884 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

259
gotelegram-bot/bot.py
View File

@@ -6,14 +6,17 @@ Uses python-telegram-bot v21+
"""
import asyncio
import csv
import html
import json
import logging
import os
import re
import subprocess
import time
import toml
from datetime import datetime
from io import StringIO
from pathlib import Path
from typing import Tuple, Optional, List, Dict, Any
@@ -47,7 +50,7 @@ logger = logging.getLogger(__name__)
# CONFIGURATION
# ============================================================================
GOTELEGRAM_VERSION = "2.2.0"
GOTELEGRAM_VERSION = "2.3.0"
GOTELEGRAM_CONFIG = "/opt/gotelegram/config.json"
TELEMT_CONFIG = "/etc/telemt/config.toml"
TELEMT_SERVICE = "telemt"
@@ -69,7 +72,7 @@ for _id_str in ALLOWED_IDS_STR.split(","):
except ValueError:
logging.warning(f"Invalid ALLOWED_IDS entry: {_id_str}")
QUICK_DOMAINS = [
LITE_DOMAINS = [
"google.com",
"microsoft.com",
"cloudflare.com",
@@ -277,10 +280,13 @@ def get_main_menu() -> InlineKeyboardMarkup:
InlineKeyboardButton("🎁 Promo", callback_data="menu_promo"),
],
[
InlineKeyboardButton("📊 Traffic Stats", callback_data="menu_stats"),
InlineKeyboardButton("🗑️ Remove", callback_data="menu_remove"),
InlineKeyboardButton("ℹ️ Credits", callback_data="menu_credits"),
],
[InlineKeyboardButton("❌ Close", callback_data="close_menu")],
[
InlineKeyboardButton("ℹ️ Credits", callback_data="menu_credits"),
InlineKeyboardButton("❌ Close", callback_data="close_menu"),
],
]
return InlineKeyboardMarkup(buttons)
@@ -403,19 +409,135 @@ async def get_status_text() -> str:
return "\n".join(lines)
async def cb_menu_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Status callback."""
async def get_traffic_stats() -> str:
"""Get formatted traffic statistics."""
# Read current snapshot
current_file = "/run/gotelegram/stats_current.json"
history_file = "/opt/gotelegram/stats_history.csv"
try:
with open(current_file, "r") as f:
current = json.load(f)
except Exception:
return "📊 <b>Статистика</b>\n\n<i>Данные недоступны. Убедитесь что модуль статистики включён.</i>"
# Read history
history = []
try:
with open(history_file, "r") as f:
reader = csv.reader(f)
for row in reader:
if len(row) >= 3:
history.append({
"ts": int(row[0]),
"proxy": int(row[1]),
"site": int(row[2]),
})
except Exception:
pass
now = int(time.time())
def format_bytes(b):
if b < 1024:
return f"{b} B"
if b < 1048576:
return f"{b/1024:.1f} KB"
if b < 1073741824:
return f"{b/1048576:.1f} MB"
return f"{b/1073741824:.1f} GB"
def format_rate(bps):
if bps < 1024:
return f"{bps:.0f} B/s"
if bps < 1048576:
return f"{bps/1024:.1f} KB/s"
return f"{bps/1048576:.1f} MB/s"
def calc_for_period(secs, key):
target_ts = now - secs
# Find closest snapshot to target_ts
closest = None
for h in history:
if h["ts"] <= target_ts:
if closest is None or h["ts"] > closest["ts"]:
closest = h
if closest is None:
return "—", "—"
current_val = current.get(f"{key}_bytes", 0)
diff = current_val - closest[key]
if diff < 0:
diff = 0
elapsed = now - closest["ts"]
if elapsed <= 0:
elapsed = 1
rate = diff / elapsed
return format_bytes(diff), format_rate(rate)
periods = [
("1 мин", 60),
("5 мин", 300),
("60 мин", 3600),
("1 день", 86400),
("7 дней", 604800),
("30 дней", 2592000),
("365 дней", 31536000),
]
lines = ["📊 <b>Статистика трафика</b>\n"]
for label, key in [("Proxy (telemt)", "proxy"), ("Сайт (nginx)", "site")]:
lines.append(f"\n<b>{label}:</b>")
lines.append("<pre>")
lines.append(f"{'Период':<10} │ {'Трафик':>10} │ {'Скорость':>10}")
lines.append("─" * 36)
for name, secs in periods:
total, rate = calc_for_period(secs, key)
lines.append(f"{name:<10} │ {total:>10} │ {rate:>10}")
lines.append("</pre>")
return "\n".join(lines)
async def cb_menu_stats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Show traffic statistics."""
query = update.callback_query
await query.answer()
await safe_edit_message(query,"⏳ Checking status...")
stats_text = await get_traffic_stats()
status_text = await get_status_text()
keyboard = InlineKeyboardMarkup(
[[InlineKeyboardButton("« Back", callback_data="menu_main")]]
keyboard = [
[InlineKeyboardButton("🔄 Обновить", callback_data="menu_stats")],
[InlineKeyboardButton("« Меню", callback_data="menu_main")],
]
await safe_edit_message(
query,
stats_text,
reply_markup=InlineKeyboardMarkup(keyboard),
parse_mode="HTML",
)
await safe_edit_message(query,
status_text, reply_markup=keyboard, parse_mode="HTML"
async def cb_menu_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Status callback — show detailed proxy/server status."""
query = update.callback_query
await query.answer()
if not await require_auth(update, context):
return
text = await get_status_text()
keyboard = [
[InlineKeyboardButton("🔄 Обновить", callback_data="menu_status")],
[InlineKeyboardButton("« Меню", callback_data="menu_main")],
]
await safe_edit_message(
query,
text,
reply_markup=InlineKeyboardMarkup(keyboard),
parse_mode="HTML",
)
@@ -427,8 +549,8 @@ async def cb_menu_status(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
def get_install_mode_menu() -> InlineKeyboardMarkup:
"""Install mode selection menu."""
buttons = [
[InlineKeyboardButton("⚡ Quick Mode", callback_data="install_mode_quick")],
[InlineKeyboardButton("🔒 Stealth Mode", callback_data="install_mode_stealth")],
[InlineKeyboardButton("⚡ Lite", callback_data="install_mode_lite")],
[InlineKeyboardButton("🛡 Pro", callback_data="install_mode_pro")],
[InlineKeyboardButton("« Back", callback_data="menu_main")],
]
return InlineKeyboardMarkup(buttons)
@@ -452,7 +574,7 @@ async def cb_menu_install(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
)
buttons = [
[InlineKeyboardButton("🔄 Migrate from v1", callback_data="install_migrate")],
[InlineKeyboardButton("✨ Fresh Install", callback_data="install_mode_quick")],
[InlineKeyboardButton("✨ Fresh Install", callback_data="install_mode_lite")],
[InlineKeyboardButton("« Back", callback_data="menu_main")],
]
keyboard = InlineKeyboardMarkup(buttons)
@@ -465,39 +587,39 @@ async def cb_menu_install(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
)
async def cb_install_mode_quick(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Quick mode domain selection."""
async def cb_install_mode_lite(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Lite mode domain selection."""
query = update.callback_query
await query.answer()
# Show domains with pagination (4 per row, 2 rows)
buttons = []
for i in range(0, len(QUICK_DOMAINS), 2):
for i in range(0, len(LITE_DOMAINS), 2):
row = []
for j in range(2):
if i + j < len(QUICK_DOMAINS):
domain = QUICK_DOMAINS[i + j]
if i + j < len(LITE_DOMAINS):
domain = LITE_DOMAINS[i + j]
row.append(
InlineKeyboardButton(
domain, callback_data=f"quick_dom_{i+j}"
domain, callback_data=f"lite_dom_{i+j}"
)
)
buttons.append(row)
buttons.append([InlineKeyboardButton("« Back", callback_data="menu_install")])
text = "Select a domain for quick mode:"
text = "Select a domain for Lite mode:"
keyboard = InlineKeyboardMarkup(buttons)
await safe_edit_message(query,text, reply_markup=keyboard)
async def cb_quick_domain(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Quick domain selection callback."""
async def cb_lite_domain(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Lite domain selection callback."""
query = update.callback_query
data = query.data
try:
domain_idx = int(data.split("_")[-1])
domain = QUICK_DOMAINS[domain_idx]
domain = LITE_DOMAINS[domain_idx]
except (ValueError, IndexError):
await query.answer("Invalid domain selection")
return
@@ -507,7 +629,7 @@ async def cb_quick_domain(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
# Simulate installation (in real scenario, call install script)
config = {
"mode": "quick",
"mode": "lite",
"domain": domain,
"port": 443,
"installed_at": datetime.now().isoformat(),
@@ -515,9 +637,9 @@ async def cb_quick_domain(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
if save_json(GOTELEGRAM_CONFIG, config):
text = (
f"✅ <b>Quick mode installed!</b>\n\n"
f"✅ <b>Lite mode installed!</b>\n\n"
f"<b>Domain:</b> {domain}\n"
f"<b>Mode:</b> Quick\n\n"
f"<b>Mode:</b> Lite\n\n"
f"Service starting... Check status in 10 seconds."
)
keyboard = InlineKeyboardMarkup(
@@ -535,8 +657,8 @@ async def cb_quick_domain(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
)
async def cb_install_mode_stealth(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Stealth mode - show template categories."""
async def cb_install_mode_pro(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Pro mode - show template categories."""
query = update.callback_query
await query.answer()
@@ -555,22 +677,22 @@ async def cb_install_mode_stealth(update: Update, context: ContextTypes.DEFAULT_
buttons.append(
[
InlineKeyboardButton(
f"📁 {cat['name']}", callback_data=f"stealth_cat_{cat['id']}"
f"📁 {cat['name']}", callback_data=f"pro_cat_{cat['id']}"
)
]
)
buttons.append([InlineKeyboardButton("« Back", callback_data="menu_install")])
text = "Stealth Mode - Select Template Category:"
text = "Pro Mode - Select Template Category:"
keyboard = InlineKeyboardMarkup(buttons)
await safe_edit_message(query,text, reply_markup=keyboard)
async def cb_stealth_category(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
async def cb_pro_category(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Show templates in category."""
query = update.callback_query
data = query.data
cat_id = data.removeprefix("stealth_cat_")
cat_id = data.removeprefix("pro_cat_")
await query.answer()
@@ -597,22 +719,22 @@ async def cb_stealth_category(update: Update, context: ContextTypes.DEFAULT_TYPE
buttons.append(
[
InlineKeyboardButton(
f"🎨 {tpl['name']}", callback_data=f"stealth_tpl_{tpl['id']}"
f"🎨 {tpl['name']}", callback_data=f"pro_tpl_{tpl['id']}"
)
]
)
buttons.append([InlineKeyboardButton("« Back", callback_data="install_mode_stealth")])
buttons.append([InlineKeyboardButton("« Back", callback_data="install_mode_pro")])
text = f"Select template from <b>{html.escape(category['name'])}</b>:"
keyboard = InlineKeyboardMarkup(buttons)
await safe_edit_message(query,text, reply_markup=keyboard, parse_mode="HTML")
async def cb_stealth_template(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
async def cb_pro_template(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Show template preview and confirm."""
query = update.callback_query
data = query.data
tpl_id = data.removeprefix("stealth_tpl_")
tpl_id = data.removeprefix("pro_tpl_")
await query.answer()
@@ -651,26 +773,26 @@ async def cb_stealth_template(update: Update, context: ContextTypes.DEFAULT_TYPE
buttons = [
[
InlineKeyboardButton(
"✅ Install", callback_data=f"stealth_confirm_{tpl_id}"
"✅ Install", callback_data=f"pro_confirm_{tpl_id}"
)
],
[InlineKeyboardButton("« Back", callback_data="install_mode_stealth")],
[InlineKeyboardButton("« Back", callback_data="install_mode_pro")],
]
keyboard = InlineKeyboardMarkup(buttons)
await safe_edit_message(query,text, reply_markup=keyboard, parse_mode="HTML")
async def cb_stealth_confirm(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Confirm and install stealth template."""
async def cb_pro_confirm(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Confirm and install pro template."""
query = update.callback_query
data = query.data
tpl_id = data.removeprefix("stealth_confirm_")
tpl_id = data.removeprefix("pro_confirm_")
await query.answer()
await safe_edit_message(query,"⏳ Installing template...")
config = {
"mode": "stealth",
"mode": "pro",
"template": tpl_id,
"port": 443,
"installed_at": datetime.now().isoformat(),
@@ -678,9 +800,9 @@ async def cb_stealth_confirm(update: Update, context: ContextTypes.DEFAULT_TYPE)
if save_json(GOTELEGRAM_CONFIG, config):
text = (
f"✅ <b>Stealth mode installed!</b>\n\n"
f"✅ <b>Pro mode installed!</b>\n\n"
f"<b>Template:</b> {html.escape(tpl_id)}\n"
f"<b>Mode:</b> Stealth\n\n"
f"<b>Mode:</b> Pro\n\n"
f"Service starting... Check status in 10 seconds."
)
keyboard = InlineKeyboardMarkup(
@@ -1074,8 +1196,8 @@ async def cb_menu_change(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
await query.answer()
buttons = [
[InlineKeyboardButton("⚡ Switch to Quick Mode", callback_data="change_quick")],
[InlineKeyboardButton("🔒 Switch to Stealth Mode", callback_data="change_stealth")],
[InlineKeyboardButton("⚡ Switch to Lite Mode", callback_data="change_lite")],
[InlineKeyboardButton("🛡 Switch to Pro Mode", callback_data="change_pro")],
[InlineKeyboardButton("« Back", callback_data="menu_main")],
]
keyboard = InlineKeyboardMarkup(buttons)
@@ -1084,20 +1206,20 @@ async def cb_menu_change(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
)
async def cb_change_quick(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Switch to quick mode — show domain selection."""
async def cb_change_lite(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Switch to lite mode — show domain selection."""
query = update.callback_query
await query.answer()
# Reuse the quick mode domain selection flow
await cb_install_mode_quick(update, context)
# Reuse the lite mode domain selection flow
await cb_install_mode_lite(update, context)
async def cb_change_stealth(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Switch to stealth mode — show template categories."""
async def cb_change_pro(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Switch to pro mode — show template categories."""
query = update.callback_query
await query.answer()
# Reuse the stealth mode template selection flow
await cb_install_mode_stealth(update, context)
# Reuse the pro mode template selection flow
await cb_install_mode_pro(update, context)
async def cb_install_migrate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
@@ -1328,27 +1450,28 @@ async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
"menu_promo": cb_menu_promo,
"menu_credits": cb_menu_credits,
"menu_remove": cb_menu_remove,
"install_mode_quick": cb_install_mode_quick,
"install_mode_stealth": cb_install_mode_stealth,
"install_mode_lite": cb_install_mode_lite,
"install_mode_pro": cb_install_mode_pro,
"backup_create": cb_backup_create,
"backup_list": cb_backup_list,
"ssl_renew": cb_ssl_renew,
"ssl_status": cb_ssl_status,
"remove_confirm": cb_remove_confirm,
"change_quick": cb_change_quick,
"change_stealth": cb_change_stealth,
"change_lite": cb_change_lite,
"change_pro": cb_change_pro,
"install_migrate": cb_install_migrate,
"menu_stats": cb_menu_stats,
}
# Pattern-based handlers
if data.startswith("quick_dom_"):
await cb_quick_domain(update, context)
elif data.startswith("stealth_cat_"):
await cb_stealth_category(update, context)
elif data.startswith("stealth_tpl_"):
await cb_stealth_template(update, context)
elif data.startswith("stealth_confirm_"):
await cb_stealth_confirm(update, context)
if data.startswith("lite_dom_"):
await cb_lite_domain(update, context)
elif data.startswith("pro_cat_"):
await cb_pro_category(update, context)
elif data.startswith("pro_tpl_"):
await cb_pro_template(update, context)
elif data.startswith("pro_confirm_"):
await cb_pro_confirm(update, context)
elif data.startswith("restore_idx_"):
await cb_restore_backup(update, context)
elif data in handlers: