TG Bot for receiving applications Python

Тг бот для приема заявок.
Для использования замените BOT_TOKEN на свой токен, admin Id на свой id и при надобности измените вопросы в самом коде

Код

import asyncio
import logging
import json

from aiogram import Bot, Dispatcher, types, F
from aiogram.filters import CommandStart, Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import ReplyKeyboardRemove, InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.exceptions import TelegramForbiddenError
from aiogram.client.default import DefaultBotProperties

# Замените на свой токен и ID администратора
BOT_TOKEN = "7916714809:AAGMeW8m8irrP0UDewrJn1ITEFH4TEJD1NY"
ADMIN_ID = [URL='https://lolz.live/tel:1364958631']1364958631[/URL]
AUTHORIZED_USERS_FILE = "authorized_users.json" # Fixed filename


logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
log = logging.getLogger(__name__)

try:
with open(AUTHORIZED_USERS_FILE, "r") as f:
AUTHORIZED_USERS = set(json.load(f))
log.info(f"Loaded AUTHORIZED_USERS from {AUTHORIZED_USERS_FILE}: {AUTHORIZED_USERS}")
except FileNotFoundError:
AUTHORIZED_USERS = set()
log.warning(f"{AUTHORIZED_USERS_FILE} not found. Starting with an empty AUTHORIZED_USERS list.")
except json.JSONDecodeError:
AUTHORIZED_USERS = set()
log.error(f"Error decoding JSON in {AUTHORIZED_USERS_FILE}. Starting with an empty AUTHORIZED_USERS list.")

# Создаем бота и диспетчер
bot_properties = DefaultBotProperties(parse_mode="HTML")
bot = Bot(token=BOT_TOKEN, default=bot_properties)
dp = Dispatcher()

class UserForm(StatesGroup):
reason = State()
experience = State()
time_commitment = State()


@dp.message(Command("admin"), F.from_user.id == ADMIN_ID)
async def admin_command(message: types.Message):
log.info(f"Администратор {message.from_user.id} вызвал команду /admin")
await message.answer("Привет, администратор! Какие будут указания?")

@dp.message(Command("admin"))
async def not_admin(message: types.Message):
await message.reply("Вы не являетесь администратором!")

@dp.message(CommandStart())
async def start_command(message: types.Message, state: FSMContext):
user_id = message.from_user.id
log.info(f"Пользователь {user_id} запустил бота")

if user_id in AUTHORIZED_USERS:
log.info(f"Пользователь {user_id} имеет полный доступ.")
await message.answer("ℹ <i>Информация <b>Vors Team</b></i> \n\n<b> Проценты профитов</b>")
await state.clear() # очистить на всякий случай
else:
log.info(f"Пользователь {user_id} начинает заполнение анкеты.")
await state.clear()
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="Да", callback_data="start_yes"),
InlineKeyboardButton(text="Нет", callback_data="start_no"),
]
])
text = "Привет мой дорогой друг!<b>\n\nДля вступления в нашу команду, я задам тебе пару вопросов, на которые ты должен ответить честно \n\nСогласен?</b>"
await message.answer(text, reply_markup=keyboard)

@dp.callback_query(lambda query: query.data in ("start_yes", "start_no"))
async def handle_start_callback(callback: types.CallbackQuery, state: FSMContext):
if callback.data == "start_yes":
log.info(f"Пользователь {callback.from_user.id} нажал 'Да'")
await callback.message.answer("<b>⚡Почему именно наша команда?</b>", reply_markup=ReplyKeyboardRemove())
await state.set_state(UserForm.reason)
else:
log.info(f"Пользователь {callback.from_user.id} нажал 'Нет'")
await callback.message.answer("Увы, но тебе не сюда, бро.", reply_markup=ReplyKeyboardRemove())

@dp.message(UserForm.reason)
async def process_reason(message: types.Message, state: FSMContext):
log.info(f"Пользователь {message.from_user.id} ввел причину")
await state.update_data(reason=message.text)
await message.answer("<b>⚡Ваш опыт работы?</b>", reply_markup=ReplyKeyboardRemove())
await state.set_state(UserForm.experience)

