feat!: send messages to telegram

src/check_phat_nguoi/modules/notify/telegram.py

@@ -0,0 +1,48 @@
+from typing import LiteralString
+from check_phat_nguoi.models.notify.telegram_notify import TelegramNotifyModel
+from check_phat_nguoi.utils.constants import SEND_MESSAGE_API_URL_TELEGRAM as API_URL
+import requests
+from logging import getLogger
+from threading import Thread
+
+logger = getLogger(__name__)
+
+
+class Telegram:
+    def __init__(
+        self,
+        telegram_notify_object: TelegramNotifyModel,
+        message_dict: dict[str, LiteralString],
+    ):
+        self._telegram_notify_object: TelegramNotifyModel = telegram_notify_object
+        self._message_dict: dict[str, LiteralString] = message_dict
+
+    def _send_message(self, message: LiteralString, timeout=10) -> None:
+        url = API_URL.format(bot_token=self._telegram_notify_object.telegram.bot_token)
+        payload = {
+            "chat_id": self._telegram_notify_object.telegram.chat_id,
+            "text": message,
+            "parse_mode": "Markdown",
+        }
+        try:
+            response = requests.post(url, json=payload)
+            response.raise_for_status()
+            logger.info(f"Request successful: {response.status_code}")
+        except requests.exceptions.ConnectionError:
+            logger.error(f"Unable to connect to {url}")
+        except requests.exceptions.Timeout:
+            logger.error(f"Time out of {timeout} seconds from URL {url}")
+
+    def mutithread_send_message(self) -> None:
+        threads: list[Thread] = []
+        for _, message in self._message_dict.items():
+            thread = Thread(target=self._send_message, args=(message))
+            threads.append(thread)
+            thread.start()
+        for idx, thread in enumerate(threads, start=1):
+            try:
+                thread.join()
+            except Exception:
+                logger.error(
+                    f"An error occurs while sending message in thread number {idx}"
+                )

src/check_phat_nguoi/utils/constants.py

@@ -6,6 +6,9 @@
 
 # API from checkphatnguoi.vn
 GET_DATA_API_URL_CHECKPHATNGUOI: str = "https://api.checkphatnguoi.vn/phatnguoi"
+SEND_MESSAGE_API_URL_TELEGRAM: str = (
+    "https://api.telegram.org/bot{bot_token}/sendMessage"
+)
 DATETIME_FORMAT_CHECKPHATNGUOI: str = "%H:%M, %d/%m/%Y"
 
 OFFICE_NAME_PATTERN = r"^\d+\."