diff --git a/.github/workflows/release-publish.yml b/.github/workflows/release-publish.yml index 656dd7f..b8b5153 100644 --- a/.github/workflows/release-publish.yml +++ b/.github/workflows/release-publish.yml @@ -80,5 +80,3 @@ jobs: - name: Publish package distributions to PyPI uses: pypa/gh-action-pypi-publish@release/v1 - # with: - # password: ${{ secrets.PYPI_TOKEN }} diff --git a/pyproject.toml b/pyproject.toml index 9bde50b..12ec286 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,6 +2,12 @@ authors = [ { name = "KevinNitroG", email = "kevinnitro@duck.com" }, { name = "NTGNguyen", email = 'ntgnguyen@duck.com' }, + { name = "WeeCiCi", email = 'wicici310@gmail.com' }, +] +maintainers = [ + { name = "KevinNitroG", email = "kevinnitro@duck.com" }, + { name = "NTGNguyen", email = 'ntgnguyen@duck.com' }, + { name = "WeeCiCi", email = 'wicici310@gmail.com' }, ] description = "cpn core" name = "cpn-core" @@ -17,7 +23,7 @@ dependencies = [ [project.optional-dependencies] discord = ["audioop-lts>=0.2.1", "discord-py>=2.4.0"] ocr = ["pytesseract>=0.3.13"] - +curl = ["curl-cffi>=0.7.4"] [build-system] requires = ["hatchling"] build-backend = "hatchling.build" diff --git a/src/cpn_core/get_data/etraffic.py b/src/cpn_core/get_data/etraffic.py new file mode 100644 index 0000000..221ee58 --- /dev/null +++ b/src/cpn_core/get_data/etraffic.py @@ -0,0 +1,176 @@ +from datetime import datetime +from logging import getLogger +from typing import Final, Literal, LiteralString, TypedDict, cast, override + +from curl_cffi import CurlError, requests + +from cpn_core.models.plate_info import PlateInfo +from cpn_core.models.violation_detail import ViolationDetail +from cpn_core.types.api import ApiEnum +from cpn_core.types.vehicle_type import get_vehicle_enum + +from .base import BaseGetDataEngine + +logger = getLogger(__name__) +API_TOKEN_URL = "https://etraffic.gtelict.vn/api/citizen/v2/auth/login" +API_URL = "https://etraffic.gtelict.vn/api/citizen/v2/property/deferred/fines" + +RESPONSE_DATETIME_FORMAT: LiteralString = "%H:%M, %d/%m/%Y" + + +class _DataPlateInfoResponse(TypedDict): + violationId: str | None + licensePlate: str + licensePlateType: str + vehicleType: Literal["Ô tô con", "Xe máy", "Xe máy điện"] + vehicleTypeText: Literal["Ô tô con", "Xe máy", "Xe máy điện"] + violationType: str | None + violationTypeText: str + violationAt: str + violationAtText: str + violationAddress: str + handlingAddress: str + propertyName: str + statusType: Literal["Đã xử phạt", "Chưa xử phạt"] + statusTypeText: Literal["Đã xử phạt", "Chưa xử phạt"] + departmentName: str + contactPhone: str + + +class _FoundResponse(TypedDict): + tag: Literal["found_response"] + status: int + message: str + data: list[_DataPlateInfoResponse] + + +class _LimitResponse(TypedDict): + tag: Literal["limit_response"] + guid: str + code: str + message: Literal[ + "Số lượt tìm kiếm thông tin phạt nguội đã đạt giới hạn trong ngày.\nVui lòng thử lại sau" + ] + status: int + path: str + method: str + timestamp: str + error: str | None + + +_Response = _LimitResponse | _FoundResponse + + +class _EtrafficGetDataParseEngine: + def __init__( + self, plate_info: PlateInfo, data: tuple[_DataPlateInfoResponse, ...] + ) -> None: + self._plate_info = plate_info + self._data = data + self._violations_details_set = set() + + def _parse_violation(self, data: _DataPlateInfoResponse) -> None: + plate: str = data["licensePlate"] + date: str = data["violationAt"] + type: Literal["Ô tô con", "Xe máy", "Xe máy điện"] = data["vehicleType"] + color: str = data["licensePlateType"] + location: str = data["handlingAddress"] + status: str = data["statusType"] + enforcement_unit: str = data["propertyName"] + resolution_offices: tuple[str, ...] = tuple(data["departmentName"]) + violation_detail: ViolationDetail = ViolationDetail( + plate=plate, + color=color, + type=get_vehicle_enum(type), + date=datetime.strptime(str(date), RESPONSE_DATETIME_FORMAT), + location=location, + status=status == "Đã xử phạt", + enforcement_unit=enforcement_unit, + resolution_offices=resolution_offices, + violation=None, + ) + self._violations_details_set.add(violation_detail) + + def parse(self) -> tuple[ViolationDetail, ...] | None: + for violations in self._data: + self._parse_violation(violations) + return tuple(self._violations_details_set) + + +class EtrafficGetDataEngine(BaseGetDataEngine): + @property + def api(self): + """The api property.""" + return ApiEnum.etraffic_gtelict_vn + + headers = { + "Content-Type": "application/json", + "User-Agent": "C08_CD/1.1.8 (com.ots.global.vneTrafic; build:32; iOS 18.2.1) Alamofire/5.10.2", + } + + def __init__( + self, citizen_indentify: str, password: str, time_out: float = 10 + ) -> None: + self._citizen_indetify = citizen_indentify + self._password = password + self._time_out = time_out + + def _request_token(self, plate_info: PlateInfo) -> str | None: + data: Final[dict[str, str]] = { + "citizenIndentify": self._citizen_indetify, + "password": self._password, + } + try: + response = requests.post( + url=API_TOKEN_URL, + headers=self.headers, + json=data, + allow_redirects=False, + verify=False, + ) + data_dict = response.json() + return data_dict["value"]["refreshToken"] + except CurlError as e: + logger.error( + f"Error occurs while getting token for plate {plate_info.plate} in API {API_TOKEN_URL}: {e}" + ) + except Exception as e: + logger.error(f"Error occurs:{e}") + + def _request(self, plate_info: PlateInfo) -> _Response | None: + headers: Final[dict[str, str]] = { + "Authorization": f"Bearer {self._request_token(plate_info)}", + "User-Agent": "C08_CD/1.1.8 (com.ots.global.vneTrafic; build:32; iOS 18.2.1) Alamofire/5.10.2", + } + params: Final[dict[str, str]] = { + "licensePlate": plate_info.plate, + "type": f"{get_vehicle_enum(plate_info.type)}", + } + try: + response = requests.get(url=API_URL, headers=headers, params=params) + plate_detail_raw = response.json() + return cast(_Response, plate_detail_raw) + except CurlError as e: + logger.error( + f"Error occurs while getting data for plate {plate_info.plate} in API {API_TOKEN_URL}: {e}" + ) + except Exception as e: + logger.error(f"Error occurs:{e}") + + @override + async def _get_data( + self, plate_info: PlateInfo + ) -> tuple[ViolationDetail, ...] | None: + plate_detail_typed = self._request(plate_info) + if not plate_detail_typed: + logger.error(f"Failed to get data from api:{self.api}") + return + if plate_detail_typed["tag"] == "limit_response": + logger.error("You are limited to send more requests") + return + violation_details: tuple[ViolationDetail, ...] | None = ( + _EtrafficGetDataParseEngine( + plate_info=plate_info, data=tuple(plate_detail_typed["data"]) + ).parse() + ) + return violation_details diff --git a/src/cpn_core/types/api.py b/src/cpn_core/types/api.py index 8a9e8de..0f4f79d 100644 --- a/src/cpn_core/types/api.py +++ b/src/cpn_core/types/api.py @@ -6,6 +6,7 @@ class ApiEnum(str, Enum): csgt_vn = "csgt.vn" phatnguoi_vn = "phatnguoi.vn" zm_io_vn = "zm.io.vn" + etraffic_gtelict_vn = "etraffic.gtelict.vn" __all__ = ["ApiEnum"]