@dp.message(UserForm.experience)
async def process_experience(message: types.Message, state: FSMContext):
log.info(f"Пользователь {message.from_user.id} ввел опыт")
await state.update_data(experience=message.text)
await message.answer("<b>⚡Сколько времени будете уделять работе?</b>", reply_markup=ReplyKeyboardRemove())
await state.set_state(UserForm.time_commitment)

@dp.message(UserForm.time_commitment)
async def process_time_commitment(message: types.Message, state: FSMContext):
log.info(f"Пользователь {message.from_user.id} ввел время")
await state.update_data(time_commitment=message.text)
await show_summary_and_buttons(message, state)

async def show_summary_and_buttons(message: types.Message, state: FSMContext):
log.info(f"Формирование и отправка сводки пользователю {message.from_user.id}")
data = await state.get_data()
summary = (
"Пожалуйста, проверьте введенные данные:\n"
f"<b>⚡Почему именно наша команда?</b>: {data.get('reason', 'Не указано')}\n" # HTML
f"<b>⚡Ваш опыт работы?</b>: {data.get('experience', 'Не указано')}\n" # HTML
f"<b>⚡Сколько времени будете уделять работе?</b>: {data.get('time_commitment', 'Не указано')}\n"
)
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="Принять", callback_data="accept"),
InlineKeyboardButton(text="Отказаться", callback_data="decline"),
]
])
try:
log.info(f"Попытка отправки сообщения: {summary}")
sent_message = await message.answer(summary, reply_markup=keyboard)
log.info(f"Сообщение с кнопками отправлено, ID: {sent_message.message_id}")
await state.update_data(last_message_id=sent_message.message_id)
except Exception as e:
log.exception(f"Ошибка при отправке сообщения с кнопками: {e}, Тип ошибки: {type(e)}")
log.info("show_summary_and_buttons завершена")

@dp.callback_query(lambda query: query.data in ("accept", "decline"))
async def process_callback(callback: types.CallbackQuery, state: FSMContext):
log.info(f"Callback запрос получен: {callback.data}")
await callback.answer()

if callback.data == "accept":
log.info(f"Пользователь {callback.from_user.id} нажал 'Принять'")
try:
await bot.send_message(callback.from_user.id, " Отлично, твоя заявка отправлена администратору.")
except TelegramForbiddenError:
log.error(f"Пользователь {callback.from_user.id} заблокировал бота.")
elif callback.data == "decline":
log.info(f"Пользователь {callback.from_user.id} нажал 'Отказаться'")
try:
await bot.send_message(callback.from_user.id, "Ваша заявка отклонена. Пожалуйста, попробуйте еще раз.")
except TelegramForbiddenError:
log.error(f"Пользователь {callback.from_user.id} заблокировал бота.")

await callback.message.delete_reply_markup()
user_id = callback.from_user.id
data = await state.get_data()
await state.clear()

await send_to_admin(callback.message, data, user_id)

async def send_to_admin(message: types.Message, data, user_id):
log.info(f"Отправка заявки администратору...")
admin_message = (
"Новая заявка:\n"
f"Имя пользователя: {message.from_user.full_name}\n"
f"ID пользователя: {message.from_user.id}\n"
f"<b>⚡Почему именно наша команда?</b>: {data.get('reason', 'Не указано')}\n" # HTML
f"<b>⚡Ваш опыт работы?</b>: {data.get('experience', 'Не указано')}\n" # HTML
f"<b>⚡Сколько времени будете уделять работе?</b>: {data.get('time_commitment', 'Не указано')}\n" # HTML
)

admin_keyboard = InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="Принять", callback_data=f"admin_accept:{user_id}"),
InlineKeyboardButton(text="Отказать", callback_data=f"admin_decline:{user_id}"),
]
])
try:
await bot.send_message(ADMIN_ID, admin_message, reply_markup=admin_keyboard)
log.info(f"Заявка успешно отправлена администратору: {admin_message}")
except Exception as e:
log.exception(f"Ошибка при отправке заявки администратору: {e}")

@dp.callback_query(lambda query: query.data.startswith("admin_"))
async def process_admin_callback(callback: types.CallbackQuery):
action, user_id = callback.data.split(":")
user_id = int(user_id)
log.info(f"Администратор нажал {action} для пользователя {user_id}")

