Merge pull request #2 from TimNekk/develop

feat: add ping command 
fix: fix command filter

tgbot/config.py

@@ -21,6 +21,7 @@ def __post_init__(self) -> None:
 @dataclass
 class Commands:
     send_all: CommandInfo
+    ping: CommandInfo
 
     def __iter__(self) -> Generator[CommandInfo, None, None]:
         return (getattr(self, field.name) for field in fields(self))
@@ -86,6 +87,7 @@ def load_config(path: str | None = None) -> Config:
             use_redis=env.bool("USE_REDIS"),
             commands=Commands(
                 send_all=CommandInfo("send_all", "Рассылка", is_admin=True),
+                ping=CommandInfo("ping", "Пинг", is_admin=True),
             )
         ),
         db=DbConfig(

tgbot/filters/command.py

@@ -15,4 +15,4 @@ async def check(self, obj: types.base.TelegramObject) -> bool:
         if not isinstance(obj, types.Message):
             raise NotImplementedError("CommandFilter can only be used with Message")
         message: types.Message = obj
-        return f"/{self.command.command}" == message.text or self.command.alias == message.text
+        return message.text.startswith(f"/{self.command.command}") or self.command.alias == message.text

tgbot/handlers/admin/__init__.py

@@ -1,7 +1,9 @@
 from aiogram import Dispatcher
 
 from .send_all import register_send_all_handlers
+from .ping import register_ping_handlers
 
 
 def register_admin_handlers(dp: Dispatcher) -> None:
     register_send_all_handlers(dp)
+    register_ping_handlers(dp)

tgbot/handlers/admin/ping.py

@@ -0,0 +1,80 @@
+from datetime import datetime
+
+from aiogram import types, Dispatcher
+from aiogram.dispatcher import FSMContext
+
+from tgbot.config import Config
+from tgbot.models.user_tg import UserTG
+from tgbot.states import PingState
+
+
+async def ping(message: types.Message, state: FSMContext, user: UserTG, config: Config) -> None:
+    delta = get_time_delta(message.date)
+    aim = 10
+    results = [delta]
+
+    args = message.text.split()[1:]
+    if args and args[0].isdigit():
+        aim = int(args[0])
+        if aim < 1:
+            await user.send_message("<b>Минимальное количество раз:</b> 1")
+            return
+
+    text = get_request_text(config.tg_bot.commands.ping.command, aim, len(results) - 1)
+    base_message = await user.send_message(text)
+
+    await PingState.waiting_for_ping.set()
+    await state.update_data(aim=aim,
+                            results=results,
+                            base_message_id=base_message.message_id)
+
+
+async def get_ping_data(message: types.Message, state: FSMContext, user: UserTG, config: Config) -> None:
+    delta = get_time_delta(message.date)
+
+    await message.delete()
+
+    data = await state.get_data()
+    aim: int = data.get("aim")
+    results: list[float] = data.get("results")
+    base_message_id: int = data.get("base_message_id")
+
+    if results and len(results) < aim:
+        results.append(delta)
+        await state.update_data(results=results)
+        text = get_request_text(config.tg_bot.commands.ping.command, aim, len(results) - 1)
+        await user.edit_message_text(text, base_message_id)
+    else:
+        await user.delete_message(base_message_id)
+        average_ping = round(sum(results) / len(results) * 1000)
+        circle = get_color_circle(average_ping)
+        await user.send_message(f"{circle} <b>Средний пинг:</b> <code>{average_ping}</code> ms")
+        await state.finish()
+
+
+def get_request_text(commend: str, aim: int, results_len: int) -> str:
+    return f"Отправьте /{commend} еще {aim - results_len} раз"
+
+
+def get_time_delta(receive_time: datetime) -> float:
+    return (datetime.now() - receive_time).total_seconds()
+
+
+def get_color_circle(average_ping: int) -> str:
+    if average_ping <= 1200:
+        return "🟢"
+    elif average_ping <= 2000:
+        return "🟡"
+    else:
+        return "🔴"
+
+
+def register_ping_handlers(dp: Dispatcher) -> None:
+    config: Config = dp.bot.get("config")
+    dp.register_message_handler(ping,
+                                command=config.tg_bot.commands.ping,
+                                is_admin=True)
+    dp.register_message_handler(get_ping_data,
+                                command=config.tg_bot.commands.ping,
+                                is_admin=True,
+                                state=PingState.waiting_for_ping)

tgbot/models/user_tg.py

@@ -658,3 +658,17 @@ async def edit_message_live_location(self,
                                             heading=heading,
                                             proximity_alert_radius=proximity_alert_radius,
                                             reply_markup=reply_markup)
+
+    async def delete_message(self,
+                             message_id: typing.Optional[base.Integer] = None,
+                             ) -> base.Boolean | None:
+        try:
+            return await self.bot.delete_message(self.id, message_id)
+        except (exceptions.MessageToDeleteNotFound, exceptions.MessageCantBeDeleted) as e:
+            logger.exception(f"{self}: {e.match}")
+        except (exceptions.BotBlocked, exceptions.ChatNotFound, exceptions.UserDeactivated) as e:
+            logger.debug(f"{self}: {e.match}")
+            await self.update(is_banned=True).apply()
+        except exceptions.TelegramAPIError as e:
+            logger.exception(f"{self}: {e}")
+        return None

tgbot/states/__init__.py

@@ -1 +1,2 @@
 from .send_all import SendAllState
+from .ping import PingState

tgbot/states/ping.py

@@ -0,0 +1,5 @@
+from aiogram.dispatcher.filters.state import State, StatesGroup
+
+
+class PingState(StatesGroup):
+    waiting_for_ping = State()