Pull to refresh

Получаем список товаров из чека ИФНС (Raspberry + FreePBX + telegram + sheets)

Level of difficultyMedium
Reading time11 min
Views9.3K

Для рабочих целей есть потребность учитывать совершённые за наличные расходы. Раньше это делалось вручную - вписывалось наименование покупки и её цена в телеграм-чатик, потом вручную переносилось в google sheets. Потом перенос в google sheets автоматизировался с помощью скрипта python и google-api, но т.к. товаров в чеке могло быть много, поэтому список из 10 (например) позиций сокращался до какой-то общей типа "инструменты" (например) с указанием общей суммы чека, что не особо годилось для возможной дальнейшей аналитики. Как следующий этап развития, возникла идея получать данные о товарах с помощью qr-кода и API ИФНС.

В наличии имеется постоянно подключенный к сети одноплатный компьютер Raspberry PI с установленным FreeBPX (настроено по этой статье) с модемом E1550, перепрошитым на голосовые функции. Соответственно, все настройки выполнялись на этом комплекте.

В моём случае Raspberry и модем уже настроены и работают несколько лет, могут звонить и получать СМС. Дальнейшая задача - получить список товаров из чека.

ИФНС позволяет это сделать двумя путями: 1) использовать API-ключ (и тогда теоретически можно обойтись без модема); 2) проходить авторизацию по СМС.

Т.к. у меня есть модем и возможность получать СМС, иду по второму пути.

Принцип авторизации следующий: отправляем запрос на сайт налоговой, в котором указываем номер телефона, в ответ на номер телефона приходит смс для авторизации, после чего отправляем запрос с данными чека и кодом из смс, в ответ на который получаем JSON с данными из чека, если чека находится в базе ИФНС, либо информацию с ошибкой, если чека нет.

Принцип работы моей схемы:

  1. Пользователь сканирует qr-код с чека и отправляет текст qr-кода в телеграм-чат;

  2. Телеграм-бот получает текст сообщения и запускает скрипт Python, который проверяет, является ли сообщение QR-кодом и, если да, то делает запрос в налоговую, проходя авторизацию по СМС, получает список товаров и возвращает его попозиционно в телеграм, а также записывает в google sheets;

  3. Если сообщение не является QR-кодом, то с помощью скрипта осуществляется запись текста сообщения в google sheets.

Для создания скрипта использовалась информация из этой статьи, а также репозиторий на гитхабе.

Исходный скрипт с гитхаба был немного доработан под мои нужды, а именно: добавлена функция проверки срока жизни кода, чтобы не запрашивать каждый раз новый код в случае необходимости проверить несколько чеков подряд. Полученный СМС-код действует 120 секунд.

Как определить жив ли код и остаток его жизни?

Для этого добавил в исходный скрипт функцию проверки жизнеспособности ранее полученного кода check_sms. Работает это так: делаем post-запрос, включающий номер телефона, в ответ получаем статус ранее отправленного (если такой был) кода:

  • 429 - код есть и действует, также увидим в ответе оставшееся время его действия в секундах (максимум 120);

  • 204 - код устарел, либо его не было.

Учитываем, что asterisk настроен на сохранение входящих смс в файл /var/log/asterisk/sms.txt. В принципе, можно настроить сохранение смс в базу asteriskcdrdb, а потом читать смс оттуда, но настроить сохранение в файл - дело 1 строчки в конфиге, а сохранение в базу чуть посложнее.

Т.к. я не программист, то много комментировал код, а также добавил кучу print для понимания в процессе доработки, что происходит и какой элемент за что отвечает, поэтому в целом должно быть всё понятно. Доработанный скрипт:

nalog-python.py
import json
import time
import requests
import os
from datetime import datetime

