Загрузка...

[TG BOT] хз как эта хуета называетса

Тема в разделе Python создана пользователем zxcFlezyi 25 янв 2025. (поднята 27 апр 2025) 371 просмотр

Загрузка...
Сортировать
  1. zxcFlezyi
    Кому понравилось (1) Скопировать BBCode
    zxcFlezyi Автор темы 25 янв 2025 65 26 окт 2024
    кароче эта чота на подобии какой-то хуйни тут есть оповещения о включения и выключения бота /on /off так же есть бан и разбан и бд вот да
    лично сам его использую для связи ибо лс в тг засорен :finger_up:


    Код
    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()
     
    25 янв 2025
  2. zxcFlezyi
    Кому понравилось (1) Скопировать BBCode
    zxcFlezyi Автор темы 25 янв 2025 65 26 окт 2024
    бот хуйня никому не нужен но мне он полезен :finger_up:
     
    25 янв 2025
    1. Seniia
      zxcFlezyi, нормальный бот, не нужно себя принижать, возьму его себе на вооружение)
      25 янв 2025
    2. zxcFlezyi Автор темы
      Seniia, тыж мое золотце:love:
      25 янв 2025
  3. Zizion
    Скопировать BBCode
    Zizion 25 янв 2025 что то смешное 92 15 июн 2023
    я тоже хуй знает зачем он мне но сохраню
     
    25 янв 2025
    1. zxcFlezyi Автор темы
      25 янв 2025
Top