main.py - master - Flyoxel/TestBot

main.py
320 строк13 KB
developer
token: fix variables
25 апр 2026, 14:03
25 апр 2026, 14:033d3b142
import asyncio
import json
import string
import time
import logging
import sys
import html
from typing import Dict, Any
import os

import random
import aiogram.types
from aiogram import Bot, Dispatcher, F, Router
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import Message, CallbackQuery, BufferedInputFile
from aiogram.utils.keyboard import InlineKeyboardBuilder, ReplyKeyboardBuilder
from aiogram.filters.callback_data import CallbackData

from captcha.image import ImageCaptcha
TOKEN = os.getenv("API_TOKEN")
CHANNEL_ID = os.getenv("CHANNEL_ID")
DB_ID = os.getenv("DB_ID")
ADMIN_ID = os.getenv("ADMIN_ID")


class AdminAction(CallbackData, prefix="admin"):
    action: str
    user_id: int


class Database:
    def __init__(self, bot: Bot, channel_id: int, backend: str = "telegram"):
        self.bot = bot
        self.backend = backend
        self.CHANNEL_ID = channel_id
        self.db: Dict[str, Any] = {}

    async def save(self):
        if self.backend == "telegram":
            try:
                data = json.dumps(self.db).encode("utf-8")
                file = BufferedInputFile(data, filename=f"database-{time.time()}.json")
                message = await self.bot.send_document(self.CHANNEL_ID, file, caption="Database saved")
                await self.bot.pin_chat_message(self.CHANNEL_ID, message.message_id, disable_notification=True)
                logging.info("Database saved to Telegram successfully.")
            except Exception as e:
                logging.error(f"Failed to save DB: {e}")

    async def load(self):
        if self.backend == "telegram":
            try:
                channel = await self.bot.get_chat(self.CHANNEL_ID)
                pinned = channel.pinned_message
                if pinned and pinned.document:
                    file_info = await self.bot.get_file(pinned.document.file_id)
                    stream = await self.bot.download_file(file_info.file_path)
                    self.db = json.loads(stream.read().decode("utf-8"))
                    logging.info("Database loaded successfully.")
            except Exception as e:
                logging.error(f"Failed to load DB (starting fresh): {e}")
                self.db = {}

    async def background_save(self):
        while True:
            await asyncio.sleep(1800)  # 30 минут
            await self.save()


bot: Bot
database: Database
CHANNEL: aiogram.types.Chat
ADMIN: aiogram.types.Chat

dp = Dispatcher()
captcha_router = Router()
dp.include_router(captcha_router)


class Captcha(StatesGroup):
    generate_captcha = State()


def generate_captcha_text(length: int) -> str:
    return "".join(random.choices(string.digits, k=length))



def create_captcha_image(text: str) -> bytes:
    image = ImageCaptcha()
    return image.generate(text).read()

async def check_user_birthday(user_id: int) -> bool:
    chat_user = await bot.get_chat(user_id)
    user_birthday = chat_user.birthdate
    if user_birthday and user_birthday.year:
        return True
    return False


@dp.message(CommandStart())
async def command_start_handler(message: Message) -> None:
    member = await bot.get_chat_member(CHANNEL.id, message.from_user.id)
    # Если юзер уже в канале, игнорируем
    if member.status != "left":
        return

    user_id_str = str(message.from_user.id)
    if user_id_str in database.db:
        await second_try(message)
        return

    builder = InlineKeyboardBuilder()
    builder.button(text="Мне есть 18", callback_data="yes")
    builder.button(text="Нет", callback_data="no")

    await message.answer("Вам есть 18?", reply_markup=builder.as_markup())