class NalogRuPython:
    HOST = 'irkkt-mobile.nalog.ru:8888'
    DEVICE_OS = 'iOS'
    CLIENT_VERSION = '2.9.0'
    DEVICE_ID = '7C82010F-16CC-446B-8F66-FC4080C66521'
    ACCEPT = '*/*'
    USER_AGENT = 'billchecker/2.9.0 (iPhone; iOS 13.6; Scale/2.00)'
    ACCEPT_LANGUAGE = 'ru-RU;q=1, en-US;q=0.9'
    CLIENT_SECRET = 'IyvrAbKt9h/8p6a7QPh8gpkXYQ4='
    OS = 'Android'

    def __init__(self):
        self.__session_id = None
        self.check_sms()
        self.set_session_id()

    def check_sms(self): #проверка срока действия кода из смс и запрос нового, если старый код из смс уже не подойдет
#        """
#        Authorization using phone and SMS code
#        """
        self.__phone = str('+71234567890') #номер телефона, на который будет приходить смс (asterisk freePBX)

        url = f'https://{self.HOST}/v2/auth/phone/request'
        payload = {
            'phone': self.__phone,
            'client_secret': self.CLIENT_SECRET,
            'os': self.OS
        }
        headers = {
            'Host': self.HOST,
            'Accept': self.ACCEPT,
            'Device-OS': self.DEVICE_OS,
            'Device-Id': self.DEVICE_ID,
            'clientVersion': self.CLIENT_VERSION,
            'Accept-Language': self.ACCEPT_LANGUAGE,
            'User-Agent': self.USER_AGENT,
        }

        r = requests.post(url, json=payload, headers=headers) #запрос с параметрами на проверку срока действия кода (варианты - есть и действует какое-то время (не более 120 сек)  или нет (шлет новый тогда))
        print (datetime.now())

################### СМС код запрошен ######################################################
#статус-код - 204 - код не действует, отправлен новый, 429 - код действует, можно использовать старый
        print ("status code: " + str(r.status_code))

#ответные headers - время, до какого действует код из смс, время в сек действия, время текущее
        print ('headers: ' + str(r.headers))

# время действия кода оставшееся
        time_left = int(r.headers ['Retry-After']) # сколько секунд ещё действует код из СМС
        print ("time_left: " + str(time_left))

        if r.status_code == 204: # Если кода давно не было (более 120 сек.), то считываем новое СМС
           print ("код устарел, получаю новый")
           time.sleep(10) # ждем доставку смс
        elif r.status_code == 429 and time_left <5: # если код есть, но через 5 сек он уже не будет действовать, то ждем 6 сек и получаем новый
           time.sleep(6)
           self.check_sms()
        elif r.status_code == 429 and time_left >=5: # если код есть и действует более 5 сек, то используем его

#читаем СМС из файла
        f_read = open("/var/log/asterisk/sms.txt", "r") #открываем для чтения
        last_line = f_read.readlines()[-1] # считываем последнюю строку
        print (last_line)
        #пример строки: "2023-09-15 21:09:02 - dongle0 - KKT.NALOG: Проверка чеков ФНС России. Код: 2872"
        f_read.close() #закрываем файл

        sms_lst = last_line.split(' - ')[2] # делим последнюю строку на элементы списка, используя разделитель в виде " - " (пробел минус пробел)
        print ("sms last: " + sms_lst)

        self.__code = sms_lst.strip().split(': ')[2] # берем последний элемент (третий) из этого списка, удаляем из него перенос строки и пробелы, чтобы получить только цифры
        print ("code: -" + self.__code + "-") # минусы поставил, чтобы понять, что перед цифрами и после нет пробелов пример: -0123-
