utils/check.business.ru_monitor/check.business.ru_monitor.py

#!/usr/bin/env python3

""" check.business.ru console monitor.
    WEB-UI:
        https://check.business.ru/monitor

    API doc:
        https://app.swaggerhub.com/apis/Business.Ru/check.business.ru/1.05
        https://app.swaggerhub.com/apis-docs/Business.Ru/check.business.ru/1.05

Returns:
    str: print today's status summary.
"""


import json
from argparse import ArgumentParser
from datetime import datetime
from hashlib import md5
from random import choice
from string import ascii_letters, digits
import requests


class CheckBusinessRu():
    """ API request handler.
    """

    def __init__(self, app_id: str, secret: str) -> None:
        self._app_id = app_id
        self._secret = secret
        self._apiurl = 'https://check.business.ru/open-api/v1'

    def _sign(self, params: dict) -> str:
        params_string = json.dumps(params).replace(' ', '')
        return md5((params_string + self._secret).encode('utf-8')).hexdigest()

    def _nonce(self, length: int = 16) -> str:
        return ''.join(choice(ascii_letters + digits) for i in range(length))

    def _call_state_system(self) -> dict:
        api_path = '/StateSystem'
        params = {
            "app_id": self._app_id,
            "nonce": self._nonce()
            }
        headers = {
            'Content-type': 'application/json; charset=utf-8',
            'accept' : 'application/json',
            'sign': self._sign(params)
            }
        response = requests.get(
            url=self._apiurl + api_path,
            params=params,
            headers=headers,
            timeout=15
            )
        return response.json()

    def _call_check(self) -> dict:
        api_path = '/Check'
        params = {
            "app_id": self._app_id,
            "filter_date_from": datetime.today().strftime('%Y-%m-%dT00:00:00'),
            "nonce": self._nonce()
            }
        headers = {
            'Content-type': 'application/json; charset=utf-8',
            'accept' : 'application/json',
            'sign': self._sign(params)
            }
        response = requests.get(
            url=self._apiurl + api_path,
            params=params,
            headers=headers,
            timeout=15
            )
        return response.json()

    def as_json(self) -> dict:
        """ Status summary as raw parser.

        Returns:
            dict: status summary as json.
        """
        response = {**self._call_state_system(), **self._call_check()}

        check_amount = 0
        check_progress = 0
        check_completed = 0
        check_completed_in_queue = 0
        check_completed_in_error = 0
        check_completed_success = 0
        if response['data'] is not None:
            check_amount = len(response['data'])
            for check in response['data']:
                if check['status'] != 3:
                    check_progress += 1
                else:
                    check_completed += 1
                    if check['result'] == 0:
                        check_completed_in_queue += 1
                    if check['result'] == 1:
                        check_completed_success += 1
                    if check['result'] == 2:
                        check_completed_in_error += 1

        return {
            'date_last_connect_app': response['date_last_connect_app'],
            'date_last_connect_ofd': response['date_last_connect_ofd'],
            'printer_status': response['printer_status'],
            'check_amount': check_amount,
            'check_progress': check_progress,
            'check_completed': check_completed,
            'check_completed_in_queue': check_completed_in_queue,
            'check_completed_in_error': check_completed_in_error,
            'check_completed_success': check_completed_success
            }

    def as_text(self) -> str:
        """ Status summary as json parser.

        Returns:
            str: status summary as text.
        """
        response = self.as_json()

        result = [
            'Последнее соединение с Приложение: ' + response['date_last_connect_app'],
            'Последнее соединение с ОФД-Сервис: ' + response['date_last_connect_ofd'],
            ]
        if response['printer_status'] == 0:
            result.append('Статус ККТ: ' + 'Не найден')
        if response['printer_status'] == 1:
            result.append('Статус ККТ: ' + 'Готов')
        if response['printer_status'] == 2:
            result.append('Статус ККТ: ' + 'Нет чековой ленты или смена превысила 24 часа')
        result.append('Чеки за сегодня: ' + str(response['check_amount']))
        result.append('Обрабатывается: ' + str(response['check_progress']))
        result.append('Очередь на кассу: ' + str(response['check_completed_in_queue']))
        result.append('Передано с ошибкой: ' + str(response['check_completed_in_error']))
        result.append('Пробито чеков: ' + str(response['check_completed_success']))

        return '\n'.join(result)


if __name__ == "__main__":
    args = ArgumentParser(
                    prog="check-business-ru_monitor",
                    description="print today's status summary.",
                    epilog="Dependencies: "
                        "- Python 3 (tested version 3.11.2), "
                        "- Python 3 modules: requests "
                    )
    args.add_argument('--app_id', type=str, required=True,
                    help='https://check.business.ru/settings/integrations')
    args.add_argument('--secret', type=str, required=True,
                    help='https://check.business.ru/settings/integrations')
    args.add_argument('--json', action='store_true', required=False,
                    help='print in json fromat')
    args = vars(args.parse_args())

    state = CheckBusinessRu(app_id=args['app_id'], secret=args['secret'])
    if args['json']:
        print(json.dumps(state.as_json(), indent=2))
    else:
        print(state.as_text())