@captcha_router.callback_query(F.data.in_({"yes", "no", "admin_request", "cancel"}))
async def query_requests(query: CallbackQuery, state: FSMContext):
    user_id_str = str(query.from_user.id)

    if query.data == "yes":
        await query.message.delete()
        database.db[user_id_str] = {"18+": True, "attempt": time.time() + 300}
        link = await CHANNEL.create_invite_link(name=user_id_str, expire_date=int(time.time() + 3600), member_limit=1)
        await query.message.answer(f"Приглашение в канал: {link.invite_link}")

    elif query.data == "no":
        await query.message.delete()
        database.db[user_id_str] = {"18+": False}
        builder = ReplyKeyboardBuilder()
        builder.button(text="Повторная проверка")
        await query.message.answer("Доступ разрешен только лицам старше 18 лет",
                                   reply_markup=builder.as_markup(resize_keyboard=True))

    elif query.data == "admin_request":
        if not user_id_str in database.db:
            await query.message.delete()
            await query.answer()
        user_data = database.db.get(user_id_str, {})


        if "attempt" in user_data and user_data["attempt"] > time.time():
            builder = InlineKeyboardBuilder()
            builder.button(text="Повторная проверка", callback_data="admin_request")
            await query.message.delete()
            timeleft = round(user_data["attempt"] - time.time())
            await query.message.answer(
                f"Пожалуйста, подождите {timeleft} секунд перед тем, как вы сможете подать заявку на ручное рассмотрение.",
                reply_markup=builder.as_markup())
            return

        await query.message.delete()

        # Проверка даты рождения
        has_valid_birthday = await check_user_birthday(query.from_user.id)
        if not has_valid_birthday:
            database.db.setdefault(user_id_str, {})["attempt"] = time.time() + 60
            builder = InlineKeyboardBuilder()
            builder.button(text="Повторная проверка", callback_data="admin_request")
            await query.message.answer(
                "Заявка отклонена автоматически!\n\n- Не установлена полная дата рождения в профиле.\n- Укажите дату рождения в профиле для отправки заявки на ручное рассмотрение.",
                reply_markup=builder.as_markup())
            return

        await state.set_state(Captcha.generate_captcha)
        answer = generate_captcha_text(5)
        await state.update_data(generate_captcha=answer)

        image_bytes = await asyncio.to_thread(create_captcha_image, answer)
        file = BufferedInputFile(image_bytes, filename="captcha.png")

        builder = InlineKeyboardBuilder()
        builder.button(text="Отменить", callback_data="cancel")
        await query.message.answer_photo(photo=file, caption="Решите капчу:", reply_markup=builder.as_markup())

    elif query.data == "cancel":
        await query.message.delete()
        await state.clear()
        await query.message.answer("Действие отменено.")

    await query.answer()


@captcha_router.callback_query(AdminAction.filter())
async def admin_actions(query: CallbackQuery, callback_data: AdminAction):
    if query.from_user.id != int(ADMIN_ID):
        await query.answer("У вас нет прав!", show_alert=True)
        return

    target_user_id = str(callback_data.user_id)

    if target_user_id not in database.db:
        database.db[target_user_id] = {}

    if callback_data.action == "accept":
        database.db[target_user_id]["attempt"] = time.time() + 300
        link = await CHANNEL.create_invite_link(name=target_user_id, expire_date=int(time.time() + 3600),
                                                member_limit=1)
        try:
            await bot.send_message(target_user_id,
                                   f"Заявка на вступление одобрена!\n\nСсылка на 1 час для вступления:\n{link.invite_link}")
        except Exception as e:
            logging.error(f"Не удалось отправить сообщение юзеру {target_user_id}: {e}")
        await query.message.answer(f"Заявка пользователя {target_user_id} одобрена.")

    elif callback_data.action == "decline":
        database.db[target_user_id]["attempt"] = time.time() + 300
        try:
            await bot.send_message(target_user_id,
                                   "Ваша заявка на вступление была отклонена.\n\nВы можете попробовать снова через некоторое время.")
        except Exception:
            pass
        await query.message.answer(f"Заявка пользователя {target_user_id} отклонена.")

    await query.message.delete()
    await query.answer()


