Initial commit

app/core/models/_telegram/__init__.py

@@ -0,0 +1 @@
+from .wrapped_bot import Wrapped_CBotChat

app/core/models/_telegram/wrapped_bot.py

@@ -0,0 +1,129 @@
+from aiogram import Bot
+
+from app.core.logger import make_log
+
+
+class Wrapped_CBotChat:
+    def __init__(self, api_key: str, chat_id: int = None, **kwargs):
+        if isinstance(api_key, Bot):
+            self._bot_key = api_key.token
+            self._bot = api_key
+        elif type(api_key) is str:
+            self._bot_key = api_key
+            self._bot = Bot(token=api_key)
+        else:
+            raise TypeError(f'api_key must be Bot or str, not {type(api_key)}')
+
+        self._chat_id = chat_id
+        self.options = kwargs
+
+    def __repr__(self):
+        if self.options.get('cbot_id'):
+            return f'Bot, chat_id={self.options["cbot_id"]} ' + '\\'
+
+        return "Bot"
+
+    async def send_message(self, text: str, **kwargs):
+        try:
+            make_log(self, f"Send message to {self._chat_id}. Text len: {len(text)}", level='debug')
+            return await self._bot.send_message(
+                self._chat_id,
+                text,
+                parse_mode='html',
+                disable_web_page_preview=True,
+                **kwargs
+            )
+        except BaseException as e:
+            make_log(self, f"Error sending message to {self._chat_id}. Error: {e}", level='warning')
+            return None
+
+    async def edit_message(self, message_id, text, **kwargs):
+        try:
+            make_log(self, f"Edit message {self._chat_id}/{message_id}. Text len: {len(text)}", level='debug')
+            return await self._bot.edit_message_text(
+                text,
+                chat_id=self._chat_id,
+                message_id=message_id,
+                parse_mode='html',
+                disable_web_page_preview=True,
+                **kwargs
+            )
+        except Exception as e:
+            make_log(self, f"Error editing message {self._chat_id}/{message_id}. Error: {e}", level='warning')
+            if 'exactly the same as a current content' in f'{e}':
+                return True
+
+            return None
+
+    async def delete_message(self, message_id):
+        try:
+            make_log(self, f"Delete message {self._chat_id}/{message_id}", level='debug')
+            return await self._bot.delete_message(
+                self._chat_id,
+                message_id
+            )
+        except Exception as e:
+            make_log(self, f"Error deleting message {self._chat_id}/{message_id}. Error: {e}", level='warning')
+            return None
+
+    async def send_photo(self, file_id, **kwargs):
+        try:
+            make_log(self, f"Send photo to {self._chat_id}. File: {file_id}", level='debug')
+            return await self._bot.send_photo(
+                self._chat_id,
+                file_id,
+                **kwargs
+            )
+        except Exception as e:
+            make_log(self, f"Error sending photo to {self._chat_id}. Error: {e}", level='warning')
+            return None
+
+    async def send_document(self, file_id, **kwargs):
+        try:
+            make_log(self, f"Send document to {self._chat_id}. File: {file_id}", level='debug')
+            return await self._bot.send_document(
+                self._chat_id,
+                file_id,
+                **kwargs
+            )
+        except Exception as e:
+            make_log(self, f"Error sending document to {self._chat_id}. Error: {e}", level='warning')
+            return None
+
+    async def send_video(self, file_id, **kwargs):
+        try:
+            make_log(self, f"Send video to {self._chat_id}. File: {file_id}", level='debug')
+            return await self._bot.send_video(
+                self._chat_id,
+                file_id,
+                **kwargs
+            )
+        except Exception as e:
+            make_log(self, f"Error sending video to {self._chat_id}. Error: {e}", level='warning')
+            return None
+
+    async def copy_message(self, from_chat_id, message_id, **kwargs):
+        try:
+            make_log(self, f"Copy message from {from_chat_id}/{message_id} to {self._chat_id}", level='debug')
+            return await self._bot.copy_message(
+                self._chat_id,
+                from_chat_id,
+                message_id,
+                **kwargs
+            )
+        except Exception as e:
+            make_log(self, f"Error copying message from {from_chat_id}/{message_id} to {self._chat_id}. Error: {e}", level='warning')
+            return None
+
+    async def forward_message(self, from_chat_id, message_id, **kwargs):
+        try:
+            make_log(self, f"Forward message from {from_chat_id}/{message_id} to {self._chat_id}", level='debug')
+            return await self._bot.forward_message(
+                self._chat_id,
+                from_chat_id,
+                message_id,
+                **kwargs
+            )
+        except Exception as e:
+            make_log(self, f"Error forwarding message from {from_chat_id}/{message_id} to {self._chat_id}. Error: {e}", level='warning')
+            return None