Jepetobot/main.py at master · Alexvidalcor/Jepetobot · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
# Telegram libraries
from telegram import ForceReply, Update, ReplyKeyboardMarkup, ReplyKeyboardRemove, InlineKeyboardMarkup, InlineKeyboardButton, InlineQueryResultArticle, InputTextMessageContent, constants
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters, ConversationHandler, CallbackQueryHandler,  InlineQueryHandler, CallbackContext


# Custom importation
from src.env.app_public_env import appVersion
from src.env.app_secrets_env import telegramToken
from src.modules import db, responses, settings
from src.modules.security import security_user


# Start Function
@security_user.UsersFirewall
async def Start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Send a message when the command /start is issued.
    user = update.effective_user

    await update.message.reply_html(
        rf"Hi {user.mention_html()}!",
        reply_markup=ForceReply(selective=True),
    )


# Help function
@security_user.UsersFirewall
async def HelpCommand(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Send a message when the command /help is issued.
    await update.message.reply_text(
        "---------------------------\n\n" \
        "<strong>Developer: </strong>Alexvidalcor\n\n" \
        "<strong>Source code: </strong><a href=\"https://github.com/Alexvidalcor/jepetobot\">Github Page\n\n</a>" \
        f"<strong>Version: </strong>{appVersion}\n\n" \
        "---------------------------",
    parse_mode=constants.ParseMode.HTML
)


# Cancel function
@security_user.UsersFirewall
async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    """Cancels and ends the conversation."""

    await update.message.reply_text(
        "Bye! I hope we can talk again some day.", reply_markup=ReplyKeyboardRemove()
    )
    return ConversationHandler.END


def main() -> None:

    # Start the bot.
    # Create the Application and pass it your bot's token.
    application = Application.builder().token(telegramToken).build()

    # On different commands - answer in Telegram
    application.add_handler(CommandHandler("start", Start))
    application.add_handler(CommandHandler("help", HelpCommand))
    application.add_handler(CommandHandler("cancel", cancel))

    # Inline query handler
    application.add_handler(InlineQueryHandler(responses.TextInputInline))

    # Conversation handler to define custom settings
    convHandler1 = ConversationHandler(

        entry_points=[CommandHandler("settings", settings.SettingsMenu)],

        states={
            settings.settingSelected: [MessageHandler(filters.TEXT, settings.ValueAnswer)],
            settings.buttonSelected: [CallbackQueryHandler(settings.Button)],
            settings.customAnswer: [MessageHandler(filters.TEXT, settings.CustomAnswer)]
        },

        fallbacks=[CommandHandler("cancel", cancel)],
        per_chat=True,
        per_user=True
    )
    application.add_handler(convHandler1)

    # on non command i.e VOICE message - reply the message on Telegram
    application.add_handler(MessageHandler(
        filters.VOICE & ~filters.COMMAND, responses.VoiceInput))

    # on non command i.e TEXT message - reply the message on Telegram
    application.add_handler(MessageHandler(
        filters.TEXT & ~filters.COMMAND, responses.TextInput))

    # on non command i.e IMAGE message - reply the message on Telegram
    application.add_handler(MessageHandler(
        filters.PHOTO & ~filters.COMMAND, responses.ImageInput))

    # Run the bot until the user presses Ctrl-C
    application.run_polling()


if __name__ == "__main__":
    db.TestDbConnection()
    main()