feat: get data from zmio-api

config.sample.json

@@ -21,6 +21,12 @@
       "apis": "checkphatnguoi.vn",
       "type": 1
     },
+    {
+      "plate": "29A34809",
+      "owner": "Oreo",
+      "apis": "etraffic.gtelict.vn",
+      "type": 1
+    },
     {
       "plate": "98A56604",
       "type": "car",

src/check_phat_nguoi/constants/__init__.py

@@ -4,6 +4,7 @@
     API_URL_CSGT_QUERY_1,
     API_URL_CSGT_QUERY_2,
     API_URL_PHATNGUOI,
+    API_URL_ZM_IO,
     DATETIME_FORMAT_CHECKPHATNGUOI,
     GET_DATA_API_URL_CHECKPHATNGUOI,
     OFFICE_NAME_PATTERN,
@@ -28,4 +29,5 @@
     "MESSAGE_MARKDOWN_PATTERN",
     "RESOLUTION_LOCATION_MARKDOWN_PATTERN",
     "API_URL_PHATNGUOI",
+    "API_URL_ZM_IO",
 ]

src/check_phat_nguoi/constants/get_data.py

@@ -12,7 +12,7 @@
     "https://www.csgt.vn/?mod=contact&task=tracuu_post&ajax"
 )
 API_URL_CSGT_QUERY_2: LiteralString = "https://www.csgt.vn/tra-cuu-phuong-tien-vi-pham.html?&LoaiXe={vehicle_type}&BienKiemSoat={plate}"
-
+API_URL_ZM_IO: LiteralString = "https://api.zm.io.vn/v1/csgt/tracuu"
 DATETIME_FORMAT_CHECKPHATNGUOI: LiteralString = "%H:%M, %d/%m/%Y"
 
 OFFICE_NAME_PATTERN: LiteralString = r"^\d+\."

src/check_phat_nguoi/get_data/engines/__init__.py

@@ -1,9 +1,11 @@
 from .base import BaseGetDataEngine
 from .check_phat_nguoi import CheckPhatNguoiGetDataEngine
 from .csgt import CsgtGetDataEngine
+from .zm_io import ZMIOGetDataEngine
 
 __all__ = [
     "BaseGetDataEngine",
     "CheckPhatNguoiGetDataEngine",
     "CsgtGetDataEngine",
+    "ZMIOGetDataEngine",
 ]

src/check_phat_nguoi/get_data/engines/zm_io.py

