make the bot.
This commit is contained in:
@@ -7,6 +7,12 @@ class Settings(BaseSettings):
|
||||
Application settings loaded from environment variables or a .env file.
|
||||
"""
|
||||
bot_token: SecretStr
|
||||
group_chat_id: int
|
||||
admin_ids: str
|
||||
|
||||
@property
|
||||
def admins(self) -> list[int]:
|
||||
return [int(x.strip()) for x in self.admin_ids.split(",") if x.strip().isdigit()]
|
||||
|
||||
# Tell Pydantic to read from a .env file if it exists, ignoring extra fields
|
||||
model_config = SettingsConfigDict(
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
from aiogram import Router
|
||||
from . import user, admin
|
||||
|
||||
from . import user, admin, get_chat_id
|
||||
|
||||
def get_handlers_router() -> Router:
|
||||
"""
|
||||
Combines all feature routers (user, admin, etc.) into one main router.
|
||||
"""
|
||||
router = Router()
|
||||
router.include_router(user.router)
|
||||
router.include_router(admin.router)
|
||||
router.include_router(get_chat_id.router)
|
||||
router.include_router(user.router)
|
||||
return router
|
||||
|
||||
@@ -1,14 +0,0 @@
|
||||
from aiogram import Router
|
||||
from aiogram.filters import Command
|
||||
from aiogram.types import Message
|
||||
|
||||
router = Router()
|
||||
|
||||
|
||||
@router.message(Command("admin"))
|
||||
async def cmd_admin(message: Message):
|
||||
"""
|
||||
Handles /admin command (placeholder).
|
||||
In production, you'd restrict this using a custom filter or admin ID list.
|
||||
"""
|
||||
await message.answer("Welcome to the Admin panel. Currently, no admin tasks are configured.")
|
||||
@@ -0,0 +1,10 @@
|
||||
from aiogram import Router
|
||||
from aiogram.filters import Command
|
||||
from aiogram.types import Message
|
||||
from bot.utils.i18n import _
|
||||
|
||||
router = Router()
|
||||
|
||||
@router.message(Command("id"))
|
||||
async def cmd_id(message: Message, locale: str):
|
||||
await message.answer(_("id_command", locale=locale, id=message.chat.id))
|
||||
+92
-48
@@ -1,56 +1,100 @@
|
||||
from aiogram import F, Router
|
||||
from aiogram.filters import Command, CommandStart
|
||||
from aiogram.types import CallbackQuery, Message
|
||||
from aiogram import Router, F, Bot
|
||||
from aiogram.filters import CommandStart
|
||||
from aiogram.types import Message, BufferedInputFile
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.fsm.state import State, StatesGroup
|
||||
|
||||
from bot.keyboards.inline import get_links_keyboard
|
||||
from bot.keyboards.reply import get_main_menu_keyboard
|
||||
from bot.config import settings
|
||||
from bot.utils.captcha import generate_captcha
|
||||
from bot.utils.db import db
|
||||
from bot.utils.i18n import _
|
||||
|
||||
router = Router()
|
||||
|
||||
class CaptchaStates(StatesGroup):
|
||||
waiting_for_captcha = State()
|
||||
|
||||
@router.message(CommandStart())
|
||||
async def cmd_start(message: Message):
|
||||
"""
|
||||
Handles /start command. Greets user and sends the main reply menu.
|
||||
"""
|
||||
await message.answer(
|
||||
f"Hello, {message.from_user.full_name}! Welcome to Frankenstein Bot.\n"
|
||||
f"This bot is bootstrapped with a modular and clean structure.",
|
||||
reply_markup=get_main_menu_keyboard(),
|
||||
async def cmd_start(message: Message, locale: str):
|
||||
db.log_user(message.from_user)
|
||||
await message.answer(_("welcome", locale=locale))
|
||||
|
||||
@router.message(CaptchaStates.waiting_for_captcha)
|
||||
async def process_captcha(message: Message, state: FSMContext, bot: Bot, locale: str):
|
||||
data = await state.get_data()
|
||||
correct_captcha = data.get("captcha_text")
|
||||
original_msg_id = data.get("original_msg_id")
|
||||
|
||||
if message.text and message.text.upper() == correct_captcha:
|
||||
await state.clear()
|
||||
|
||||
try:
|
||||
db_msg_id = data.get("db_msg_id", "N/A")
|
||||
prefix = f"Forwarding, ID [{db_msg_id}]:"
|
||||
target_chat_id = settings.group_chat_id
|
||||
|
||||
if data.get('has_media'):
|
||||
new_caption = prefix
|
||||
if data.get('original_caption'):
|
||||
new_caption += f"\n{data.get('original_caption')}"
|
||||
|
||||
await bot.copy_message(
|
||||
chat_id=target_chat_id,
|
||||
from_chat_id=message.chat.id,
|
||||
message_id=original_msg_id,
|
||||
caption=new_caption
|
||||
)
|
||||
elif data.get('is_text'):
|
||||
await bot.send_message(
|
||||
chat_id=target_chat_id,
|
||||
text=f"{prefix}\n{data.get('original_text')}"
|
||||
)
|
||||
else:
|
||||
# Other types (stickers, etc.)
|
||||
await bot.copy_message(
|
||||
chat_id=target_chat_id,
|
||||
from_chat_id=message.chat.id,
|
||||
message_id=original_msg_id
|
||||
)
|
||||
await bot.send_message(chat_id=target_chat_id, text=prefix)
|
||||
|
||||
await message.answer(_("captcha_solved", locale=locale))
|
||||
|
||||
except Exception as e:
|
||||
await message.answer(_("captcha_error", locale=locale))
|
||||
print(f"Error forwarding: {e}")
|
||||
else:
|
||||
# Generate new captcha
|
||||
captcha_text, captcha_img = generate_captcha()
|
||||
await state.update_data(captcha_text=captcha_text)
|
||||
await message.answer_photo(
|
||||
BufferedInputFile(captcha_img, filename="captcha.png"),
|
||||
caption=_("incorrect_captcha", locale=locale)
|
||||
)
|
||||
|
||||
@router.message(F.chat.type == "private")
|
||||
async def handle_any_message(message: Message, state: FSMContext, locale: str):
|
||||
# Log everything
|
||||
db_msg_id = db.log_message(message)
|
||||
|
||||
# Generate captcha
|
||||
captcha_text, captcha_img = generate_captcha()
|
||||
|
||||
has_media = bool(message.photo or message.video or message.document or message.audio or message.voice)
|
||||
is_text = bool(message.text)
|
||||
|
||||
await state.set_state(CaptchaStates.waiting_for_captcha)
|
||||
await state.update_data(
|
||||
captcha_text=captcha_text,
|
||||
db_msg_id=db_msg_id,
|
||||
original_msg_id=message.message_id,
|
||||
original_text=message.text,
|
||||
original_caption=message.caption,
|
||||
has_media=has_media,
|
||||
is_text=is_text
|
||||
)
|
||||
|
||||
|
||||
@router.message(Command("help"))
|
||||
@router.message(F.text == "Help ❓")
|
||||
async def cmd_help(message: Message):
|
||||
"""
|
||||
Handles /help command or 'Help ❓' reply keyboard button click.
|
||||
"""
|
||||
await message.answer(
|
||||
"Here is what I can do:\n"
|
||||
"/start - Start the bot\n"
|
||||
"/help - Get help information\n"
|
||||
"/about - Learn more about the bot",
|
||||
reply_markup=get_links_keyboard(),
|
||||
|
||||
await message.answer_photo(
|
||||
BufferedInputFile(captcha_img, filename="captcha.png"),
|
||||
caption=_("solve_captcha", locale=locale)
|
||||
)
|
||||
|
||||
|
||||
@router.message(Command("about"))
|
||||
@router.message(F.text == "About ℹ️")
|
||||
async def cmd_about(message: Message):
|
||||
"""
|
||||
Handles /about command or 'About ℹ️' reply keyboard button click.
|
||||
"""
|
||||
await message.answer(
|
||||
"Frankenstein Bot is a high-performance Python Telegram bot built on aiogram v3, "
|
||||
"designed to work seamlessly with systemd process management."
|
||||
)
|
||||
|
||||
|
||||
@router.callback_query(F.data == "support_contact")
|
||||
async def process_support_callback(callback: CallbackQuery):
|
||||
"""
|
||||
Processes the inline keyboard callback query 'support_contact'.
|
||||
"""
|
||||
await callback.message.answer("Support: Open an issue or contact our administrator.")
|
||||
await callback.answer()
|
||||
|
||||
@@ -1,13 +0,0 @@
|
||||
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
|
||||
from aiogram.utils.keyboard import InlineKeyboardBuilder
|
||||
|
||||
|
||||
def get_links_keyboard() -> InlineKeyboardMarkup:
|
||||
"""
|
||||
Creates an inline keyboard markup with external links and callback query button.
|
||||
"""
|
||||
builder = InlineKeyboardBuilder()
|
||||
builder.add(InlineKeyboardButton(text="Google", url="https://google.com"))
|
||||
builder.add(InlineKeyboardButton(text="Contact Support 💬", callback_data="support_contact"))
|
||||
builder.adjust(1)
|
||||
return builder.as_markup()
|
||||
@@ -1,13 +0,0 @@
|
||||
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup
|
||||
from aiogram.utils.keyboard import ReplyKeyboardBuilder
|
||||
|
||||
|
||||
def get_main_menu_keyboard() -> ReplyKeyboardMarkup:
|
||||
"""
|
||||
Creates the main menu reply keyboard markup.
|
||||
"""
|
||||
builder = ReplyKeyboardBuilder()
|
||||
builder.add(KeyboardButton(text="Help ❓"))
|
||||
builder.add(KeyboardButton(text="About ℹ️"))
|
||||
builder.adjust(2)
|
||||
return builder.as_markup(resize_keyboard=True)
|
||||
@@ -24,3 +24,19 @@ class LoggingMiddleware(BaseMiddleware):
|
||||
except Exception as e:
|
||||
logger.error("Error handling event %s: %s", event.__class__.__name__, e, exc_info=True)
|
||||
raise e
|
||||
|
||||
|
||||
class I18nMiddleware(BaseMiddleware):
|
||||
async def __call__(
|
||||
self,
|
||||
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
|
||||
event: TelegramObject,
|
||||
data: Dict[str, Any],
|
||||
) -> Any:
|
||||
from_user = data.get("event_from_user")
|
||||
if from_user and from_user.language_code:
|
||||
data["locale"] = from_user.language_code
|
||||
else:
|
||||
data["locale"] = "en"
|
||||
|
||||
return await handler(event, data)
|
||||
|
||||
+1
-4
@@ -7,9 +7,6 @@ async def set_bot_commands(bot: Bot):
|
||||
Sets default commands in the Telegram client's command menu.
|
||||
"""
|
||||
commands = [
|
||||
BotCommand(command="start", description="Start Frankenstein Bot"),
|
||||
BotCommand(command="help", description="Show help information"),
|
||||
BotCommand(command="about", description="About the bot"),
|
||||
BotCommand(command="admin", description="Admin panel placeholder"),
|
||||
BotCommand(command="start", description="Start the bot"),
|
||||
]
|
||||
await bot.set_my_commands(commands, scope=BotCommandScopeDefault())
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
import random
|
||||
import string
|
||||
from io import BytesIO
|
||||
from captcha.image import ImageCaptcha
|
||||
|
||||
def generate_captcha():
|
||||
"""Generates a random captcha and returns (text, image_bytes)"""
|
||||
captcha_text = ''.join(random.choices(string.ascii_uppercase + string.digits, k=5))
|
||||
image = ImageCaptcha(width=280, height=90)
|
||||
data = image.generate(captcha_text)
|
||||
return captcha_text, data.getvalue()
|
||||
@@ -0,0 +1,31 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
class I18n:
|
||||
def __init__(self, default_locale="en"):
|
||||
self.default_locale = default_locale
|
||||
self.translations = {}
|
||||
self.load_translations()
|
||||
|
||||
def load_translations(self):
|
||||
locales_dir = "locales"
|
||||
if not os.path.exists(locales_dir):
|
||||
return
|
||||
|
||||
for filename in os.listdir(locales_dir):
|
||||
if filename.endswith(".json"):
|
||||
locale = filename[:-5]
|
||||
with open(os.path.join(locales_dir, filename), "r", encoding="utf-8") as f:
|
||||
self.translations[locale] = json.load(f)
|
||||
|
||||
def get(self, key, locale=None, **kwargs):
|
||||
if locale is None:
|
||||
locale = self.default_locale
|
||||
text = self.translations.get(locale, {}).get(key, self.translations.get(self.default_locale, {}).get(key, key))
|
||||
try:
|
||||
return text.format(**kwargs)
|
||||
except (KeyError, IndexError):
|
||||
return text
|
||||
|
||||
i18n = I18n()
|
||||
_ = i18n.get
|
||||
Reference in New Issue
Block a user