FeedBack бот в котором есть почти все что ты хочешь
Тут есть оповещения о оффе или включения бота /on /off
так же есть баз и разбан.
Тут есть оповещения о оффе или включения бота /on /off
так же есть баз и разбан.
Python:
import logging
import asyncio
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters, CommandHandler
import signal
import os
import sys
import time
import sqlite3
BOT_TOKEN = "1"
ADMIN_USER_ID = 1
SHUTDOWN_MESSAGE = "Бот выключен"
RESTART_MESSAGE = "Бот перезапускается..."
START_MESSAGE = "Бот включен и готов к работе."
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
DATABASE_NAME = "bot_data.db"
user_data = {}
admin_message_to_user_id = {}
banned_users = {}
bot_running = True
bot_active = True
bot_instance = None
def create_connection():
conn = None
try:
conn = sqlite3.connect(DATABASE_NAME)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
user_id INTEGER PRIMARY KEY,
username TEXT,
tag TEXT,
notified_shutdown INTEGER,
has_messaged INTEGER
)
""")
conn.commit()
logging.info("Соединение с базой данных установлено и таблица создана (если необходимо).")
except sqlite3.Error as e:
logging.error(f"Ошибка при соединении с базой данных или создании таблицы: {e}")
if conn:
conn.close()
return None
return conn
def load_users_from_db():
conn = create_connection()
if not conn:
return
cursor = conn.cursor()
try:
cursor.execute("SELECT user_id, username, tag, notified_shutdown, has_messaged FROM users")
rows = cursor.fetchall()
for row in rows:
user_id, username, tag, notified_shutdown, has_messaged = row
user_data[user_id] = {"tag": tag, "username": username, "notified_shutdown": notified_shutdown, "has_messaged": has_messaged}
logging.info(f"Загружено {len(rows)} пользователей из базы данных.")
except sqlite3.Error as e:
logging.error(f"Ошибка при загрузке пользователей из базы данных: {e}")
finally:
conn.close()
def save_user_to_db(user_id, username, tag, notified_shutdown=0, has_messaged=0):
conn = create_connection()
if not conn:
return
cursor = conn.cursor()
try:
cursor.execute("INSERT INTO users (user_id, username, tag, notified_shutdown, has_messaged) VALUES (?, ?, ?, ?, ?)",
(user_id, username, tag, notified_shutdown, has_messaged))
conn.commit()
logging.info(f"Пользователь {user_id} ({username}) с тегом '{tag}' сохранен в базу данных.")
except sqlite3.Error as e:
logging.error(f"Ошибка при сохранении пользователя в базу данных: {e}")
finally:
conn.close()
def update_user_notification(user_id, notified_shutdown):
conn = create_connection()
if not conn:
return
cursor = conn.cursor()
try:
cursor.execute("UPDATE users SET notified_shutdown = ? WHERE user_id = ?", (notified_shutdown, user_id))
conn.commit()
logging.info(f"Флаг notified_shutdown пользователя {user_id} установлен в {notified_shutdown} в базе данных.")
except sqlite3.Error as e:
logging.error(f"Ошибка при обновлении статуса уведомления пользователя в базе данных: {e}")
finally:
conn.close()
def update_user_has_messaged(user_id, has_messaged):
conn = create_connection()
if not conn:
return
cursor = conn.cursor()
try:
cursor.execute("UPDATE users SET has_messaged = ? WHERE user_id = ?", (has_messaged, user_id))
conn.commit()
logging.info(f"Флаг has_messaged пользователя {user_id} установлен в {has_messaged} в базе данных.")
except sqlite3.Error as e:
logging.error(f"Ошибка при обновлении статуса has_messaged пользователя в базе данных: {e}")
finally:
conn.close()
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
global bot_running, bot_active
if not bot_running or not bot_active:
return
if update.message:
user_id = update.message.from_user.id
if user_id == ADMIN_USER_ID:
logging.info("Получено сообщение от админа, пропуск обработки пользователя.")
return
if user_id in banned_users:
ban_reason = banned_users[user_id]
try:
await context.bot.send_message(chat_id=user_id, text=f"Вы забанены. Причина: {ban_reason}")
except Exception as e:
logging.error(f"Не удалось отправить сообщение о бане пользователю {user_id}: {e}")
return
user_username = update.message.from_user.username if update.message.from_user.username else update.message.from_user.first_name
message_text = update.message.text
if user_id not in user_data:
tag_number = len(user_data) + 1
user_tag = f"User_{tag_number}"
user_data[user_id] = {"tag": user_tag, "username": user_username, "notified_shutdown": 0, "has_messaged": 1}
save_user_to_db(user_id, user_username, user_tag, has_messaged=1)
else:
user_tag = user_data[user_id]["tag"]
if user_data[user_id]["has_messaged"] == 0:
user_data[user_id]["has_messaged"] = 1
update_user_has_messaged(user_id, 1)
try:
sent_message = await context.bot.send_message(chat_id=ADMIN_USER_ID,
text=f"Сообщение от {user_tag} ({user_username}) : {message_text}")
admin_message_to_user_id[sent_message.message_id] = user_id
except Exception as e:
logging.error(f"Не удалось отправить сообщение админу: {e}")
async def handle_new_chat_members(update: Update, context: ContextTypes.DEFAULT_TYPE):
global bot_running, bot_active
if not bot_running:
return
if update.message.new_chat_members:
for user in update.message.new_chat_members:
if user.id != ADMIN_USER_ID:
try:
if user.id not in user_data:
tag_number = len(user_data) + 1
user_tag = f"User_{tag_number}"
user_data[user.id] = {"tag": user_tag,
"username": user.username if user.username else user.first_name,
"notified_shutdown": 0, "has_messaged": 0}
save_user_to_db(user.id, user.username if user.username else user.first_name, user_tag)
if user.id in banned_users:
await context.bot.send_message(chat_id=user.id, text=f"Вы забанены. Причина: {banned_users[user.id]}")
elif user_data[user.id]["has_messaged"] == 0 and bot_running and bot_active:
await context.bot.send_message(chat_id=user.id, text=SHUTDOWN_MESSAGE)
except Exception as e:
logging.error(f"Не удалось отправить сообщение о присоединении пользователю {user.id}: {e}")
async def handle_admin_reply(update: Update, context: ContextTypes.DEFAULT_TYPE):
global bot_running, bot_active
if not bot_running or not bot_active:
return
if update.message and update.message.from_user.id == ADMIN_USER_ID:
if update.message.reply_to_message:
reply_message_id = update.message.reply_to_message.message_id
if reply_message_id in admin_message_to_user_id:
user_id = admin_message_to_user_id[reply_message_id]
reply_text = update.message.text
try:
await context.bot.send_message(chat_id=user_id, text=reply_text,
reply_to_message_id=reply_message_id)
await context.bot.send_message(chat_id=ADMIN_USER_ID, text=f"Ответ пользователю успешно отправлен.")
except Exception as e:
logging.error(f"Не удалось отправить ответ пользователю: {e}")
await context.bot.send_message(chat_id=ADMIN_USER_ID,
text=f"Не удалось отправить сообщение пользователю.")
else:
await context.bot.send_message(chat_id=ADMIN_USER_ID,
text="Это не ответ на пересланное сообщение от пользователя.")
async def handle_non_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
global bot_running, bot_active
if not bot_running or not bot_active:
return
if update.message and update.message.from_user.id != ADMIN_USER_ID and update.message.text is None:
user_id = update.message.from_user.id
if user_id not in user_data:
tag_number = len(user_data) + 1
user_tag = f"User_{tag_number}"
user_data[user_id] = {"tag": user_tag,
"username": update.message.from_user.username if update.message.from_user.username else update.message.from_user.first_name,
"notified_shutdown": 0, "has_messaged": 0}
save_user_to_db(user_id,
update.message.from_user.username if update.message.from_user.username else update.message.from_user.first_name,
user_tag)
if user_id in banned_users:
try:
await context.bot.send_message(chat_id=user_id, text=f"Вы забанены. Причина: {banned_users[user_id]}")
except Exception as e:
logging.error(f"Не удалось отправить сообщение о бане пользователю {user_id}: {e}")
elif user_data[user_id]["has_messaged"] == 0 and bot_running and bot_active:
try:
await context.bot.send_message(chat_id=user.id, text=SHUTDOWN_MESSAGE)
except Exception as e:
logging.error(f"Не удалось отправить сообщение о нетекстовом сообщении пользователю {user_id}: {e}")
async def reply_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
global bot_running, bot_active
if not bot_running or not bot_active:
return
if update.message and update.message.from_user.id == ADMIN_USER_ID:
args = context.args
if len(args) < 2:
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Используйте: /reply <номер тега> <текст>")
return
try:
target_tag_number = int(args[0])
except ValueError:
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Некорректный номер тега. Используйте число.")
return
reply_text = " ".join(args[1:])
for user_id, data in user_data.items():
if data["tag"] == f"User_{target_tag_number}":
try:
await context.bot.send_message(chat_id=user_id, text=reply_text)
await context.bot.send_message(chat_id=ADMIN_USER_ID,
text=f"Ответ пользователю User_{target_tag_number} успешно отправлен.")
except Exception as e:
logging.error(f"Не удалось отправить ответ пользователю: {e}")
await context.bot.send_message(chat_id=ADMIN_USER_ID,
text=f"Не удалось отправить сообщение пользователю User_{target_tag_number}.")
return
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Пользователь с таким номером тега не найден.")
async def ban_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.message.from_user.id == ADMIN_USER_ID:
args = context.args
if len(args) < 2:
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Используйте: /ban <номер тега> <причина>")
return
try:
target_tag_number = int(args[0])
except ValueError:
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Некорректный номер тега. Используйте число.")
return
ban_reason = " ".join(args[1:])
for user_id, data in user_data.items():
if data["tag"] == f"User_{target_tag_number}":
banned_users[user_id] = ban_reason
await context.bot.send_message(chat_id=ADMIN_USER_ID, text=f"Пользователь {data['tag']} ({data['username']}) забанен. Причина: {ban_reason}")
try:
await context.bot.send_message(chat_id=user_id, text=f"Вы забанены. Причина: {ban_reason}")
except Exception as e:
logging.error(f"Не удалось отправить сообщение о бане пользователю {user_id}: {e}")
return
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Пользователь с таким номером тега не найден.")
async def unban_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.message.from_user.id == ADMIN_USER_ID:
args = context.args
if not args:
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Используйте: /unban <номер тега>")
return
try:
target_tag_number = int(args[0])
except ValueError:
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Некорректный номер тега. Используйте число.")
return
for user_id, data in user_data.items():
if data["tag"] == f"User_{target_tag_number}":
if user_id in banned_users:
del banned_users[user_id]
await context.bot.send_message(chat_id=ADMIN_USER_ID, text=f"Пользователь {data['tag']} ({data['username']}) разбанен.")
try:
await context.bot.send_message(chat_id=user_id, text=f"Вы разбанены.")
except Exception as e:
logging.error(f"Не удалось отправить сообщение о разбане пользователю {user_id}: {e}")
return
else:
await context.bot.send_message(chat_id=ADMIN_USER_ID, text=f"Пользователь {data['tag']} ({data['username']}) не был забанен.")
return
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Пользователь с таким номером тега не найден.")
async def send_shutdown_notification(bot):
for user_id, data in user_data.items():
if data["notified_shutdown"] == 0:
try:
await bot.send_message(chat_id=user_id, text=SHUTDOWN_MESSAGE)
user_data[user_id]["notified_shutdown"] = 1
update_user_notification(user_id, 1)
except Exception as e:
logging.error(f"Не удалось отправить уведомление о выключении пользователю {user_id}: {e}")
async def send_start_notification(bot):
for user_id in user_data:
try:
await bot.send_message(chat_id=user_id, text=START_MESSAGE)
user_data[user_id]["notified_shutdown"] = 0
update_user_notification(user_id, 0)
except Exception as e:
logging.error(f"Не удалось отправить уведомление о включении пользователю {user_id}: {e}")
async def off_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
global bot_active
if update.message.from_user.id == ADMIN_USER_ID:
bot_active = False
await send_shutdown_notification(context.bot)
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Бот перешел в спящий режим.")
async def on_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
global bot_active
if update.message.from_user.id == ADMIN_USER_ID:
bot_active = True
await send_start_notification(context.bot)
await context.bot.send_message(chat_id=ADMIN_USER_ID, text="Бот включен и готов к работе.")
async def on_shutdown(app, bot):
global bot_running
logging.info("Бот выключается...")
bot_running = False
logging.info("Соединение с базой данных закрыто.")
loop = asyncio.get_event_loop()
loop.stop()
def main():
global bot_instance
application = ApplicationBuilder().token(BOT_TOKEN).build()
bot_instance = application.bot
load_users_from_db()
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_admin_reply))
application.add_handler(CommandHandler("reply", reply_command))
application.add_handler(MessageHandler(filters.StatusUpdate.NEW_CHAT_MEMBERS, handle_new_chat_members))
application.add_handler(MessageHandler(filters.ALL & ~filters.TEXT & ~filters.COMMAND, handle_non_message))
application.add_handler(CommandHandler("off", off_command))
application.add_handler(CommandHandler("on", on_command))
application.add_handler(CommandHandler("ban", ban_command))
application.add_handler(CommandHandler("unban", unban_command))
async def shutdown_handler(signum, frame):
await on_shutdown(application, application.bot)
signal.signal(signal.SIGINT, shutdown_handler)
application.run_polling()
if __name__ == '__main__':
main()Ответить - /reply
Забанить - /ban
разбанить /unban
Забанить - /ban
разбанить /unban