#################### пошло дальше уже с кодом ######################################



    def set_session_id(self) -> None:
        print ('session_id')
        url = f'https://{self.HOST}/v2/auth/phone/verify' #запрос уже с кодом
        payload = {
        'phone': self.__phone,
        'client_secret': self.CLIENT_SECRET,
        'code': self.__code,
        "os": self.OS
        }

        headers = {
            'Host': self.HOST,
            'Accept': self.ACCEPT,
            'Device-OS': self.DEVICE_OS,
            'Device-Id': self.DEVICE_ID,
            'clientVersion': self.CLIENT_VERSION,
            'Accept-Language': self.ACCEPT_LANGUAGE,
            'User-Agent': self.USER_AGENT,
        }


        resp = requests.post(url, json=payload, headers=headers)

        print ('headers: ')
        print (resp.headers)
        print ('status: ')
        print (resp.status_code)

        self.__session_id = resp.json()['sessionId']
        self.__refresh_token = resp.json()['refresh_token']

    def refresh_token_function(self) -> None:
        print ('refresh_token_function')
        url = f'https://{self.HOST}/v2/mobile/users/refresh'
        payload = {
            'refresh_token': self.__refresh_token,
            'client_secret': self.CLIENT_SECRET
        }

        headers = {
            'Host': self.HOST,
            'Accept': self.ACCEPT,
            'Device-OS': self.DEVICE_OS,
            'Device-Id': self.DEVICE_ID,
            'clientVersion': self.CLIENT_VERSION,
            'Accept-Language': self.ACCEPT_LANGUAGE,
            'User-Agent': self.USER_AGENT,
        }

        resp = requests.post(url, json=payload, headers=headers)

        self.__session_id = resp.json()['sessionId']
        self.__refresh_token = resp.json()['refresh_token']

    def _get_ticket_id(self, qr: str) -> str:
#        """
#        Get ticker id by info from qr code
#        :param qr: text from qr code. Example "t=20200727T174700&s=746.00&fn=9285000100206366&i=34929&fp=3951774668&n=1"
#        :return: Ticket id. Example "5f3bc6b953d5cb4f4e43a06c"
#        """
        url = f'https://{self.HOST}/v2/ticket'
        payload = {'qr': qr}
        headers = {
            'Host': self.HOST,
            'Accept': self.ACCEPT,
            'Device-OS': self.DEVICE_OS,
            'Device-Id': self.DEVICE_ID,
            'clientVersion': self.CLIENT_VERSION,
            'Accept-Language': self.ACCEPT_LANGUAGE,
            'sessionId': self.__session_id,
            'User-Agent': self.USER_AGENT,
        }

        resp = requests.post(url, json=payload, headers=headers)

        return resp.json()["id"]

    def get_ticket(self, qr: str) -> dict:
#        """
#        Get JSON ticket
#        :param qr: text from qr code. Example "t=20200727T174700&s=746.00&fn=9285000100206366&i=34929&fp=3951774668&n=1"
#        :return: JSON ticket
#        """
        ticket_id = self._get_ticket_id(qr)
        url = f'https://{self.HOST}/v2/tickets/{ticket_id}'
        headers = {
            'Host': self.HOST,
            'sessionId': self.__session_id,
            'Device-OS': self.DEVICE_OS,
            'clientVersion': self.CLIENT_VERSION,
            'Device-Id': self.DEVICE_ID,
            'Accept': self.ACCEPT,
            'User-Agent': self.USER_AGENT,
            'Accept-Language': self.ACCEPT_LANGUAGE,
            'Content-Type': 'application/json'
        }

        resp = requests.get(url, headers=headers)
        print (datetime.now())

        return resp.json()

Далее создаём телеграм-бот (процесс не описываю,- в сети достаточно мануалов), а также скрипт для обработки сообщений.

Для записи расходов есть телеграм-чат, куда записываются сообщения в формате "расход;сумма" (например: пирог;100), либо текст qr-кода с чека (например: t=20200924T1837&s=349.93&fn=9282440300682838&i=46534&fp=1273019065&n=1).

Принцип действия простой: если сообщение не qr-код, то просто записываем его в google sheets, а если qr-код, то запускаем процесс получения списка товаров, а в google sheets записываем информацию о товарах.

Далее описываю скрипт по частям:

Задаём начальные данные, указываем токен бота
#sheets
import gspread
from oauth2client.service_account import ServiceAccountCredentials

#телеграм бот
import telebot
import time
import datetime
import json
from nalog_python import NalogRuPython # делает запрос в налоговую, проходит авторизацию по смс, берет данные из смс и с ними получает данные чека

#обработчик фото чеков (берет фото чека и сканирует с него qr-код, если смог его найти, либо выдаёт error)
import photo5


# Создаем экземпляр бота
bot = telebot.TeleBot('токен бота')