@@ -0,0 +1,120 @@
+from asyncio import TimeoutError
+from datetime import datetime
+from logging import getLogger
+from typing import Literal, TypedDict, cast, override
+
+from aiohttp import ClientError
+
+from check_phat_nguoi.config import PlateInfo
+from check_phat_nguoi.constants import API_URL_ZM_IO
+from check_phat_nguoi.constants import DATETIME_FORMAT_CHECKPHATNGUOI as DATETIME_FORMAT
+from check_phat_nguoi.context import PlateDetail, ViolationDetail
+from check_phat_nguoi.types import (
+    ApiEnum,
+    VehicleTypeEnum,
+    get_vehicle_enum,
+)
+from check_phat_nguoi.utils import HttpaioSession
+
+from .base import BaseGetDataEngine
+
+logger = getLogger(__name__)
+
+
+class _DataResponse(TypedDict):
+    bienkiemsoat: str
+    maubien: str
+    loaiphuongtien: Literal["Ô tô", "Xe máy", "Xe máy điện"]
+    thoigianvipham: str
+    diadiemvipham: str
+    trangthai: str
+    donviphathienvipham: str
+    noigiaiquyetvuviec: str
+
+
+class _Response(TypedDict):
+    json: tuple[_DataResponse, ...] | None
+    html: str
+    css: str
+
+
+class _ZMIOGetDataParseEngine:
+    def __init__(self, plate_info: PlateInfo, plate_detail_dict: _Response) -> None:
+        self._plate_info: PlateInfo = plate_info
+        self._plate_detail_typed: _Response = plate_detail_dict
+        self._violations_details_set: set[ViolationDetail] = set()
+
+    def _parse_violation(self, data: _DataResponse) -> None:
+        plate: str = data["bienkiemsoat"]
+        date: str = data["thoigianvipham"]
+        type: Literal["Ô tô", "Xe máy", "Xe máy điện"] = data["loaiphuongtien"]
+        color: str = data["maubien"]
+        location: str = data["diadiemvipham"]
+        status: str = data["trangthai"]
+        enforcement_unit: str = data["donviphathienvipham"]
+        # NOTE: this api just responses 1 resolution_office
+        resolution_offices: tuple[str, ...] = (data["noigiaiquyetvuviec"],)
+        violation_detail: ViolationDetail = ViolationDetail(
+            plate=plate,
+            color=color,
+            type=get_vehicle_enum(type),
+            date=datetime.strptime(str(date), DATETIME_FORMAT),
+            location=location,
+            status=status == "Đã xử phạt",
+            enforcement_unit=enforcement_unit,
+            resolution_offices=resolution_offices,
+        )
+        self._violations_details_set.add(violation_detail)
+
+    def parse(self) -> tuple[ViolationDetail, ...] | None:
+        if not self._plate_detail_typed["json"]:
+            return
+        for data in self._plate_detail_typed["json"]:
+            self._parse_violation(data)
+        return tuple(self._violations_details_set)
+
+
+class ZMIOGetDataEngine(HttpaioSession, BaseGetDataEngine):
+    api = ApiEnum.zm_io_vn
+
+    def __init__(self):
+        HttpaioSession.__init__(self)
+
+    async def _request(self, plate_info: PlateInfo) -> dict | None:
+        url = f"{API_URL_ZM_IO}?licensePlate={plate_info.plate}&vehicleType={get_vehicle_enum(plate_info.type)}"
+        try:
+            async with self._session.get(url) as response:
+                json = await response.json()
+                return json["data"]
+        except TimeoutError as e:
+            logger.error(
+                f"Plate {plate_info.plate}: Time out ({self.timeout}s) getting data from API {self.api.value}. {e}"
+            )
+        except ClientError as e:
+            logger.error(
+                f"Plate {plate_info.plate}: Error occurs while getting data from API {self.api.value}. {e}"
+            )
+        except Exception as e:
+            logger.error(
+                f"Plate {plate_info.plate}: Error occurs while getting data (internally) {self.api.value}. {e}"
+            )
+
+    @override
+    async def get_data(self, plate_info: PlateInfo) -> PlateDetail | None:
+        plate_detail_raw: dict | None = await self._request(plate_info)
+        if not plate_detail_raw:
+            return
+        plate_detail_typed: _Response = cast(_Response, plate_detail_raw)
+        type: VehicleTypeEnum = get_vehicle_enum(plate_info.type)
+        return PlateDetail(
+            plate=plate_info.plate,
+            owner=plate_info.owner,
+            type=type,
+            violations=_ZMIOGetDataParseEngine(
+                plate_info=plate_info, plate_detail_dict=plate_detail_typed
+            ).parse(),
+        )
+
+    @override
+    async def __aexit__(self, exc_type, exc_value, exc_traceback) -> None:
+        return await HttpaioSession.__aexit__(self, exc_type, exc_value, exc_traceback)

src/check_phat_nguoi/get_data/get_data.py

@@ -10,7 +10,12 @@
 from check_phat_nguoi.get_data.engines.phat_nguoi import PhatNguoiGetDataEngine
 from check_phat_nguoi.types import ApiEnum
 
-from .engines import BaseGetDataEngine, CheckPhatNguoiGetDataEngine, CsgtGetDataEngine
+from .engines import (
+    BaseGetDataEngine,
+    CheckPhatNguoiGetDataEngine,
+    CsgtGetDataEngine,
+    ZMIOGetDataEngine,
+)
 
 logger = getLogger(__name__)
 
@@ -20,6 +25,7 @@ def __init__(self) -> None:
         self._checkphatnguoi_engine: CheckPhatNguoiGetDataEngine
         self._csgt_engine: CsgtGetDataEngine
         self._phatnguoi_engine: PhatNguoiGetDataEngine
+        self._zmio_engine: ZMIOGetDataEngine
         self._plates_details: set[PlateDetail] = set()
 
     async def _get_data_for_plate(self, plate_info: PlateInfo) -> None:
@@ -34,6 +40,8 @@ async def _get_data_for_plate(self, plate_info: PlateInfo) -> None:
                     engine = self._csgt_engine
                 case ApiEnum.phatnguoi_vn:
                     engine = self._phatnguoi_engine
+                case ApiEnum.zm_io_vn:
+                    engine = self._zmio_engine
             logger.info(
                 f"Plate {plate_info.plate}: Getting data with API: {api.value}..."
             )
@@ -55,6 +63,7 @@ async def get_data(self) -> None:
             CheckPhatNguoiGetDataEngine() as self._checkphatnguoi_engine,
             CsgtGetDataEngine() as self._csgt_engine,
             PhatNguoiGetDataEngine() as self._phatnguoi_engine,
+            ZMIOGetDataEngine() as self._zmio_engine,
         ):
             if config.asynchronous:
                 await gather(

src/check_phat_nguoi/types/api.py

@@ -5,6 +5,7 @@ class ApiEnum(str, Enum):
     checkphatnguoi_vn = "checkphatnguoi.vn"
     csgt_vn = "csgt.vn"
     phatnguoi_vn = "phatnguoi.vn"
+    zm_io_vn = "zm.io.vn"
 
 
 __all__ = ["ApiEnum"]