IDM API

import typing

import requests


class IDMException(Exception):
    ...


class IDMAPIException(IDMException):

    def __init__(self, code, detail, **kw):
        self.code = code
        self.detail = detail
        self.kwargs = kw

    def __str__(self):
        return f"[{self.code}] {self.detail}"


class IDMAPICallbackException(IDMException):
    CALLBACK_CODES = {
        1: "Пустой запрос",
        2: "Неизвестный тип сигнала",
        3: "Пара пользователь/секрет не найдены",
        4: "Беседа не привязана",
        5: "Пользователь заблокирован",
        6: "Дежурный не активен",
        7: "Пользователь не активен",
        10: "Не удалось связать беседу"
    }

    def __init__(
            self,
            response: str,
            error_code: typing.Optional[int] = None,
            error_message: typing.Optional[str] = None,
            **kw
    ):
        self.error_type = response
        self.error_code = error_code
        self.error_message = error_message
        self.kwargs = kw

    def __str__(self):
        if self.error_type == 'error':
            if self.error_code in self.CALLBACK_CODES:
                return f"[{self.error_code}] {self.CALLBACK_CODES[self.error_code]}"
            elif self.error_message:
                return f"[{self.error_code}] {self.error_message}"
            else:
                return f"[{self.error_code}] Нe известная ошибка"
        else:
            if self.error_message:
                return f"VK [{self.error_code}] {self.error_message}"
            else:
                return f"VK [{self.error_code}] Нe известная ошибка"


class IDMAPI:
    """
    api = IDMAPI('https://idmduty.ru', 'IDMLP(id;secret)')

    try:
        cfg = api.get_lp_info('token')
        cfg['config']['secret_code']
    except IDMAPIException as ex:
        print(ex)

    try:
        api.send_my_signal(..., {"id": event['id'], "is_win_platform": True})
    except IDMAPICallbackException as ex:
        print(ex)
    """

    _session: typing.Optional[requests.Session]

    def __init__(self, base_domain: str, header: str):
        self.base_domain = base_domain
        self.header_info = header
        self._session = None

        self._get_lp_info_link = f"{self.base_domain}/api/dutys/get_lp_info/"
        self._save_lp_info_link = f"{self.base_domain}/api/dutys/save_lp_info/"
        self._callback_link = f"{self.base_domain}/callback/"

    @property
    def session(self) -> requests.Session:
        if not self._session:
            self._session = requests.Session()
        return self._session

    def get_lp_info(self, access_token: str) -> dict:
        response = self.session.post(
            self._get_lp_info_link,
            json={'access_token': access_token},
            headers={'User-Agent': self.header_info}
        ).json()
        if 'error' in response:
            raise IDMAPIException(**response['error'])
        return response['response']

    def send_request(self, url: str, data: dict) -> dict:
        response = self.session.post(url, json=data, headers={'User-Agent': self.header_info})
        response_json = response.json()
        if 'error' in response_json:
            raise IDMAPIException(**response_json['error'])
        if response_json['response'] in ('error', 'vk_error',):
            raise IDMAPICallbackException(**response_json)
        return response_json

    async def callback(self, data: dict) -> dict:
        return self.send_request(self._callback_link, data)

    async def ping(self):
        return await self.callback({
            "method": "ping",
            "message": {},
            "object": {}
        })

    def _send_signal(
            self,
            method: str,
            from_id: int,
            peer_id: int,
            conversation_message_id: int,
            date: int,
            text: str,
            vk_message: typing.Optional[dict] = None
    ):
        return self.callback({
            "method": method,
            "message": {
                "conversation_message_id": conversation_message_id,
                "from_id": from_id,
                "date": date,
                "text": text,
                "peer_id": peer_id
            },
            "object": {
                "chat": None,
                "from_id": from_id,
                "value": text,
                "conversation_message_id": conversation_message_id
            },
            "vkmessage": vk_message
        })

    def send_my_signal(
            self,
            from_id: int,
            peer_id: int,
            conversation_message_id: int,
            date: int,
            text: str,
            vk_message: typing.Optional[dict] = None
    ):
        return self._send_signal(
            'lpSendMySignal',
            from_id=from_id, peer_id=peer_id,
            conversation_message_id=conversation_message_id,
            date=date, text=text, vk_message=vk_message
        )

    def send_signal(
            self,
            from_id: int,
            peer_id: int,
            conversation_message_id: int,
            date: int,
            text: str,
            vk_message: typing.Optional[dict] = None
    ):
        return self._send_signal(
            'lpSendSignal',
            from_id=from_id, peer_id=peer_id,
            conversation_message_id=conversation_message_id,
            date=date, text=text, vk_message=vk_message
        )