Т.к. у меня происходит запись результатов в google sheets, то далее указываю ссылки на таблицу и на файл json с ключом для работы с google sheets (получил вот здесь).

Данные для интеграции с google sheets
#ссылка на таблицу, куда записывать
#В таблице колонки: Date, chatID, chat_username, chat_title, username, firstname, lastname, message, (reply src msg), fileName, fileUrl, e
sheet_url = 'https://docs.google.com/spreadsheets/d/sdfsdhbetretr4324dsarf' #рабочий

#что-то для работы с API google
scope = ["https://spreadsheets.google.com/feeds",'https://www.googleapis.com/auth/spreadsheets',"https://www.googleapis.com/auth/drive.file","https://www.googleapis.com/auth/drive"]

#путь к файлу с API (json) гугл console.cloud.google.com
credentials = ServiceAccountCredentials.from_json_keyfile_name('/home/qr/filename.json', scope)

# Авторизация в google
gc = gspread.authorize(credentials)

Далее функция для обработки сообщения от пользователя,— проверяем, является ли оно текстом qr-кода с чека, т.е. начинается ли с t=202

Почему "t=202"

Если отсканировать qr-код c чека, то результатом будет текст типа такого:

t=20200924T1837&s=349.93&fn=9282440300682838&i=46534&fp=1273019065&n=1

Этот текст состоит из нескольких частей, разделённых знаком "&", а именно:
дата и время (ггггммддТччмм): t=20200924T1837
сумма чека: s=349.93
номер фискального накопителя (ФН): fn=9282440300682838
номер фискального документа (ФД): i=46534
значение фискального признака (ФП): fp=1273019065
тип операции (приход/возврат прихода/расход/возврат расхода): n=1

Таким образом, все qr-коды чеков до 31.12.2029 года включительно, если ничего не поменяется, будут начинаться с t=202

Является ли сообщение QR-кодом?
# Получение сообщений от юзера
@bot.message_handler(content_types=["text"])
def handle_text(message):
    print (message.text) # текст сообщения
    qr_code = message.text # считаем, что нам прислали расшифрованный qr код с чека
    time1 = datetime.datetime.now().strftime('%d.%m.%Y') #дата и время в нужном формате для sheets для записи в таблицу
    sht2 = gc.open_by_url(sheet_url_my).get_worksheet(0) #открываем гугл-таблицу

    if qr_code.startswith("t=202"): #проверяем, так ли это (qr-код начинается с t=202 пример: t=20220110T1730&s=56998.00&fn=9960440301285687&i=93230&fp=3805313241&n=1
        print ('Это qr-код')

Если да, то делаем запрос в ИФНС
        client = NalogRuPython() # это уже для обращения к сайту ифнс
        ticket = client.get_ticket(qr_code) # делаем запрос в ифнс для получения данных из чека
        dictData = json.loads(json.dumps(ticket, indent=4, ensure_ascii=False)) # присваиваем переменной данные из чека

Дальше проверяем, есть ли чек в базе ИФНС (бывает не успел загрузиться или ещё что-нибудь) и если да, то парсим ответ и пишем в google sheets и телеграм. Нужно учитывать, что есть ограничение на частоту отправки сообщений в телеграм (опыт показал, что 5 секунд между сообщениями достаточно).

