qiwi_module/qiwi_module.py

# Qiwi module v1.10
# 17/05/2021
# https://t.me/ssleg  © 2020 – 2021

import logging
from datetime import datetime, timedelta
from pathlib import Path

import requests

headers = {
    'accept': 'application/json',
    'content-type': 'application/json'
}

url = 'https://api.qiwi.com/partner/bill/v1/bills/'

error_flag = True
theme_code = None


# не подлежит прямому вызову.
# проверяет, что модуль инициализирован.
def init_check():
    if error_flag:
        levent = 'module is not initialized.'
        logging.error(levent)
        return False
    return True


# не подлежит прямому вызову.
# проверяет корректность суммы счета.
def zero_check(amount):
    if type(amount) == int or type(amount) == float:
        if amount <= 0:
            levent = 'bill amount is zero or negative.'
            logging.error(levent)
            return False
        return True
    else:
        levent = 'bill amount is not a number. positive int or float values possible.'
        logging.error(levent)
        return False


# не подлежит прямому вызову.
# возврашает строку со временем действия счета в формате сервера Qiwi.
def get_valid_to_time(hours, mins):
    if type(hours) == int and type(mins) == int:
        if hours >= 0 and mins > 0:
            now = datetime.now()
            delta = timedelta(hours=hours, minutes=mins)
            valid_to = now + delta
            valid_to_string = str(valid_to)[0:19]
            rezult_string = valid_to_string.replace(' ', 'T') + '+03:00'
            return rezult_string
        else:
            levent = 'hours or minutes must be above zero.'
            logging.error(levent)
            return False
    else:
        levent = 'hours or minutes is not integer.'
        logging.error(levent)
        return False


# создание счета на оплату. при успехе возвращает url с формой оплаты для клиента.
# при неуспехе - False.
# примеры использования смотрите в sample.py и adv_sample.py
def create_bill(bill_amount, bill_id, comment_string=None, valid_hours=0, valid_mins=15):
    if not init_check():
        return False

    if not zero_check(bill_amount):
        return False

    if type(bill_amount) == float:
        full_penny = bill_amount * 100
        fract = full_penny - int(full_penny)
        if fract != 0:
            levent = 'bill amount must have integer number of penny (kopeek).'
            logging.error(levent)
            return False

    if type(bill_id) != str or bill_id == '':
        levent = 'bill id is not a string or empty.'
        logging.error(levent)
        return False

    valid_to = get_valid_to_time(valid_hours, valid_mins)
    if not valid_to:
        return False

    amount = {'currency': 'RUB', 'value': '{:.2f}'.format(bill_amount)}

    request_data = {'amount': amount, 'expirationDateTime': valid_to}

    if comment_string is not None:
        if type(comment_string) == str and comment_string != '':
            request_data['comment'] = comment_string
        else:
            levent = 'comment is not a string or empty.'
            logging.warning(levent)

    if theme_code is not None:
        custom = {'themeCode': theme_code}
        request_data['customFields'] = custom

    request_url = url + bill_id

    try:
        response = requests.put(request_url, json=request_data, headers=headers, timeout=5)
        response_code = response.status_code
        if response_code == 200:
            response_dict = response.json()
            return response_dict.get('payUrl')
        elif response_code == 401:
            levent = 'Qiwi autorization error. invalid secret key.'
            logging.error(levent)
            return False
        else:
            response_text = response.text
            levent = 'Qiwi server error (create bill). code - ' + str(response_code) + ', response - ' + response_text
            logging.error(levent)
            return False

    except Exception as e:
        levent = 'protocol error (create bill): ' + str(e)
        logging.error(levent)
        return False


# проверка статуса счета,на входе его идентификатор (текст), на выходе один из 4х статусов, если успешно:
# 'WAITING' - cчет выставлен, ожидает оплаты.
# 'PAID' - cчет оплачен.
# 'REJECTED' -	счет отменен.
# 'EXPIRED' - счет не оплачен и истек срок его действия.
# можно вызывать 1 раз в секунду и реже.
# если неуспешно - возвращает False.
def bill_status(bill_id):
    if not init_check():
        return False

    if type(bill_id) != str or bill_id == '':
        levent = 'bill id is not a string or empty.'
        logging.error(levent)
        return False

    request_url = url + bill_id

    try:
        response = requests.get(request_url, headers=headers, timeout=5)
        response_code = response.status_code
        if response_code == 200:
            response_dict = response.json()
            status = response_dict.get('status')
            return status.get('value')
        elif response_code == 401:
            levent = 'Qiwi autorization error. invalid secret key.'
            logging.error(levent)
            return False
        else:
            response_text = response.text
            levent = 'Qiwi server error (bill status). code - ' + str(response_code) + ', response - ' + response_text
            logging.error(levent)
            return False

    except Exception as e:
        levent = 'protocol error (bill status): ' + str(e)
        logging.error(levent)
        return False


# отмена счета, на входе его идентификатор (текст).
# в случае успеха возвращает REJECTED, иначе False
def cancel_bill(bill_id):
    if not init_check():
        return False

    if type(bill_id) != str or bill_id == '':
        levent = 'bill id is not a string or empty.'
        logging.error(levent)
        return False

    request_url = url + bill_id + '/reject'

    try:
        response = requests.post(request_url, headers=headers, timeout=5)
        response_code = response.status_code
        if response_code == 200:
            response_dict = response.json()
            status = response_dict.get('status')
            return status.get('value')
        elif response_code == 401:
            levent = 'Qiwi autorization error. invalid secret key.'
            logging.error(levent)
            return False
        else:
            response_text = response.text
            levent = 'Qiwi server error (cancel bill). code - ' + str(response_code) + ', response - ' + response_text
            logging.error(levent)
            return False

    except Exception as e:
        levent = 'protocol error (cancel bill): ' + str(e)
        logging.error(levent)
        return False


# инициализация модуля - загрузка секретного ключа из файла и настройка пользовательской темы для формы оплаты.
def init(theme=None):
    global error_flag
    global theme_code

    if theme is not None:
        if type(theme) == str and theme != '':
            theme_code = theme
        else:
            levent = 'custom theme code is not a string or empty. theme not used.'
            logging.warning(levent)

    key_path = Path('qiwi_key.txt')
    if not key_path.exists():
        levent = 'Qiwi key file not found, module work is not possible. write secret key to file qiwi_key.txt'
        logging.error(levent)
        return False

    file = open('qiwi_key.txt')
    qiwi_key = file.readline()
    file.close()

    if qiwi_key.find('\n') > -1:
        qiwi_key = qiwi_key[0:len(qiwi_key) - 1]

    if not 180 < len(qiwi_key) < 230:
        levent = 'Qiwi key not found in file, module work is not possible. write secret key to file qiwi_key.txt'
        logging.error(levent)
        return False

    headers['Authorization'] = 'Bearer ' + qiwi_key

    levent = 'Qiwi key loaded, init completed.'
    logging.info(levent)
    error_flag = False
    return True