fix(message): resolve KeyError in string formatting

src/check_phat_nguoi/__init__.py

@@ -3,7 +3,10 @@
 
 from check_phat_nguoi.config import config
 from check_phat_nguoi.config.config_reader import _config_reader
+from check_phat_nguoi.context.plate_context.models.plate import PlatesModel
 from check_phat_nguoi.get_data.check_phat_nguoi import GetDataCheckPhatNguoi
+from check_phat_nguoi.notify.message import Message
+from check_phat_nguoi.notify.telegram import Telegram
 
 from .utils.setup_logger import setup_logger
 
@@ -15,6 +18,12 @@ async def _main():
     logger.debug(config)
     get_data_object = GetDataCheckPhatNguoi(_config_reader().data)
     data = await get_data_object.get_data()
+    plate_object = PlatesModel(plates=data)
+    message_object = Message(plate_context_object=plate_object)
+    message_dict = message_object.format_messages()
+    for notify_object in _config_reader().notify:
+        telegram_object = Telegram(notify_object, message_dict)
+        await telegram_object.send_messages()
 
 
 def main():

src/check_phat_nguoi/modules/constants/notify.py

@@ -1,29 +1,30 @@
+from string import Template
 from typing import LiteralString
 
 SEND_MESSAGE_API_URL_TELEGRAM: LiteralString = (
     "https://api.telegram.org/bot{bot_token}/sendMessage"
 )
-MESSAGE_MARKDOWN_PATTERN: LiteralString = """
+MESSAGE_MARKDOWN_PATTERN = Template("""
 *🚗 **Thông tin phương tiện**:*
-- **Biển kiểm soát:** `{plate}`
-- **Chủ sở hữu:** `{owner}'
+- **Biển kiểm soát:** `${plate}`
+- **Chủ sở hữu:** `${owner}`
 
 *⚠️ **Thông tin vi phạm**:*
-- **Hành vi vi phạm:** `{action}`
-- **Trạng thái:** {status}
-- **Thời gian vi phạm:** `{date}`
-- **Địa điểm vi phạm** {location}
+- **Hành vi vi phạm:** `${action}`
+- **Trạng thái:** ${status}
+- **Thời gian vi phạm:** `${date}`
+- **Địa điểm vi phạm:** `${location}`
 
 *🏢 **Đơn vị phát hiện vi phạm**:*
-- **{enforcement_unit}**
+- **`${enforcement_unit}`**
 
 *📍 **Nơi giải quyết vụ việc**:*
-{resolution_locations}
-"""
+${resolution_locations}
+""")
 
-RESOLUTION_LOCATION_MARKDOWN_PATTERN: LiteralString = """
-{idx}. **{location_name}
-- **Địa chỉ:** {address}
-- **Số điện thoại liên lạc:** {phone}
+RESOLUTION_LOCATION_MARKDOWN_PATTERN = Template("""
+${idx}. **${location_name}
+- **Địa chỉ:** `${address}`
+- **Số điện thoại liên lạc:** `${phone}`
 
-"""
+""")

src/check_phat_nguoi/notify/message.py

@@ -21,7 +21,7 @@ def format_location(
             return ""
         resolution_markdown: str = ""
         for idx, location_detail in enumerate(locations_info, start=1):
-            resolution: str = RESOLUTION_LOCATION_MARKDOWN_PATTERN.format(
+            resolution: str = RESOLUTION_LOCATION_MARKDOWN_PATTERN.substitute(
                 idx=idx,
                 location_name=location_detail.location_name,
                 address=location_detail.address
@@ -37,17 +37,22 @@ def format_message(
         plate_info_context: PlateInfoModel,
     ) -> tuple[str, ...]:
         return tuple(
-            MESSAGE_MARKDOWN_PATTERN.format(
-                plate=plate_info_context.plate,
-                owner=plate_info_context.owner,
-                action=vio.action,
-                status=vio.status,
-                date=f"{vio.date}",
-                location=vio.enforcement_unit,
-                resolution_locations=Message.format_location(vio.resolution_office),
-            )
-            for vio in plate_info_context.violation
-            if vio.status
+            [
+                MESSAGE_MARKDOWN_PATTERN.substitute(
+                    plate=plate_info_context.plate,
+                    owner="Không biết"
+                    if not plate_info_context.owner
+                    else plate_info_context.owner,
+                    action=vio.action,
+                    status="Đã nộp phạt" if vio.status else "Chưa nộp phạt",
+                    date=f"{vio.date}",
+                    location=vio.location,
+                    enforcement_unit=vio.enforcement_unit,
+                    resolution_locations=Message.format_location(vio.resolution_office),
+                )
+                for vio in plate_info_context.violation
+                if vio.status
+            ]
         )
 
     def format_messages(self) -> dict[str, tuple[str, ...]]:

src/check_phat_nguoi/notify/telegram.py

@@ -1,6 +1,5 @@
 import asyncio
 from logging import getLogger
-from typing import LiteralString
 
 from aiohttp import ClientConnectionError, ClientSession, ClientTimeout
 
@@ -15,13 +14,13 @@ class Telegram:
     def __init__(
         self,
         telegram_notify: TelegramNotifyDTO,
-        message_dict: dict[str, LiteralString],
+        message_dict: dict[str, tuple[str, ...]],
     ):
         self._telegram_notify_object: TelegramNotifyDTO = telegram_notify
-        self._message_dict: dict[str, LiteralString] = message_dict
+        self._message_dict: dict[str, tuple[str, ...]] = message_dict
         self.timeout = 10
 
-    async def _send_message(self, message: LiteralString) -> None:
+    async def _send_message(self, message: str) -> None:
         if not self._telegram_notify_object.enabled:
             logger.info("Not enable to sending")
             return
@@ -57,14 +56,17 @@ async def _send_message(self, message: LiteralString) -> None:
                     bot_token=self._telegram_notify_object.telegram.bot_token,
                 )
             )
+        except Exception as e:
+            logger.error(e)
         finally:
             await session.close()
 
     async def send_messages(self) -> None:
         async def _concurent_send_messages():
-            tasks = [
-                self._send_message(message) for _, message in self._message_dict.items()
-            ]
+            tasks = []
+            for _, message_tuple in self._message_dict.items():
+                for message in message_tuple:
+                    tasks.append(self._send_message(message))
             await asyncio.gather(*tasks)
 
         await _concurent_send_messages()