Проверяем чек, получаем и записываем список товаров
        if "ticket" in dictData: #проверяем, есть ли в ответе информация о товарах (раздел ticket), если нет, то чек не читается в базе ИФНС
            for each in dictData["ticket"]["document"]["receipt"]["items"]: # для каждого товара из чека выводим имя, количество, цену за единицу (делим её на 100, чтобы получить в руб.), сумму позиции (тоже делим)
                name = str(each['name']).replace(';',',') # имя товара, меняем точку с запятой в имени на запятую
                quantity = str(each['quantity']).replace('.',',') # количество с запятой вместо точки
                itemprice = str((float(each['price']))/100).replace('.',',') # цена в копейках
                itemsum = str((float(each['sum']))/100).replace('.',',') # сумма в копейках с запятой вместо точки
                msg1 = name + ";" + itemsum + ";" + str(quantity) + ";" + itemprice + ";" + dictData["organization"]["name"] # в нужный нам вид переводим
                report_line = [str(time1), message.chat.id, message.chat.username, message.chat.title, message.from_user.username, message.from_user.first_name, message.from_user.last_name, msg1, "", "", "", "", qr_code] #отправляем в sheets
                sht2.append_row(report_line, table_range='A1') # записываем строку в sheets
                bot.send_message(message.chat.id, name + " (колво: " + str(quantity) + ")" + ";" + itemsum + "(Цена за ед.: " + itemprice + ")") # шлем данные о позиции в чат (наименование количество сумма)
                time.sleep(5) # ограничение на частоту отправки сообщений в телеграм - не более 20 сообщений в минуту в один чат (или типа того)
                print ('goods were written')

Если информации о чеке нет в ИФНС, то пишем об этом в телеграм и записываем текст qr-кода чека в лист ожидания, а другой скрипт (в данной статье не привожу) дважды в день эти записи проверяет.

Если информации о чеке нет в ИФНС, то:
        else: # если инфы о товарах нет, то сообщаем об ошибке
                msg1 = "ошибка чека (скорее всего чек не загрузился в базу ИФНС), попробую позже и буду проверять сам 2 раза в день в течение месяца. Когда загрузится, запишу инфу в таблицу. Если не загрузится, то тоже запишу, но без списка товаров,- просто сумму (" + str((float(dictData["operation"]["sum"]))/100).replace('.',',') + ")" # в нужный нам вид переводим (инфа об ошибке + сумма чека)
                bot.send_message(message.chat.id, msg1) # шлем данные об ошибке в чат (сумма чека, инфа об ошибке)
                
                wrongList = str(time1) + ',' + qr_code + ',' + str(message.chat.id) + ',' + str(message.from_user.first_name) + ',' + str(message.from_user.last_name) #составляем список неправильных qr-кодов и данных о сообщении
                
                with open('/home/qr/wrongList.txt', 'a') as f: #записываем список неправильных чеков в файл
                    f.write(wrongList + '\n') #каждый чек на новой строке (дата - qr - chatid - user - user_last_name), значения через запятую
                print ('ticket error, written to the wronglist')
                 
                time.sleep(5) # ограничение на частоту отправки сообщений в телеграм - не более 20 сообщений в минуту в один чат (или типа того)

                

Также обрабатываем обычные сообщения (в таком случае записываем в google sheets в формате "товар;сумма", а также добавляем данные об отправителе, id чата, имени пользователя)

Если сообщение от пользователя не является qr-кодом, то записываем его в google sheets
    else: #если сообщение не qr-код и не ошибка, то записываем его в гугл таблицу
        report_line = [str(time1), message.chat.id, message.chat.username, message.chat.title, message.from_user.username, message.from_user.first_name, message.from_user.last_name, message.text, "", "", "", ""] #отправляем в sheets сообщение полностью, если оно не qr-код
        sht2.append_row(report_line, table_range='A1') # записываем сообщение в таблицу (из переменной выше)
        print ('message is not qrcode and was written to sheets')

Запускаем бота
# Запускаем бота
bot.polling(none_stop=True, interval=0)

В итоге автоматизировался рутинный процесс, а также появилась возможность анализировать стоимость и количество товаров. В моём режиме работы проверяется не более 10 чеков в день, позиций за день не больше 100. В таком режиме работает уже около двух лет, проблем не наблюдалось, кроме как в единичных случаях отсутствуют чеки в базе ИФНС, но это вероятно из-за того, что со стороны продавца чек не ушёл в базу.

В результате получаем запись в google sheets и в телеграм.
"Сырые" данные в sheets
"Сырые" данные в sheets
Данные в sheets после разбивки сообщения на колонки
Данные в sheets после разбивки сообщения на колонки
Данные в Телеграме
Данные в Телеграме

Tags:
Hubs:
Total votes 8: ↑8 and ↓0+8
Comments15

Articles