Как стать автором
Обновить

API и ФССП: как это работает?

Python *Программирование *API *

Часто в рамках аудиторских проверок возникает задача получения данных об исполнительном производстве (ИП) в отношении физических лиц с сайта ФССП. В принципе на сайте ФССП возможность получения такой информации есть – мы можем выгрузить сведения об ИП по нескольким реквизитам. А именно, по номеру исполнительного производства, по номеру исполнительного документа, по реквизитам физического лица и, в случае, если это юридическое лицо – по наименованию и адресу организации. Интерфейс поисковой формы интуитивно понятен, подробно ознакомиться с возможностями электронного банка данных ФССП можно в этой Памятке.

Поиск сведений об ИП через сайт ФССП удобен в случае разовых обращений к базе данных ФССП, но, если надо получить информацию о нескольких тысячах дел, то ручной ввод реквизитов для поиска и система подтверждения запросов («капча»), делают выполнение этой задачи крайне утомительным и протяженным во времени.

Столкнувшись с подобной задачей, я начал поиск информации о возможностях автоматизации получения данных с сайта ФССП. Мне удалось выяснить, что для получения автоматизированного доступа к открытой части банка данных исполнительных производств, ФССП России предоставляет интерфейс программирования приложений (API). И в этой статье я расскажу о моем опыте использования API сайта ФССП.

Возможности API сайта ФССП

ФССП официально предоставляет программный интерфейс для доступа к банку данных. Для получения доступа к этому интерфейсу необходимо пройти процедуру регистрации и получить ключ доступа (токен), который представляет собой численно-буквенное выражение длиною 12 символов. Срок действия ключа — 1 год.

API сайта ФССП позволяет выполнить три вида запросов:

  1. Запрос информации об ИП в отношении ФЛ по реквизитам

  2. Запрос информации об ИП в отношении ЮЛ по реквизитам

  3. Запрос информации об ИП по его номеру

Возможности выгрузки сведений по номеру исполнительного документа в настоящий момент отсутствует. Но в то же время API предоставляет возможность выполнения групповых запросов. Это позволило мне написать скрипт для автоматизации процесса получения сведений об исполнительных производствах по их номерам. Поскольку я начинающий программист, в качестве языка программирования мною был выбран Python, как наиболее простой и интуитивно понятный в освоении. Далее в статье будут рассмотрены наиболее важные, на мой взгляд, детали, которые необходимо знать для написания скрипта автоматизированной выгрузки информации об ИП из банка данных ФССП.

Основные принципы работы с API сайта ФССП с использованием Python

Интерфейс программирования приложений (API), сайта ФССП принимает HTTP запросы и отдает ответ в виде JSON объекта. Для работы с HTTP запросами используется библиотека requests. Запросы к API требуют авторизации пользователя, поэтому ключ доступа передается в виде параметра запроса. Запросы могут быть одиночные или групповые. Фрагменты кода на языке Python, формирующие три основных типа запросов, представлены ниже:

#запрос сведений об ИП в отношении юридического лица

data = {
  'region': '66',
  'lastname': '\u0418\u0432\u0430\u043D\u043E\u0432',
  'firstname': '\u0418\u0432\u0430\u043D',
  'token': 'yourapikey'
}

response = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/search/physical', data=data)

#запрос сведений об ИП в отношении юридического лица
data = {
  'region': '66',
  'name': '\u0420\u043E\u043C\u0430\u0448\u043A\u0430',
  'token': 'yourapikey'
}
response = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/search/legal', data=data)

#запрос сведений об ИП (поиск по номеру)

data = {
  'number': '6860/12/61/66',
  'token': 'yourapikey'
}

response = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/search/ip', data=data)

Как было сказано ранее, API сайта ФССП предоставляет возможность формирования группового запроса. Групповой запрос может содержать одиночные запросы всех трех типов (по реквизитам ФЛ, по реквизитам ЮЛ, по номеру ИП), но их количество не должно превышать 50 штук. Ограничение по количеству групповых запросов в сутки — не более 5000. Также существуют ограничения по количеству одиночных запросов, с этой информацией вы можете ознакомиться на сайте ФССП. Ниже представлен фрагмент кода по формированию группового запроса:

#групповой запрос на поиск сведений в банке данных сайта ФССП

headers = {'accept': 'application/json','Content-Type': 'application/json'}

data = {"token": "yourapikey", "request": [{"type": 1, "params": {'firstname': '\u0418\u0432\u0430\u043D',\
'lastname': '\u0418\u0432\u0430\u043D\u043E\u0432','region': '66'}},{"type": 2,\
"params": {'name': '\u0420\u043E\u043C\u0430\u0448\u043A\u0430','region': '66'}},\
{"type": 3, "params": {"number": "6860/12/61/66"}}]}

response = requests.post('https://api-ip.fssp.gov.ru/api/v1.0/search/group', headers=headers, data=data)

Интерфейс программирования приложений (API) работает в асинхронном режиме, это значит, что в ответ на запрос пользователь получает идентификатор задачи, состояние которой необходимо проверять специальным видом запроса. Пример запроса статуса задачи выглядит следующим образом:

#запрос статуса выполнения задачи

data = {'task': 'ADAB49B2-F435-49CC-80DF-55F2F57D3057','token': 'yourapikey'}

response = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/status', data=data)

Существует четыре статуса задачи:

«0» — обработка завершена, с помощью метода /result можно получить результаты;

«1» — обработка начата, с помощью метода /result можно получить частичные результаты группового запроса;

«2» — обработка не начиналась, запрос находится в очереди;

«3» — обработка завершена, имели место ошибки, с помощью метода /result можно получить частичные результаты.

После получения статуса «0» мы можем получить результаты сделанного нами запроса:

#выгрузка результатов поданного запроса

data = {'task': 'BDAB49B2-F435-49CC-80DF-55F2F57D3057','token': 'yourapikey'}

response = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/result', data=data)

После этого остается только сохранить полученные результаты в удобном для работы формате.

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

Написание скрипта для работы с API сайта ФССП

Итак, передо мной стояла задача по написанию скрипта для выполнения групповых запросов к API сайта ФССП и для этого я использовал язык программирования Python 3.8.8 и набор стандартных библиотек:

#загружаем необходимые библиотеки
import pandas as pd
import json
import requests
import time
import traceback
import os
from tkinter import Tk
from tkinter.filedialog import askopenfilename
from tkinter.filedialog import askdirectory

!python --version
    Python 3.8.8
    
print('pandas ', pd.__version__)
print('json ', json.__version__)
print('requests ', requests.__version__)

    pandas  1.2.4
    json  2.0.9
    requests  2.25.1

В первую очередь мне надо было считать входные данные и обработать их партиями по 50 строк. Для удобства использования скрипта, я воспользовался GUI фреймворком Tkinter, при помощи которого был реализован выбор файла с входными данными и целевого каталога, куда впоследствии сохранялись результаты работы скрипта. Для работы с входными данными были использованы возможности библиотеки pandas. Я сформировал датафрейм из входного массива данных, после чего добавил поле с меткой, которое разбило весь датафрейм на части по 50 строк в каждой:

#выбор файла с входными данными
root = Tk()
filename = askopenfilename()
print(filename)

#выбор целевого каталога (туда мы сохраним результат)
target_path = askdirectory(title = 'Выбор каталога для выгрузки результата',initialdir = os.path.expanduser(u'C://'))
target_path = target_path.replace(r"/","\\")
root.destroy()

#загружаем входные данные
ip = pd.read_excel(filename)
# total_list = ip['ip'].to_list()
print('загружено ',len(ip),' строк')

#разметка входных данных по "батчам" - по 50 клиентов (ограничение API)
ip['batch'] = ip.index//50 + 1

В дальнейшем это позволило мне в цикле по метке брать фрагменты данных по 50 строки формировать JSON объекты с определенной структурой данных для каждого типа запросов.

#формирование входного JSON объекта
for i in list(ip['batch'].unique()):
    dic = {}
    rq = []
    js = ''
    print('Сессия - ',i,'\n')
    for j in ip[ip['batch'] == i].index:
        rq.append({'type':3,'params':{'number':ip['ip'][j]}})
    dic = {'token':token,'request':rq}
    js = json.dumps(dic)
#     print(rq)

После формирования JSON объекта, он передается в качестве параметра в групповом запросе к API сайта ФССП. Запрос не будет принят до тех пор, пока не завершится выполнение предыдущего запроса, это занимает некоторое время, поэтому я применил функцию time. sleep() для задержки исполнения кода и добавил проверку выполнения предыдущего запроса:

while js_task['exception'] == 'Дождитесь результата предыдущего группового запроса':
        time.sleep(20)
        response = requests.post('https://api-ip.fssp.gov.ru/api/v1.0/search/group', headers=headers, data=js)
        js_task = json.loads(response.content)

Так как API работает асинхронно, вслед за отправкой запроса и получения идентификатора задачи мне надо было отслеживать статус выполнения задачи, что было выполнено при помощи стандартного запроса:

#отправка запросов о статусе задачи и о выгрузке результата
    headers = {'accept': 'application/json'}
    params = (('token', token),('task', (js_task['response'])['task']))
    
##запрос статуса задачи - надо сделать отдельной функцией
    status = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/status', headers=headers, params=params)
    js_status = json.loads(status.content)
    print(js_status)
    time.sleep(3)

Как только API возвращает код успешного завершения задачи, необходимо получить результаты выполнения запроса и сохранить их:

##получение результатов запроса
    try:
        response1 = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/result', headers=headers, params=params)
        js_res = json.loads(response1.content)
    except:
        print(sys.exc_info())
    time.sleep(3)
    
    while (js_res['code'] != 2) and (js_res['response']['task_end'] is None):
        time.sleep(20)
        response1 = requests.get('https://api-ip.fssp.gov.ru/api/v1.0/result', headers=headers, params=params)
        js_res = json.loads(response1.content)

Код для сохранения результатов работы запроса вынесен в отдельную функцию и включает в себя обработку следующих событий:

  • Элемент входных данных не найден в банке данных сайта ФССП;

  • Элемент входных данных имеет неправильный формат

  • Элемент входных данных успешно найден и получены данные из банка данных сайта ФССП;

  • Элементу входных данных соответствует несколько уникальных записей в банке данных сайта ФССП.

#функция для сохранения результата запросов IP NUMBER
def results_save(js_res):
    number = []
    name = []
    exe_prod = []
    details = []
    subj = []
    depart = []
    bailiff = []
    ip_end =[]
    
    for i in js_res['response']['result']:
#обработка не найденных в банке данных ИП (вероятно фиктивные номера или введенные с ошибкой) надо объединить эти условия
        if i['result'] == [] or i['result'] == None:
                number.append((i['query']['params'])['number'])
                name.append('empty')
                exe_prod.append('empty')
                details.append('empty')
                subj.append('empty')
                depart.append('empty')
                bailiff.append('empty')
                ip_end.append('empty')
        
#обработка номеров ИП имеющих заведомо некорректный формат
        elif i['result'] == {'code': 0, 'message': 'The given data was invalid.', 'errors': {'number': ['Поле number имеет ошибочный формат.']}}:
            number.append((i['query']['params'])['number'])
            name.append('Поле number имеет ошибочный формат.')
            exe_prod.append('Поле number имеет ошибочный формат.')
            details.append('Поле number имеет ошибочный формат.')
            subj.append('Поле number имеет ошибочный формат.')
            depart.append('Поле number имеет ошибочный формат.')
            bailiff.append('Поле number имеет ошибочный формат.')
            ip_end.append('Поле number имеет ошибочный формат.')      
        else:
#обработка найденных ИП
            if len(i['result']) == 1:
                for j in (i['result']):
                    number.append((i['query']['params'])['number'])
                    name.append(str((j)['name']))
                    exe_prod.append((j)['exe_production'])
                    details.append((j)['details'])
                    subj.append((j)['subject'])
                    depart.append((j)['department'])
                    bailiff.append((j)['bailiff'])
                    ip_end.append((j)['ip_end'])
            else:
                #корректировка списка number для компенсации дублей
                qnt = 0
                while qnt < (len(i['result'])):
                    number.append((i['query']['params'])['number'])
                    qnt += 1
                
                for dbl in (i['result']):
                    name.append((dbl)['name'])
                    exe_prod.append((dbl)['exe_production'])
                    details.append((dbl)['details'])
                    subj.append((dbl)['subject'])
                    depart.append((dbl)['department'])
                    bailiff.append((dbl)['bailiff'])
                    ip_end.append((dbl)['ip_end'])
                                
#сборка результатов                 
    query_results = pd.DataFrame({'number' : number, 'name' : name,'exe_prod' : exe_prod,'details' : details,'subj' : subj,'depart' : bailiff,'ip_end' : ip_end})
    return query_results

Таким образом, функция обрабатывает результаты выполнения запроса и возвращает датафрейм, который в цикле добавляется к результирующему датафрейму. После полной отработки скрипта данные из результирующего датафрейма экспортируются в целевой каталог в виде файла в формате *.xlsx. При желании данные полученные после каждого цикла можно дописывать в файл в формате *.csv.

Заключение

В ходе работы со скриптом я дважды столкнулся с неожиданными изменениями в работе с API сайта ФССП: менялась форма запроса в части точки входа в API и возникала проблема с кодировкой входных данных. Решить эти проблемы мне частично помогла служба техподдержки, обратиться туда можно только по адресу admin@fssp.gov.ru. Ответ на запрос приходит в промежутке времени от 5 до 10 дней.

Несмотря на то, что написанный мною скрипт очень помог мне в ряде проверок, в нем есть множество недостатков, которые я планирую устранить. Код скрипта вы найдете по ссылке. Для желающих улучшить мой скрипт или написать свой собственный вариант рекомендую ознакомиться с описанием API ФССП и HTTP-запросов на сайте службы.

Ссылки:

  1. https://fssp.gov.ru/files/fssp/db/files/pamjatka/pamjatka_po_zadolzhennosti_202111_20211191318.pdf

  2. Описание API сайта ФССП https://api-ip.fssp.gov.ru/about

  3. Информация по использованию API сайта ФССП для разработчиков https://api-ip.fssp.gov.ru/swagger

  4. Приказ ФССП России от 12.05.2012 № 248 «Об утверждении Порядка создания и ведения банка данных в исполнительном производстве Федеральной службы судебных приставов» https://fssp.gov.ru/prikaz_fssp_248_12052012/

Теги:
Хабы:
Всего голосов 1: ↑1 и ↓0 +1
Просмотры 4.9K
Комментарии Комментарии 11

Работа

Python разработчик
152 вакансии
Data Scientist
107 вакансий