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

v2.3.1: admin management (auto-register first admin, /addadmin, /deladmin, menu button)

Browse Source
This commit is contained in:
anten-ka
2026-04-09 10:46:50 +03:00
parent a21d2ebea2
commit 046a08fdb6
2 changed files with 290 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

305
gotelegram-bot/bot.py
View File

@@ -64,15 +64,67 @@ TIP_LINK = "https://pay.cloudtips.ru/p/7410814f"
PROMO_STAMP_FILE = "/opt/gotelegram/.promo_bot_last_shown"
BOT_TOKEN = os.getenv("BOT_TOKEN")
ALLOWED_IDS_STR = os.getenv("ALLOWED_IDS", "")
ENV_FILE = "/opt/gotelegram-bot/.env"
# ── Загрузка ALLOWED_IDS ────────────────────────────────────────────────────
# Поддерживает запятую, пробел, или их комбинацию как разделитель
ALLOWED_IDS: set = set()
for _id_str in ALLOWED_IDS_STR.split(","):
_id_str = _id_str.strip()
if _id_str:
try:
ALLOWED_IDS.add(int(_id_str))
except ValueError:
logging.warning(f"Invalid ALLOWED_IDS entry: {_id_str}")
_WAITING_FOR_ADMIN = False # True если список пуст → ждём первого админа
def _load_allowed_ids() -> None:
"""Загрузить ALLOWED_IDS из переменной окружения."""
global ALLOWED_IDS, _WAITING_FOR_ADMIN
raw = os.getenv("ALLOWED_IDS", "")
ALLOWED_IDS = set()
# Разделители: запятая, пробел, или оба
for part in re.split(r'[,\s]+', raw):
part = part.strip()
if part:
try:
ALLOWED_IDS.add(int(part))
except ValueError:
logging.warning(f"Invalid ALLOWED_IDS entry: {part}")
_WAITING_FOR_ADMIN = len(ALLOWED_IDS) == 0
def _save_allowed_ids() -> None:
"""Сохранить ALLOWED_IDS в .env файл и обновить os.environ."""
global _WAITING_FOR_ADMIN
ids_str = ",".join(str(i) for i in sorted(ALLOWED_IDS))
os.environ["ALLOWED_IDS"] = ids_str
_WAITING_FOR_ADMIN = len(ALLOWED_IDS) == 0
if not os.path.exists(ENV_FILE):
return
try:
with open(ENV_FILE, "r") as f:
lines = f.readlines()
found = False
new_lines = []
for line in lines:
if line.strip().startswith("ALLOWED_IDS="):
if ids_str:
new_lines.append(f"ALLOWED_IDS={ids_str}\n")
# Если пусто — удаляем строку
found = True
else:
new_lines.append(line)
if not found and ids_str:
new_lines.append(f"ALLOWED_IDS={ids_str}\n")
with open(ENV_FILE, "w") as f:
f.writelines(new_lines)
logger.info(f"ALLOWED_IDS updated in .env: {ids_str or '(empty)'}")
except OSError as e:
logger.error(f"Failed to update .env: {e}")
_load_allowed_ids()
LITE_DOMAINS = [
"google.com",
@@ -230,21 +282,41 @@ async def check_old_container() -> Optional[str]:
def is_user_allowed(user_id: int) -> bool:
"""Check if user ID is in ALLOWED_IDS."""
if not ALLOWED_IDS:
return True
"""Check if user ID is in ALLOWED_IDS. If list is empty — waiting for admin."""
if _WAITING_FOR_ADMIN:
return False # Никому не даём доступ пока не назначен админ
return user_id in ALLOWED_IDS
def add_admin(user_id: int) -> None:
"""Добавить администратора и сохранить в .env."""
ALLOWED_IDS.add(user_id)
_save_allowed_ids()
logger.info(f"Admin added: {user_id}")
def remove_admin(user_id: int) -> None:
"""Убрать администратора и сохранить в .env."""
ALLOWED_IDS.discard(user_id)
_save_allowed_ids()
logger.info(f"Admin removed: {user_id}")
async def require_auth(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
"""Check authorization and send error if not allowed."""
if not is_user_allowed(update.effective_user.id):
await update.message.reply_text(
f"Access denied. Your ID: {update.effective_user.id}"
)
logger.warning(
f"Unauthorized access attempt from user {update.effective_user.id}"
)
user_id = update.effective_user.id
# Режим ожидания первого админа — обрабатывается в cmd_start
if _WAITING_FOR_ADMIN:
return False
if not is_user_allowed(user_id):
if update.message:
await update.message.reply_text(
f"⛔ Доступ запрещён.\nВаш ID: <code>{user_id}</code>",
parse_mode="HTML",
)
logger.warning(f"Unauthorized access attempt from user {user_id}")
return False
return True
@@ -286,7 +358,10 @@ def get_main_menu() -> InlineKeyboardMarkup:
InlineKeyboardButton("🗑️ Remove", callback_data="menu_remove"),
],
[
InlineKeyboardButton("👤 Админы", callback_data="menu_admins"),
InlineKeyboardButton("ℹ️ Credits", callback_data="menu_credits"),
],
[
InlineKeyboardButton("❌ Close", callback_data="close_menu"),
],
]
@@ -299,8 +374,37 @@ def get_main_menu() -> InlineKeyboardMarkup:
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Start command - show main menu, promo once per day."""
if not await require_auth(update, context):
"""Start command - show main menu, promo once per day.
Если ALLOWED_IDS пуст — режим авто-регистрации первого админа.
"""
user = update.effective_user
user_id = user.id
# ── Режим ожидания первого админа ──
if _WAITING_FOR_ADMIN:
name = user.full_name or user.username or str(user_id)
text = (
f"<b>👋 Привет, {html.escape(name)}!</b>\n\n"
f"Бот ещё не настроен.\n"
f"Ваш Telegram ID: <code>{user_id}</code>\n\n"
f"Назначить вас администратором?"
)
keyboard = InlineKeyboardMarkup([
[
InlineKeyboardButton("✅ Да", callback_data=f"admin_confirm_{user_id}"),
InlineKeyboardButton("❌ Нет", callback_data="admin_cancel"),
]
])
await update.message.reply_text(text, reply_markup=keyboard, parse_mode="HTML")
return
# ── Проверка доступа ──
if not is_user_allowed(user_id):
await update.message.reply_text(
f"⛔ Доступ запрещён.\nВаш ID: <code>{user_id}</code>",
parse_mode="HTML",
)
return
welcome = (
@@ -327,12 +431,14 @@ async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
return
help_text = (
"<b>GoTelegram Bot Commands</b>\n\n"
"/start - Show main menu\n"
"/help - Show this help message\n"
"/status - Quick status check\n"
"/logs - Show recent logs\n\n"
"Use the inline menu for all other operations."
"<b>GoTelegram Bot — Команды</b>\n\n"
"/start — Главное меню\n"
"/help — Эта справка\n"
"/status — Быстрый статус\n"
"/logs — Последние логи\n"
"/addadmin ID — Добавить админа\n"
"/deladmin ID — Удалить админа\n\n"
"Используйте кнопки меню для остальных операций."
)
await update.message.reply_text(help_text, parse_mode="HTML")
@@ -1322,6 +1428,114 @@ async def cb_ssl_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N
await safe_edit_message(query,text, reply_markup=keyboard, parse_mode="HTML")
# ============================================================================
# ADMIN MANAGEMENT
# ============================================================================
async def cb_menu_admins(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Показать список админов и кнопки управления."""
query = update.callback_query
await query.answer()
if ALLOWED_IDS:
ids_list = "\n".join(f" • <code>{uid}</code>" for uid in sorted(ALLOWED_IDS))
text = f"<b>👤 Администраторы</b>\n\n{ids_list}\n"
else:
text = "<b>👤 Администраторы</b>\n\n<i>Список пуст — доступ для всех</i>\n"
text += (
f"\nВсего: {len(ALLOWED_IDS)}\n\n"
"Чтобы <b>добавить</b> — перешлите любое сообщение от нового админа, "
"или отправьте команду:\n"
"<code>/addadmin 123456789</code>\n\n"
"Чтобы <b>удалить</b>:\n"
"<code>/deladmin 123456789</code>"
)
keyboard = InlineKeyboardMarkup([
[InlineKeyboardButton("« Назад", callback_data="menu_main")],
])
await safe_edit_message(query, text, reply_markup=keyboard, parse_mode="HTML")
async def cmd_addadmin(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""/addadmin ID [ID2 ID3 ...] — добавить админа вручную."""
if not is_user_allowed(update.effective_user.id):
await update.message.reply_text(
f"⛔ Доступ запрещён.\nВаш ID: <code>{update.effective_user.id}</code>",
parse_mode="HTML",
)
return
args = context.args or []
if not args:
await update.message.reply_text(
"Использование: <code>/addadmin ID [ID2 ID3 ...]</code>\n"
"Пример: <code>/addadmin 123456789 987654321</code>",
parse_mode="HTML",
)
return
added = []
errors = []
for a in args:
a = a.strip().replace(",", "")
if not a:
continue
try:
uid = int(a)
add_admin(uid)
added.append(str(uid))
except ValueError:
errors.append(a)
parts = []
if added:
parts.append(f"✅ Добавлены: {', '.join(added)}")
if errors:
parts.append(f"❌ Ошибки: {', '.join(errors)}")
await update.message.reply_text("\n".join(parts), parse_mode="HTML")
async def cmd_deladmin(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""/deladmin ID — удалить админа."""
if not is_user_allowed(update.effective_user.id):
await update.message.reply_text(
f"⛔ Доступ запрещён.\nВаш ID: <code>{update.effective_user.id}</code>",
parse_mode="HTML",
)
return
args = context.args or []
if not args:
await update.message.reply_text(
"Использование: <code>/deladmin ID</code>",
parse_mode="HTML",
)
return
removed = []
for a in args:
a = a.strip().replace(",", "")
try:
uid = int(a)
if uid == update.effective_user.id:
await update.message.reply_text("⚠️ Нельзя удалить себя!")
continue
if uid in ALLOWED_IDS:
remove_admin(uid)
removed.append(str(uid))
else:
await update.message.reply_text(f"ID {uid} не найден в списке")
except ValueError:
await update.message.reply_text(f"❌ Некорректный ID: {html.escape(a)}")
if removed:
await update.message.reply_text(f"✅ Удалены: {', '.join(removed)}")
# ============================================================================
# PROMO & CREDITS
# ============================================================================
@@ -1463,9 +1677,43 @@ async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
query = update.callback_query
data = query.data
# ── Авто-регистрация админа (до проверки доступа!) ──
if data.startswith("admin_confirm_"):
await query.answer()
try:
new_admin_id = int(data.split("_")[-1])
except (ValueError, IndexError):
await safe_edit_message(query, "❌ Ошибка: некорректный ID")
return
# Безопасность: только тот кто нажал кнопку может стать админом
if update.effective_user.id != new_admin_id:
await query.answer("Эта кнопка не для вас", show_alert=True)
return
# Race condition: если кто-то уже стал админом
if not _WAITING_FOR_ADMIN:
await safe_edit_message(query, "ℹ️ Администратор уже назначен.")
return
add_admin(new_admin_id)
await safe_edit_message(
query,
f"✅ <b>Вы назначены администратором!</b>\n\n"
f"ID: <code>{new_admin_id}</code>\n\n"
f"Нажмите /start чтобы открыть меню.",
parse_mode="HTML",
)
return
if data == "admin_cancel":
await query.answer()
await safe_edit_message(
query,
"👋 Ок. Напишите /start когда будете готовы.",
)
return
# Access control
if not is_user_allowed(update.effective_user.id):
await query.answer("Access denied")
await query.answer("Доступ запрещён")
return
# Main menu
@@ -1500,6 +1748,7 @@ async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
"menu_website": cb_menu_website,
"menu_promo": cb_menu_promo,
"menu_credits": cb_menu_credits,
"menu_admins": cb_menu_admins,
"menu_remove": cb_menu_remove,
"install_mode_lite": cb_install_mode_lite,
"install_mode_pro": cb_install_mode_pro,
@@ -1560,6 +1809,8 @@ def main() -> None:
application.add_handler(CommandHandler("help", cmd_help))
application.add_handler(CommandHandler("status", cmd_status))
application.add_handler(CommandHandler("logs", cmd_logs))
application.add_handler(CommandHandler("addadmin", cmd_addadmin))
application.add_handler(CommandHandler("deladmin", cmd_deladmin))
# Callback query handler (buttons)
application.add_handler(CallbackQueryHandler(handle_callback))