if action == "admin_accept":
try:
await bot.send_message(user_id, "Администратор одобрил твою заявку! Пропишите /start")
await callback.answer("Заявка принята")
AUTHORIZED_USERS.add(user_id) # добавляем пользователя в список
log.info(f"Пользователь {user_id} добавлен в AUTHORIZED_USERS")

try:
with open(AUTHORIZED_USERS_FILE, "w") as f:
json.dump(list(AUTHORIZED_USERS), f)
log.info(f"Saved AUTHORIZED_USERS to {AUTHORIZED_USERS_FILE}")
except Exception as e:
log.exception(f"Error saving AUTHORIZED_USERS to {AUTHORIZED_USERS_FILE}: {e}")

except TelegramForbiddenError:
log.error(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")
await callback.answer(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")
except Exception as e:
log.exception(f"Ошибка при отправке сообщения пользователю {user_id}: {e}")
await callback.answer("Произошла ошибка")

elif action == "admin_decline":
try:
await bot.send_message(user_id, "Администратор отклонил вашу заявку. Пожалуйста, попробуйте еще раз.")
await callback.answer("Заявка отклонена")
except TelegramForbiddenError:
log.error(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")
await callback.answer(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")
except Exception as e:
log.exception(f"Ошибка при отправке сообщения пользователю {user_id}: {e}")
await callback.answer("Произошла ошибка")
else:
await callback.answer("Неизвестное действие администратора")

# Запуск бота
async def main():
try:
await dp.start_polling(bot)
finally:
await bot.session.close()

if __name__ == "__main__":
asyncio.run(main())

Python

import asyncio

import logging

import json 



from aiogram import Bot, Dispatcher, types, F

from aiogram.filters import CommandStart, Command

from aiogram.fsm.context import FSMContext

from aiogram.fsm.state import State, StatesGroup

from aiogram.types import ReplyKeyboardRemove, InlineKeyboardButton, InlineKeyboardMarkup

from aiogram.exceptions import TelegramForbiddenError

from aiogram.client.default import DefaultBotProperties



# Замените на свой токен и ID администратора

BOT_TOKEN = "7916714809:AAGMeW8m8irrP0UDewrJn1ITEFH4TEJD1NY"

ADMIN_ID = [URL='https://lolz.live/tel:1364958631']1364958631[/URL]

AUTHORIZED_USERS_FILE = "authorized_users.json" # Fixed filename





logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

log = logging.getLogger(__name__)



try:

    with open(AUTHORIZED_USERS_FILE, "r") as f:

        AUTHORIZED_USERS = set(json.load(f))

    log.info(f"Loaded AUTHORIZED_USERS from {AUTHORIZED_USERS_FILE}: {AUTHORIZED_USERS}")

except FileNotFoundError:

    AUTHORIZED_USERS = set()

    log.warning(f"{AUTHORIZED_USERS_FILE} not found. Starting with an empty AUTHORIZED_USERS list.")

except json.JSONDecodeError:

    AUTHORIZED_USERS = set()

    log.error(f"Error decoding JSON in {AUTHORIZED_USERS_FILE}. Starting with an empty AUTHORIZED_USERS list.")



# Создаем бота и диспетчер

bot_properties = DefaultBotProperties(parse_mode="HTML") 

bot = Bot(token=BOT_TOKEN, default=bot_properties)

dp = Dispatcher()



class UserForm(StatesGroup):

    reason = State()

    experience = State()

    time_commitment = State()





@dp.message(Command("admin"), F.from_user.id == ADMIN_ID)

async def admin_command(message: types.Message):

    log.info(f"Администратор {message.from_user.id} вызвал команду /admin")

    await message.answer("Привет, администратор! Какие будут указания?")



@dp.message(Command("admin"))

async def not_admin(message: types.Message):

     await message.reply("Вы не являетесь администратором!")



@dp.message(CommandStart())

async def start_command(message: types.Message, state: FSMContext):

    user_id = message.from_user.id

    log.info(f"Пользователь {user_id} запустил бота")



    if user_id in AUTHORIZED_USERS:

        log.info(f"Пользователь {user_id} имеет полный доступ.")

        await message.answer("ℹ <i>Информация <b>Vors Team</b></i> \n\n<b> Проценты профитов</b>")

        await state.clear() # очистить на всякий случай

    else:

        log.info(f"Пользователь {user_id} начинает заполнение анкеты.")

        await state.clear()

        keyboard = InlineKeyboardMarkup(inline_keyboard=[

            [

                InlineKeyboardButton(text="Да", callback_data="start_yes"),

                InlineKeyboardButton(text="Нет", callback_data="start_no"),

            ]

        ])

        text = "Привет мой дорогой друг!<b>\n\nДля вступления в нашу команду, я задам тебе пару вопросов, на которые ты должен ответить честно \n\nСогласен?</b>"

        await message.answer(text, reply_markup=keyboard)



@dp.callback_query(lambda query: query.data in ("start_yes", "start_no"))

async def handle_start_callback(callback: types.CallbackQuery, state: FSMContext):

    if callback.data == "start_yes":

        log.info(f"Пользователь {callback.from_user.id} нажал 'Да'")

        await callback.message.answer("<b>⚡Почему именно наша команда?</b>", reply_markup=ReplyKeyboardRemove()) 

        await state.set_state(UserForm.reason)

    else:

        log.info(f"Пользователь {callback.from_user.id} нажал 'Нет'")

        await callback.message.answer("Увы, но тебе не сюда, бро.", reply_markup=ReplyKeyboardRemove())



@dp.message(UserForm.reason)

async def process_reason(message: types.Message, state: FSMContext):

    log.info(f"Пользователь {message.from_user.id} ввел причину")

    await state.update_data(reason=message.text)

    await message.answer("<b>⚡Ваш опыт работы?</b>", reply_markup=ReplyKeyboardRemove()) 

    await state.set_state(UserForm.experience)



@dp.message(UserForm.experience)

async def process_experience(message: types.Message, state: FSMContext):

    log.info(f"Пользователь {message.from_user.id} ввел опыт")

    await state.update_data(experience=message.text)

    await message.answer("<b>⚡Сколько времени будете уделять работе?</b>", reply_markup=ReplyKeyboardRemove()) 

    await state.set_state(UserForm.time_commitment)



@dp.message(UserForm.time_commitment)

async def process_time_commitment(message: types.Message, state: FSMContext):

    log.info(f"Пользователь {message.from_user.id} ввел время")

    await state.update_data(time_commitment=message.text)

    await show_summary_and_buttons(message, state)



async def show_summary_and_buttons(message: types.Message, state: FSMContext):

    log.info(f"Формирование и отправка сводки пользователю {message.from_user.id}")

    data = await state.get_data()

    summary = (

        "Пожалуйста, проверьте введенные данные:\n"

        f"<b>⚡Почему именно наша команда?</b>: {data.get('reason', 'Не указано')}\n"  # HTML

        f"<b>⚡Ваш опыт работы?</b>: {data.get('experience', 'Не указано')}\n"  # HTML

        f"<b>⚡Сколько времени будете уделять работе?</b>: {data.get('time_commitment', 'Не указано')}\n" 

    )

    keyboard = InlineKeyboardMarkup(inline_keyboard=[

        [

            InlineKeyboardButton(text="Принять", callback_data="accept"),

            InlineKeyboardButton(text="Отказаться", callback_data="decline"),

        ]

    ])

    try:

        log.info(f"Попытка отправки сообщения: {summary}")

        sent_message = await message.answer(summary, reply_markup=keyboard)

        log.info(f"Сообщение с кнопками отправлено, ID: {sent_message.message_id}")

        await state.update_data(last_message_id=sent_message.message_id)

    except Exception as e:

        log.exception(f"Ошибка при отправке сообщения с кнопками: {e}, Тип ошибки: {type(e)}")

    log.info("show_summary_and_buttons завершена")



@dp.callback_query(lambda query: query.data in ("accept", "decline")) 

async def process_callback(callback: types.CallbackQuery, state: FSMContext):

    log.info(f"Callback запрос получен: {callback.data}")

    await callback.answer()



    if callback.data == "accept":

        log.info(f"Пользователь {callback.from_user.id} нажал 'Принять'")

        try:

            await bot.send_message(callback.from_user.id, " Отлично, твоя заявка отправлена администратору.")

        except TelegramForbiddenError:

            log.error(f"Пользователь {callback.from_user.id} заблокировал бота.")

    elif callback.data == "decline":

        log.info(f"Пользователь {callback.from_user.id} нажал 'Отказаться'")

        try:

            await bot.send_message(callback.from_user.id, "Ваша заявка отклонена. Пожалуйста, попробуйте еще раз.")

        except TelegramForbiddenError:

            log.error(f"Пользователь {callback.from_user.id} заблокировал бота.")



    await callback.message.delete_reply_markup()

    user_id = callback.from_user.id

    data = await state.get_data()

    await state.clear()



    await send_to_admin(callback.message, data, user_id)



async def send_to_admin(message: types.Message, data, user_id):

    log.info(f"Отправка заявки администратору...")

    admin_message = (

        "Новая заявка:\n"

        f"Имя пользователя: {message.from_user.full_name}\n"

        f"ID пользователя: {message.from_user.id}\n"

        f"<b>⚡Почему именно наша команда?</b>: {data.get('reason', 'Не указано')}\n"  # HTML

        f"<b>⚡Ваш опыт работы?</b>: {data.get('experience', 'Не указано')}\n"  # HTML

        f"<b>⚡Сколько времени будете уделять работе?</b>: {data.get('time_commitment', 'Не указано')}\n"  # HTML

    )



    admin_keyboard = InlineKeyboardMarkup(inline_keyboard=[

        [

            InlineKeyboardButton(text="Принять", callback_data=f"admin_accept:{user_id}"),

            InlineKeyboardButton(text="Отказать", callback_data=f"admin_decline:{user_id}"),

        ]

    ])

    try:

        await bot.send_message(ADMIN_ID, admin_message, reply_markup=admin_keyboard)

        log.info(f"Заявка успешно отправлена администратору: {admin_message}")

    except Exception as e:

        log.exception(f"Ошибка при отправке заявки администратору: {e}")



@dp.callback_query(lambda query: query.data.startswith("admin_"))

async def process_admin_callback(callback: types.CallbackQuery):

    action, user_id = callback.data.split(":")

    user_id = int(user_id)

    log.info(f"Администратор нажал {action} для пользователя {user_id}")



    if action == "admin_accept":

        try:

            await bot.send_message(user_id, "Администратор одобрил твою заявку! Пропишите /start")

            await callback.answer("Заявка принята")

            AUTHORIZED_USERS.add(user_id) # добавляем пользователя в список

            log.info(f"Пользователь {user_id} добавлен в AUTHORIZED_USERS")



            try:

                with open(AUTHORIZED_USERS_FILE, "w") as f:

                    json.dump(list(AUTHORIZED_USERS), f)

                log.info(f"Saved AUTHORIZED_USERS to {AUTHORIZED_USERS_FILE}")

            except Exception as e:

                log.exception(f"Error saving AUTHORIZED_USERS to {AUTHORIZED_USERS_FILE}: {e}")



        except TelegramForbiddenError:

            log.error(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")

            await callback.answer(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")

        except Exception as e:

            log.exception(f"Ошибка при отправке сообщения пользователю {user_id}: {e}")

            await callback.answer("Произошла ошибка")



    elif action == "admin_decline":

        try:

            await bot.send_message(user_id, "Администратор отклонил вашу заявку. Пожалуйста, попробуйте еще раз.")

            await callback.answer("Заявка отклонена")

        except TelegramForbiddenError:

            log.error(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")

            await callback.answer(f"Не удалось отправить сообщение пользователю {user_id}, он заблокировал бота")

        except Exception as e:

            log.exception(f"Ошибка при отправке сообщения пользователю {user_id}: {e}")

            await callback.answer("Произошла ошибка")

    else:

        await callback.answer("Неизвестное действие администратора")



# Запуск бота

async def main():

    try:

        await dp.start_polling(bot)

    finally:

        await bot.session.close()



if __name__ == "__main__":

    asyncio.run(main())

Буду рад, если оцените мою работу!