refactor: using tuple of class instead of a complex dict

pyproject.toml

@@ -6,6 +6,7 @@ authors = [
 description = "Check phạt nguội"
 name = "check-phat-nguoi"
 version = "0.1.0-dev.1"
+requires-python = ">=3.13"
 dependencies = [
     "aiohttp[speedups]>=3.11.11",
     "pydantic>=2.10.4",

src/check_phat_nguoi/config/dto/config.py

@@ -6,7 +6,7 @@
 
 
 class ConfigDTO(BaseModel):
-    model_config = ConfigDict(use_enum_values=True, validate_desfault=True)
+    model_config = ConfigDict(use_enum_values=True, validate_default=True)
 
     data: tuple[PlateInfoDTO, ...] = Field(
         description="Danh sách các biển xe", default_factory=tuple

src/check_phat_nguoi/context/plate_context/models/plate_info.py

@@ -6,7 +6,7 @@
 class PlateInfoModel(BaseModel):
     plate: str
     owner: str | None
-    violation: tuple[ViolationModel] = Field(
+    violation: tuple[ViolationModel, ...] = Field(
         description="Danh sách các vi phạm của 1 biển xe", default_factory=tuple
     )
 

src/check_phat_nguoi/context/plate_context/models/resolution_office.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel, Field
+
+
+class ResolutionOfficeModel(BaseModel):
+    location_name: str = Field(description="Nơi giải quyết vụ việc")
+    address: str | None = Field(description="Địa chỉ", default=None)
+    phone: str | None = Field(description="Số điện thoại", default=None)

src/check_phat_nguoi/context/plate_context/models/violation.py

@@ -1,10 +1,9 @@
-import re
 from datetime import datetime
 from typing import Dict, Literal
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field
 
-from check_phat_nguoi.utils.constants import OFFICE_NAME_PATTERN
+from .resolution_office import ResolutionOfficeModel
 
 
 class ViolationModel(BaseModel):
@@ -20,29 +19,6 @@ class ViolationModel(BaseModel):
     enforcement_unit: str | None = Field(
         description="Đơn vị phát hiện vi phạm", default=None
     )
-    resolution_office: Dict[str, Dict] | None = Field(
+    resolution_office: tuple[ResolutionOfficeModel, ...] | None = Field(
         description="Nơi giải quyết vụ việc", default=None
     )
-
-    @field_validator("resolution_office", mode="before")
-    def parse_resolution_office(values) -> Dict[str, Dict]:
-        offices_list: list[str] = values["raw_data"]
-        parsed_office_dict: Dict[str, Dict] = []
-        current_name = None
-
-        for office_info in offices_list:
-            if re.match(OFFICE_NAME_PATTERN, office_info):
-                current_name = office_info.split(".", 1)[1].strip()
-                parsed_office_dict[current_name] = {"Address": None, "Phone": None}
-            elif "Địa chỉ" in office_info:
-                if current_name:
-                    parsed_office_dict[current_name]["Address"] = office_info.split(
-                        ":", 1
-                    )[1].strip()
-            elif "Số điện thoại" in office_info:
-                if current_name:
-                    parsed_office_dict[current_name]["Phone"] = office_info.split(
-                        ":", 1
-                    )[1].strip()
-
-        return parsed_office_dict

src/check_phat_nguoi/modules/get_data/check_phat_nguoi.py

@@ -1,17 +1,22 @@
 import asyncio
+import re
 from datetime import datetime
 from logging import getLogger
 from typing import Dict, override
 
-import aiohttp
+from aiohttp import ClientConnectionError, ClientSession, ClientTimeout
+from pydantic_core.core_schema import tuple_positional_schema
 
 from check_phat_nguoi.config import PlateInfoDTO
-from check_phat_nguoi.config.dto.config import ConfigDTO
 from check_phat_nguoi.context import PlateInfoModel, ViolationModel
+from check_phat_nguoi.context.plate_context.models.resolution_office import (
+    ResolutionOfficeModel,
+)
 from check_phat_nguoi.utils.constants import (
     DATETIME_FORMAT_CHECKPHATNGUOI as DATETIME_FORMAT,
 )
 from check_phat_nguoi.utils.constants import GET_DATA_API_URL_CHECKPHATNGUOI as API_URL
+from check_phat_nguoi.utils.constants import OFFICE_NAME_PATTERN
 
 from .get_data_base import GetDataBase
 
@@ -20,19 +25,22 @@
 
 class GetDataCheckPhatNguoi(GetDataBase):
     def __init__(
-        self, plate_infos: tuple[PlateInfoDTO], config: ConfigDTO, timeout: int = 10
+        self, plate_infos: tuple[PlateInfoDTO, ...], timeout: int = 10
     ) -> None:
-        super().__init__(plate_infos, config)
+        super().__init__(plate_infos)
         self.data_dict: Dict[PlateInfoDTO, None | Dict] = {}
-        self.session = aiohttp.ClientSession()
+        self.session = ClientSession()
         self.timeout = timeout
         self.headers = {"Content-Type": "application/json"}
 
     async def _get_data_request(self, plate_info_object: PlateInfoDTO) -> None:
         payload: dict[str, str] = {"bienso": plate_info_object.plate}
         try:
             async with self.session.post(
-                API_URL, headers=self.headers, json=payload, timeout=self.timeout
+                API_URL,
+                headers=self.headers,
+                json=payload,
+                timeout=ClientTimeout(self.timeout),
             ) as response:
                 response.raise_for_status()
 
@@ -42,7 +50,7 @@ async def _get_data_request(self, plate_info_object: PlateInfoDTO) -> None:
             logger.error(
                 f"Time out of {self.timeout} seconds from URL {API_URL} for plate: {plate_info_object.plate}"
             )
-        except aiohttp.ClientConnectionError:
+        except ClientConnectionError:
             logger.error(
                 f"Error occurs while connecting to {API_URL} for plate: {plate_info_object.plate}"
             )
@@ -54,9 +62,38 @@ async def _get_data(self) -> None:
     @staticmethod
     def get_plate_violation(
         plate_violation_dict: Dict | None,
-    ) -> tuple[ViolationModel]:
+    ) -> tuple[ViolationModel, ...]:
         if plate_violation_dict is None:
-            return []
+            return ()
+
+        def _create_resolution_office_mode(
+            resolution_offices: list[str],
+        ) -> tuple[ResolutionOfficeModel, ...]:
+            parsed_office_dict: Dict[str, Dict] = {}
+            current_name = None
+            for office_info in resolution_offices:
+                if re.match(OFFICE_NAME_PATTERN, office_info):
+                    current_name = office_info.split(".", 1)[1].strip()
+                    parsed_office_dict[current_name] = {"Address": None, "Phone": None}
+                elif "Địa chỉ" in office_info:
+                    if current_name:
+                        parsed_office_dict[current_name]["Address"] = office_info.split(
+                            ":", 1
+                        )[1].strip()
+                elif "Số điện thoại" in office_info:
+                    if current_name:
+                        parsed_office_dict[current_name]["Phone"] = office_info.split(
+                            ":", 1
+                        )[1].strip()
+
+            return tuple(
+                ResolutionOfficeModel(
+                    location_name=location_name,
+                    address=location_detail["Address"],
+                    phone=location_detail["Phone"],
+                )
+                for location_name, location_detail in parsed_office_dict.items()
+            )
 
         def _create_violation_model(data: Dict):
             return ViolationModel(
@@ -66,7 +103,9 @@ def _create_violation_model(data: Dict):
                 action=data["Hành vi vi phạm"],
                 status=data["Trạng thái"],
                 enforcement_unit=data["Đơn vị phát hiện vi phạm"],
-                resolution_office=data["Nơi giải quyết vụ việc"],
+                resolution_office=_create_resolution_office_mode(
+                    data["Nơi giải quyết vụ việc"]
+                ),
             )
 
         return tuple(
@@ -75,9 +114,9 @@ def _create_violation_model(data: Dict):
         )
 
     @override
-    async def get_data(self) -> tuple[PlateInfoModel]:
+    async def get_data(self) -> tuple[PlateInfoModel, ...]:
         await self._get_data()
-        plate_infos: tuple[PlateInfoModel] = tuple(
+        plate_infos: tuple[PlateInfoModel, ...] = tuple(
             PlateInfoModel(
                 plate=plate_info_object.plate,
                 owner=plate_info_object.owner,
@@ -86,10 +125,6 @@ async def get_data(self) -> tuple[PlateInfoModel]:
                 ),
             )
             for plate_info_object, plate_violation_dict in self.data_dict.items()
-            if (
-                not self._config.unpaid_only
-                or self.data_dict[plate_info_object]["Trạng thái"] == "Chưa xử phạt"
-            )
         )
         await self.session.close()
 

src/check_phat_nguoi/modules/get_data/get_data_base.py

@@ -6,10 +6,9 @@
 
 
 class GetDataBase:
-    def __init__(self, plate_infos: tuple[PlateInfoDTO], config: ConfigDTO) -> None:
-        self._plate_infos: tuple[PlateInfoDTO] = plate_infos
-        self._config = config
+    def __init__(self, plate_infos: tuple[PlateInfoDTO, ...]) -> None:
+        self._plate_infos: tuple[PlateInfoDTO, ...] = plate_infos
 
     @abstractmethod
-    def get_data(self) -> list[PlateInfoModel]:
+    async def get_data(self) -> tuple[PlateInfoModel, ...]:
         pass

src/check_phat_nguoi/modules/notify/message.py

@@ -1,6 +1,9 @@
 from typing import Dict, LiteralString
 
 from check_phat_nguoi.context import PlateInfoModel, PlatesModel
+from check_phat_nguoi.context.plate_context.models.resolution_office import (
+    ResolutionOfficeModel,
+)
 from check_phat_nguoi.utils.constants import (
     MESSAGE_MARKDOWN_PATTERN,
     RESOLUTION_LOCATION_MARKDOWN_PATTERN,
@@ -12,28 +15,28 @@ def __init__(self, plate_context_object: PlatesModel):
         self._plate_context_object: PlatesModel = plate_context_object
 
     @staticmethod
-    def format_location(locations_info: Dict) -> LiteralString:
-        resolution_markdown: LiteralString = """"""
-        for idx, location_name, location_detail in enumerate(
-            locations_info.items(), start=1
-        ):
-            resolution: LiteralString = RESOLUTION_LOCATION_MARKDOWN_PATTERN.format(
+    def format_location(
+        locations_info: tuple[ResolutionOfficeModel, ...] | None,
+    ) -> str:
+        if locations_info is None:
+            return ""
+        resolution_markdown: str = ""
+        for idx, location_detail in enumerate(locations_info, start=1):
+            resolution: str = RESOLUTION_LOCATION_MARKDOWN_PATTERN.format(
                 idx=idx,
-                location_name=location_name,
-                address=location_detail["Address"]
-                if location_detail["Address"]
-                else "Không có",
-                phone=location_detail["Phone"]
-                if location_detail["Phone"]
+                location_name=location_detail.location_name,
+                address=location_detail.address
+                if location_detail.address
                 else "Không có",
+                phone=location_detail.phone if location_detail.phone else "Không có",
             )
             resolution_markdown += resolution + "\n"
         return resolution_markdown
 
     @staticmethod
     def format_message(
         plate_info_context: PlateInfoModel,
-    ) -> tuple[LiteralString]:
+    ) -> tuple[str, ...]:
         return tuple(
             MESSAGE_MARKDOWN_PATTERN.format(
                 plate=plate_info_context.plate,
@@ -48,8 +51,8 @@ def format_message(
             if vio.status
         )
 
-    def format_messages(self) -> dict[str, tuple[LiteralString]]:
-        message_dict: dict[str, tuple[LiteralString]] = {}
+    def format_messages(self) -> dict[str, tuple[str, ...]]:
+        message_dict: dict[str, tuple[str, ...]] = {}
         for plate_info_context in self._plate_context_object.plates:
             message_dict[plate_info_context.plate] = Message.format_message(
                 plate_info_context

src/check_phat_nguoi/utils/constants.py

@@ -1,19 +1,23 @@
-CONFIG_PATH: str = "config.json"
-SIMPLE_LOG_MESSAGE: str = "[%(levelname)s]: %(message)s"
-DETAIL_LOG_MESSAGE: str = (
+from typing import LiteralString
+
+CONFIG_PATH: LiteralString = "config.json"
+SIMPLE_LOG_MESSAGE: LiteralString = "[%(levelname)s]: %(message)s"
+DETAIL_LOG_MESSAGE: LiteralString = (
     "%(asctime)s [%(levelname)s] - %(message)s (%(filename)s:%(lineno)d)"
 )
 
 # API from checkphatnguoi.vn
-GET_DATA_API_URL_CHECKPHATNGUOI: str = "https://api.checkphatnguoi.vn/phatnguoi"
-SEND_MESSAGE_API_URL_TELEGRAM: str = (
+GET_DATA_API_URL_CHECKPHATNGUOI: LiteralString = (
+    "https://api.checkphatnguoi.vn/phatnguoi"
+)
+SEND_MESSAGE_API_URL_TELEGRAM: LiteralString = (
     "https://api.telegram.org/bot{bot_token}/sendMessage"
 )
-DATETIME_FORMAT_CHECKPHATNGUOI: str = "%H:%M, %d/%m/%Y"
+DATETIME_FORMAT_CHECKPHATNGUOI: LiteralString = "%H:%M, %d/%m/%Y"
 
 OFFICE_NAME_PATTERN = r"^\d+\."
 
-MESSAGE_MARKDOWN_PATTERN = """
+MESSAGE_MARKDOWN_PATTERN: LiteralString = """
 *🚗 **Thông tin phương tiện**:*
 - **Biển kiểm soát:** `{plate}`
 - **Chủ sở hữu:** `{owner}'
@@ -31,7 +35,7 @@
 {resolution_locations}
 """
 
-RESOLUTION_LOCATION_MARKDOWN_PATTERN = """
+RESOLUTION_LOCATION_MARKDOWN_PATTERN: LiteralString = """
 {idx}. **{location_name}
 - **Địa chỉ:** {address}
 - **Số điện thoại liên lạc:** {phone}

src/generate_schemas/modules/config.py

@@ -3,7 +3,7 @@
 from pydantic import TypeAdapter
 
 from check_phat_nguoi.config import ConfigDTO
-from generate_schemas.utils.constant import CONFIG_SCHEMA_PATH
+from generate_schemas.utils.constants import CONFIG_SCHEMA_PATH
 
 
 def generate_config_schema():

src/generate_schemas/utils/constants.py

@@ -0,0 +1,3 @@
+from typing import LiteralString
+
+CONFIG_SCHEMA_PATH: LiteralString = "schemas/config.json"