@captcha_router.message(Captcha.generate_captcha)
async def process_captcha(message: Message, state: FSMContext) -> None:
    data = await state.get_data()
    user_id_str = str(message.from_user.id)

    if data.get("generate_captcha") == message.text:
        has_valid_birthday = await check_user_birthday(message.from_user.id)

        if not has_valid_birthday:
            database.db.setdefault(user_id_str, {})["attempt"] = time.time() + 60
            builder = InlineKeyboardBuilder()
            builder.button(text="Повторная проверка", callback_data="admin_request")
            await message.answer("Заявка отклонена автоматически!\n\n- Не установлена дата рождения в профиле.",
                                 reply_markup=builder.as_markup())
            await state.clear()
            return

        user = message.from_user

        user_full_name = html.escape(user.full_name) if user.full_name else "Отсутствует"
        user_name = html.escape(user.username) if user.username else "Отсутствует"

        chat_user = await bot.get_chat(user.id)
        user_birthdate = f"{chat_user.birthdate.day}.{chat_user.birthdate.month}.{chat_user.birthdate.year}"

        builder = InlineKeyboardBuilder()
        builder.button(text="Принять", callback_data=AdminAction(action="accept", user_id=user.id))
        builder.button(text="Отказать", callback_data=AdminAction(action="decline", user_id=user.id))

        admin_text = (
            f"<b>Новый запрос на вступление в канал!</b>\n\n"
            f"<b>Пользователь:</b>\n"
            f"- Имя: {user_full_name}\n"
            f"- Юзернейм: @{user_name} (id: <code>{user.id}</code>)\n"
            f"- Дата рождения: {user_birthdate}"
        )

        await bot.send_message(ADMIN.id, admin_text, reply_markup=builder.as_markup())
        await message.answer("Запрос отправлен администрации на рассмотрение.")
        database.db.setdefault(user_id_str, {})["attempt"] = time.time() + 300
    else:
        await message.answer("Капча неправильная! Попробуйте отправить запрос еще раз через минуту.")
        database.db.setdefault(user_id_str, {})["attempt"] = time.time() + 20

    await state.clear()


@dp.message(F.text == 'Повторная проверка')
async def second_try(message: Message) -> None:
    user_id_str = str(message.from_user.id)

    if user_id_str in database.db:
        member = await bot.get_chat_member(CHANNEL.id, message.from_user.id)
        if member.status == "left":
            is_18_plus = database.db[user_id_str].get("18+", False)

            if is_18_plus:
                a = "Повторная отправка запроса.\n\nХотите отправить заявку на ручную проверку администратору?"
            else:
                a = "Повторная отправка запроса.\n\nРанее вы указали, что вам нет 18 лет.\nХотите отправить заявку на ручную проверку администратору?"

            builder = InlineKeyboardBuilder()
            builder.button(text="Повторная проверка", callback_data="admin_request")
            await message.answer(
                f"{a}\n\n⚠️ ВАЖНО! Перед отправкой заявки:\n\n1. Укажите в профиле Telegram дату рождения.\n2. Сделайте её видимой.\n3. После проверки сможете снова скрыть.",
                reply_markup=builder.as_markup())


async def main() -> None:
    global CHANNEL, bot, ADMIN, database

    bot = Bot(token=TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
    CHANNEL = await bot.get_chat(CHANNEL_ID)
    ADMIN = await bot.get_chat(ADMIN_ID)

    database = Database(bot=bot, channel_id=int(DB_ID), backend="telegram")
    await database.load()

    save_task = asyncio.create_task(database.background_save())

    try:
        await dp.start_polling(bot)
    except (KeyboardInterrupt, SystemExit):
        logging.info("Stopping...")
    finally:
        save_task.cancel()
        await database.save()  # Сохраняем при выключении
        await bot.session.close()  # Закрываем сессию aiohttp


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, stream=sys.stdout)
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass