Initial commit

.gitignore

@@ -0,0 +1,3 @@
+dist/
+__pycache__/
+venv/

README.md

@@ -0,0 +1,24 @@
+# Неофициальная синхронная обёртка для API сервиса lknpd.nalog.ru на Python.
+
+Служит для автоматизации отправки информации о доходах самозанятых и получения информации о созданных чеках.
+
+Inspired by [https://github.com/alexstep/moy-nalog](https://github.com/alexstep/moy-nalog)
+
+## Использование
+Установите пакет
+```bash
+pip install nalogapi
+```
+
+Инициализаци и авторизация
+```python
+from nalogapi import NalogAPI
+NalogAPI.configure("inn", "password") #password that used in lkfl
+```
+
+Отправка информации о доходе
+```python
+NalogAPI.addIncome(datetime.utcnow(), 1.0, "Предоставление информационных услуг #970/2495")
+```
+
+Функция возвращает ссылку на чек вида [https://lknpd.nalog.ru/api/v1/receipt/344111066022/200egltxe8/print](https://lknpd.nalog.ru/api/v1/receipt/344111066022/200egltxe8/print)

nalogapi.py

@@ -0,0 +1,201 @@
+import requests
+import json
+import random
+import string
+from datetime import datetime, timezone
+
+
+class ConfigurationError(Exception):
+    pass
+
+
+class AuthenticationError(Exception):
+    pass
+
+
+class NalogAPI():
+
+    apiUrl = 'https://lknpd.nalog.ru/api/v1'
+    username = None
+    password = None
+    autologin = False
+    inn = None
+    token = None
+    tokenExpireIn = None
+    refreshToken = None
+    sourceDeviceId = None
+
+    def __init__(self):
+        if self.username is None or self.password is None:
+            raise ConfigurationError("username and password are required")
+        if self.sourceDeviceId is None:
+            self.sourceDeviceId = self.createDeviceId()
+        if self.autologin:
+            self.auth(self.login, self.password)
+
+    @staticmethod
+    def configure(username, password, autologin = False):
+        NalogAPI.username = username
+        NalogAPI.password = password
+        NalogAPI.autologin = autologin
+
+    @staticmethod
+    def createDeviceId():
+        return ''.join(random.choice(string.digits) for i in range(21))
+
+    @staticmethod
+    def getUtcDateTime(timestr):
+        return datetime.strptime(timestr, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)
+
+    @staticmethod
+    def getTimeString(dt):
+        return dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+    def auth(self, login, password):
+        headers = {
+            'accept': 'application/json, text/plain, */*',
+            'accept-language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7',
+            'content-type': 'application/json',
+            'referrer': 'https://lknpd.nalog.ru/',
+            'referrerPolicy': 'strict-origin-when-cross-origin',
+        }
+        payload = {
+            'username': login,
+            'password': password,
+            'deviceInfo': {
+                'sourceDeviceId': self.sourceDeviceId,
+                'sourceType': 'WEB',
+                'appVersion': '1.0.0',
+                'metaDetails': {
+                    'userAgent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 11_2_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.192 Safari/537.36'
+                }
+            }
+        }
+        s = requests.Session()
+        retries = requests.adapters.Retry(total=3, backoff_factor=0.5, status_forcelist=[ 502, 503, 504 ])
+        s.mount('http://', requests.adapters.HTTPAdapter(max_retries=retries))
+        url = self.apiUrl + '/auth/lkfl'
+        try:
+            r = s.post(url, data=json.dumps(payload), headers=headers, timeout=5)
+        except requests.ConnectionError:
+            raise AuthenticationError("Can't connect to authentication server")
+        res = r.json()
+        if not res['refreshToken']:
+            raise AuthenticationError("Authentication failure")
+        self.inn = res['profile']['inn']
+        self.token = res['token']
+        self.tokenExpireIn = self.getUtcDateTime(res['tokenExpireIn'])
+        self.refreshToken = res['refreshToken']
+
+    def getToken(self):
+        if self.token and self.tokenExpireIn and self.tokenExpireIn > datetime.now().replace(tzinfo=timezone.utc):
+            return self.token
+
+        if self.refreshToken is None:
+            self.auth(self.username, self.password)
+            return self.token
+
+        headers = {
+            'accept': 'application/json, text/plain, */*',
+            'accept-language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7',
+            'content-type': 'application/json',
+            'referrer': 'https://lknpd.nalog.ru/sales',
+            'referrerPolicy': 'strict-origin-when-cross-origin',
+        }
+        payload = {
+            'deviceInfo': {
+                'sourceDeviceId': self.sourceDeviceId,
+                'sourceType': 'WEB',
+                'appVersion': '1.0.0',
+                'metaDetails': {
+                    'userAgent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 11_2_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.192 Safari/537.36'
+                }
+            },
+            'refreshToken': self.refreshToken
+        }
+        s = requests.Session()
+        retries = requests.adapters.Retry(total=3, backoff_factor=0.5, status_forcelist=[ 502, 503, 504 ])
+        s.mount('http://', requests.adapters.HTTPAdapter(max_retries=retries))
+        url = self.apiUrl + '/auth/token'
+        try:
+            r = s.post(url, data=json.dumps(payload), headers=headers, timeout=5)
+        except requests.ConnectionError:
+            raise Exception("Failed to fetch token")
+        res = r.json()
+        if res['refreshToken']:
+            self.refreshToken = res['refreshToken']
+        self.token = res['token']
+        self.tokenExpireIn = res['tokenExpireIn']
+        return self.token
+
+    def call(self, endpoint, payload=None):
+        post = True if payload is not None else False
+        headers = {
+            'authorization': str('Bearer ' + self.getToken()),
+            'accept': 'application/json, text/plain, */*',
+            'accept-language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7',
+            'content-type': 'application/json',
+            'referrer': 'https://lknpd.nalog.ru/sales/create',
+            'referrerPolicy': 'strict-origin-when-cross-origin',
+        }
+        s = requests.Session()
+        retries = requests.adapters.Retry(total=3, backoff_factor=0.5, status_forcelist=[ 502, 503, 504 ])
+        s.mount('http://', requests.adapters.HTTPAdapter(max_retries=retries))
+        url = self.apiUrl + '/' + endpoint
+        r = None
+        if post:
+            try:
+                r = s.post(url, data=json.dumps(payload), headers=headers, timeout=5)
+            except requests.ConnectionError:
+                raise Exception("Failed to call")
+        else:
+            try:
+                r = s.get(url, headers=headers, timeout=5)
+            except requests.ConnectionError:
+                raise Exception("Failed to call")
+        res = r.json()
+        return res
+
+    @classmethod
+    def addIncome(cls, date, amount, name):
+        self = cls()
+        payload = {
+            'paymentType': 'CASH',
+            'ignoreMaxTotalIncomeRestriction': False,
+            'client': {
+                'contactPhone': None,
+                'displayName': None,
+                'incomeType': 'FROM_INDIVIDUAL',
+                'inn': None
+            },
+            'requestTime': self.getTimeString(datetime.utcnow()),
+            'operationTime': self.getTimeString(date),
+            'services': [{
+            'name': name, # 'Предоставление информационных услуг #970/2495',
+            'amount': str(amount),
+            'quantity': 1
+            }],
+            'totalAmount': str(amount)
+        }
+        res = self.call('income', payload)
+        if not res or not 'approvedReceiptUuid' in res:
+            return {'error': res}
+        return "{}/receipt/{}/{}/print".format(self.apiUrl, self.inn, res['approvedReceiptUuid'])
+
+    @classmethod
+    def userInfo(cls):
+        self = cls()
+        print(self.call('user'))
+
+    @classmethod
+    def paymentsInfo(cls):
+        self = cls()
+        print(self.call('keys'))
+
+def main():
+    NalogAPI.configure("INN", "password") #Пароль от личного кабинета налогоплательщика lkfl
+    NalogAPI.userInfo()
+    # NalogAPI.addIncome(datetime.utcnow(), 1.0, "Предоставление информационных услуг #970/2495")
+
+if __name__ == '__main__':
+    main()

pyproject.toml

@@ -0,0 +1,16 @@
+[tool.poetry]
+name = "nalogapi"
+version = "0.1.0"
+description = "Unofficial synchronous wrapper for lknpd.nalog.ru API written in Python"
+authors = ["Mike Ovchinnikov <[email protected]>"]
+license = "MIT"
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.7"
+requests = "^2.28.1"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"