examples/chat_monitoring_bot/main.py - master - BorisPlus/Telegram-Bot-Discussion-Examples

examples/chat_monitoring_bot/main.py
162 строки4 KB
2025_10_13_initial
13 окт 2025, 00:05
13 окт 2025, 00:050fc4405
from optparse import OptionParser


from telegram import (
    InlineKeyboardMarkup,
    Update,
)
from telegram.ext import CallbackContext


from telegram_bot_discussion import Discussion, Contextual
from telegram_bot_discussion.button import Button, ButtonsMap, ButtonMeta
from telegram_bot_discussion.button.params import Params
from telegram_bot_discussion.config import Config
from telegram_bot_discussion.ext.coder import BSV_Coder
from telegram_bot_discussion.ext.logger import LogLevel, Printer
from telegram_bot_discussion.functions import (
    get_chat_id as get_chat_id,
    get_from_id as get_from_id,
    get_message as get_message,
)
from telegram_bot_discussion.handlers import ButtonsHandler
from telegram_bot_discussion.handlers import ReplicasHandler
from telegram_bot_discussion.handlers.buttons_handler import (
    ButtonDataSignificationError,
)


class UnderstandButtonParams(Params):
    action: str = ""


class UnderstandButton(Button):

    class Meta(ButtonMeta):
        title: str = "OK, I understand."

    params: UnderstandButtonParams

    def __init__(
        self,
        chat_id: int,
        recipient_id: int,
    ):
        self.params = UnderstandButtonParams(
            action=self.action(),
        )
        super().__init__(
            chat_id,
            recipient_id,
        )

    @classmethod
    async def on_click(
        cls,
        update: Update,
        _: CallbackContext,
    ):
        await update.callback_query.answer(
            text="With hope for the best 🤝",
            show_alert=True,
        )
        await update.callback_query.delete_message()


class MonitoringBotButtonsHandler(ButtonsHandler):
    async def handle(
        self,
        update: Update,
        context: CallbackContext,
    ):
        try:
            _ = await super().handle(
                update,
                context,
            )
        except ButtonDataSignificationError as e:
            await update.callback_query.answer(
                text="It's not your alert!",
                show_alert=True,
            )
        except Exception as e:
            raise e


class MonitoringBotReplicasHandler(ReplicasHandler):
    async def when_there_are_no_any_dialog(
        self,
        update: Update,
        context: CallbackContext,
    ):
        discussion: Discussion = Contextual[Discussion](context)()
        discussion.get_logger().info(update)
        message = get_message(update)
        chat_id = get_chat_id(update)
        from_id = get_from_id(update)
        if message.text and (
            "password" in message.text.lower()
            or "psw" in message.text.lower()
            or "passwd" in message.text.lower()
        ):
            await discussion.bot().send_message(
                chat_id,
                "Don't send sensitive information.",
                reply_markup=InlineKeyboardMarkup(
                    [
                        [
                            UnderstandButton(
                                chat_id=chat_id,
                                recipient_id=from_id,
                            ).as_button(
                                discussion.get_buttons_handler().coder,
                            )
                        ]
                    ]
                ),
                reply_to_message_id=message.id,
            )


def main():
    usage = "usage: %prog -t/--token <token>"

    parser = OptionParser(usage=usage)
    parser.add_option(
        "-t",
        "--token",
        help="Telegram bot-API token",
    )
    parser.add_option(
        "-s",
        "--secret",
        help="Secret phrase",
    )
    (options, _) = parser.parse_args()

    discussion = (
        Discussion(
            logger=Printer(
                name="chat_monitoring_bot",
                level=LogLevel.DEBUG,
            ),
            config=Config(token=options.token),
        )
        .set_replicas_handler(MonitoringBotReplicasHandler())
        .set_buttons_handler(
            MonitoringBotButtonsHandler(
                coder=BSV_Coder(options.secret),
                buttons_map=ButtonsMap(
                    UnderstandButton,
                ),
            )
        )
    )

    discussion.start()


# cd examples
# python3 chat_monitoring_bot/main.py -t <token>
if __name__ == "__main__":
    main()