API НСПД. Проблемы с выводом данных

Question

вот код

import httpx
from datetime import datetime
import asyncio

async def fetch_data(url, client):
    """Функция для асинхронного получения данных по URL с повторными попытками"""
    response = await client.get(url, timeout=30)
    await asyncio.sleep(2)
    if response.status_code == 200:
        return response.json()

async def api_nspd_json(code):
    result_parts_zu = None
    objectsList = None
    result_string = None
    pomList = None
    mashList = None
    landLinks = None
    buildParts = None
    permissionType = None
    result = {}
    code = code.replace(':', '%3A')
    url = f'https://nspd.gov.ru/api/geoportal/v2/search/geoportal?thematicSearchId=1&query={code}'

    async with httpx.AsyncClient(verify=False) as client:
        response_data = await fetch_data(url, client)
        
        if not response_data or 'data' not in response_data or 'features' not in response_data['data']:
            result['message'] = 'Объект не найден в НСПД, возможно снят с учета или ввели не правильно кадастровый номер.'
            return result, result_parts_zu, objectsList, result_string, pomList, mashList, landLinks, buildParts, permissionType
        
        coordinates = 'Без координат'
        for feature in response_data['data']['features']:
            if feature['geometry']['type'] == 'Polygon':
                coordinates = 'С координатами'

        properties = response_data['data']['features'][0]['properties']['options']
        feature_id = response_data['data']['features'][0]['id']
        category_id = response_data['meta'][0]['categoryId']
        
        async def get_tab_values(url):
            tab_data = await fetch_data(url, client)
            print(tab_data)
            if not tab_data:
                return "Нет данных"
            title = tab_data.get("title", "Нет заголовка")
            values = tab_data.get("value", [])
            values = [value for value in values if value]
            return f"{title}: <code>{', '.join(values)}</code>" if values else "Нет данных"

        if category_id == 36368:  # ЗУ
            result_parts_zu = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landParts&categoryId={category_id}&geomId={feature_id}')
            objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&categoryId={category_id}&geomId={feature_id}')
        elif category_id == 36384:  # ОНС
            landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&categoryId=36384&geomId={feature_id}')
        elif category_id == 36383:  # Сооружения
            pomList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&categoryId=36383&geomId={feature_id}')
            mashList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&categoryId=36383&geomId={feature_id}')
            buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=permissionType&categoryId=36383&geomId={feature_id}')
        elif category_id == 36369:  # ОКС
            permissionType = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=permissionType&categoryId=36369&geomId={feature_id}')

        for key, value in properties.items():
            if key == 'registersId' and value == 36440:
                result_string = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=compositionLand&objdocId={feature_id}&registersId=36440')
            elif key == 'registersId' and value == 36441:
                landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&objdocId={feature_id}&registersId=36441')
                objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&objdocId={feature_id}&registersId=36441')
            elif key == 'registersId' and value == 36453:
                landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&objdocId={feature_id}&registersId=36453')
                buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&objdocId={feature_id}&registersId=36453')

            if isinstance(value, str):
                if len(value) == 10 and value[4] == '-' and value[7] == '-':
                    date_obj = datetime.strptime(value, '%Y-%m-%d')
                    formatted_date = date_obj.strftime('%d.%m.%Y')
                    result[key] = formatted_date
                elif len(value) == 20 and value[10] == 'T' and value[-1] == 'Z':
                    date_obj = datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ')
                    formatted_date = date_obj.strftime('%d.%m.%Y')
                    result[key] = formatted_date
                elif isinstance(value, list):
                    result[key] = ', '.join(value) if value else None
                elif value not in [None, ""]:
                    result[key] = value
        
        result['coordinates'] = coordinates
    return result, result_parts_zu, objectsList, result_string, pomList, mashList, landLinks, buildParts, permissionType



async def main():
    result = await api_nspd_json("52:51:0010002:1139")
if __name__ == '__main__':
    asyncio.run(main())

Кадастровый номер 52:51:0010002:1139 это Многоквартирный дом, в котором есть помещения. Основной API это

code = code.replace(':', '%3A')
    url = f'https://nspd.gov.ru/api/geoportal/v2/search/geoportal?thematicSearchId=1&query={code}'

По которому можно вывести другие данные из НСПД:

Которые в коде прописаны вот таким образом (сразу скажу, что это еще не все ссылки и в дальнейшем буду "урезать" код):

if category_id == 36368:  # ЗУ
            result_parts_zu = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landParts&categoryId={category_id}&geomId={feature_id}')
            objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&categoryId={category_id}&geomId={feature_id}')
        elif category_id == 36384:  # ОНС
            landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&categoryId=36384&geomId={feature_id}')
        elif category_id == 36383:  # Сооружения
            pomList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&categoryId=36383&geomId={feature_id}')
            mashList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&categoryId=36383&geomId={feature_id}')
            buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=permissionType&categoryId=36383&geomId={feature_id}')
        elif category_id == 36369:  # ОКС
            permissionType = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=permissionType&categoryId=36369&geomId={feature_id}')

        for key, value in properties.items():
            if key == 'registersId' and value == 36440:
                result_string = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=compositionLand&objdocId={feature_id}&registersId=36440')
            elif key == 'registersId' and value == 36441:
                landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&objdocId={feature_id}&registersId=36441')
                objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&objdocId={feature_id}&registersId=36441')
            elif key == 'registersId' and value == 36453:
                landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&objdocId={feature_id}&registersId=36453')
                buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&objdocId={feature_id}&registersId=36453')

По кадастровому номеру 52:51:0010002:1139 нахожу ключ (если он есть) 'registersId' равный 36441 в этом случае идет запрос по ссылкам

elif key == 'registersId' and value == 36441:
                landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&objdocId={feature_id}&registersId=36441')
                objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&objdocId={feature_id}&registersId=36441')

и по ссылке objectsList должен выдать информацию по помещениям, которые привязаны к зданию (в самом НСПД они есть), но выдает в коде

async def get_tab_values(url):
            tab_data = await fetch_data(url, client)
            print(tab_data)

почему то сначала None (при первом запросе выдает ошибку 505) а потом (как буд то при повторном подключении) и сами данные {'title': 'Список объектов', 'object': [{'title': 'Помещения (количество)', 'value': ['4']}, {'title': 'Машино-места (колличество)', 'value': ['']}, {'title': 'Помещения (список)', 'value': ['52:51:0010002:1374', '52:51:0010002:1375', '52:51:0010002:1376', '52:51:0010002:1377']}, {'title': 'Машино-места (список)', 'value': ['']}]}. и этот None все портит, обрабатывал его через явные ожидания, через повторные переподключения, но все равно выдает первым None и только потом выдает данные.

Как можно обработать None? помогите пожалуйста

в дальнейшем сделал бота с кнопками инлайн и с колбэками и при запросе не выводит кнопку с помещениями (из за этого None)

Answer 1

Благодаря ответу @gord1402 получился вот такой код, на всех типах объектов еще не пробовал, это уже отдельно проверю:

import httpx
from datetime import datetime
import asyncio

async def fetch_data(url, client):
    try:
        response = await client.get(url, timeout=60)
        if response.status_code == 200:
            return response.json()
    except Exception as e:
        print(f"def fetch_data {url}: {e}")
        pass

async def api_nspd_json(code):
    result_parts_zu = None
    objectsList = None
    result_string = None
    pomList = None
    mashList = None
    landLinks = None
    buildParts = None
    permissionType = None
    result = {}
    code = code.replace(':', '%3A')
    url = f'https://nspd.gov.ru/api/geoportal/v2/search/geoportal?thematicSearchId=1&query={code}'

    async with httpx.AsyncClient(verify=False) as client:
        response_data = await fetch_data(url, client)
        
        if not response_data or 'data' not in response_data or 'features' not in response_data['data']:
            result['message'] = 'Объект не найден в НСПД, возможно снят с учета или ввели не правильно кадастровый номер.'
            return result, result_parts_zu, objectsList, result_string, pomList, mashList, landLinks, buildParts, permissionType
        
        coordinates = 'Без координат'
        for feature in response_data['data']['features']:
            if feature['geometry']['type'] == 'Polygon':
                coordinates = 'С координатами'

        properties = response_data['data']['features'][0]['properties']['options']
        feature_id = response_data['data']['features'][0]['id']
        category_id = response_data['meta'][0]['categoryId']
        async def get_tab_values(url):
            tab_data = await fetch_data(url, client)
            if not tab_data:
                return "Нет данных"
            
            title = tab_data.get("title", "Нет заголовка")
            if 'object' in tab_data:
                objects = tab_data['object']
                if objects:
                    values_list = []
                    for obj in objects:
                        if obj["title"] in ["Помещения (список)", "Машино-места (список)"]:
                            values = obj.get("value", [])
                            values = [value for value in values if value]
                            if values:
                                values_list.append(';'.join(values))
                    all_values_string = ', '.join(values_list)
                    return f"{title}: <code>{all_values_string}</code>" if all_values_string else "Нет данных"
            elif 'value' in tab_data:
                values = tab_data.get("value", [])
                values = [value for value in values if value]
                if values:
                    return f"{title}: <code>{';'.join(values)}</code>"
            return "Нет данных"

        if category_id == 36368:  # ЗУ
            result_parts_zu = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landParts&categoryId={category_id}&geomId={feature_id}')
            objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&categoryId={category_id}&geomId={feature_id}')
        elif category_id == 36384:  # ОНС
            landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&categoryId=36384&geomId={feature_id}')
        elif category_id == 36383:  # Сооружения
            pomList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&categoryId=36383&geomId={feature_id}')
            mashList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&categoryId=36383&geomId={feature_id}')
            buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=permissionType&categoryId=36383&geomId={feature_id}')
        elif category_id == 36369:  # ОКС
            permissionType = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=permissionType&categoryId=36369&geomId={feature_id}')

        for key, value in properties.items():
            if key == 'registersId' and value == 36440:
                result_string = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=compositionLand&objdocId={feature_id}&registersId=36440')
            elif key == 'registersId' and value == 36441:
                landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&objdocId={feature_id}&registersId=36441')
                objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&objdocId={feature_id}&registersId=36441')
            elif key == 'registersId' and value == 36453:
                landLinks = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=landLinks&objdocId={feature_id}&registersId=36453')
                buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&objdocId={feature_id}&registersId=36453')

            if isinstance(value, str):
                if len(value) == 10 and value[4] == '-' and value[7] == '-':
                    date_obj = datetime.strptime(value, '%Y-%m-%d')
                    formatted_date = date_obj.strftime('%d.%m.%Y')
                    result[key] = formatted_date
                elif len(value) == 20 and value[10] == 'T' and value[-1] == 'Z':
                    date_obj = datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ')
                    formatted_date = date_obj.strftime('%d.%m.%Y')
                    result[key] = formatted_date
                elif isinstance(value, list):
                    result[key] = ', '.join(value) if value else None
                elif value not in [None, ""]:
                    result[key] = value
        
        result['coordinates'] = coordinates
    return result, result_parts_zu, objectsList, result_string, pomList, mashList, landLinks, buildParts, permissionType

тогда выложу еще и по зонам (по территориальным, охранным и так далее), пусть будет, может кому понадобиться (но там вытаскиваю определенные данные (которые нужны мне):

value_dict = {(3, 4): 4,
            (6, 8, 9, 10, 11, 15): 5,
            (7,): 7}

async def check_numbers_in_dict(number, value_dict):
    for key in value_dict.keys():
        if number in key:
            return value_dict[key]
    return None

async def nspd_zone():
    mix = DatabaseMixin(db_path)
    try:
        result = await mix.select_kpt()
        for item in tqdm(result, desc='Зоны', total=len(result)):
            reg_numb = item[0]
            parts = reg_numb.split('-')
            if len(parts) > 1:
                second_part = parts[1].split('.')
                if second_part:
                    number_str = second_part[0]
                    if number_str.isdigit():
                        number = int(number_str)
                        value = await check_numbers_in_dict(number, value_dict)
                        if value is not None:
                            try:
                                reg_numb_r = reg_numb.replace(':', '%3A')
                                url = f'https://nspd.gov.ru/api/geoportal/v2/search/geoportal?query={reg_numb_r}&thematicSearchId={value}'
                                async with httpx.AsyncClient(verify=False) as client:
                                    response = await client.get(url, timeout=10)
                                    if response.status_code == 404:
                                        pass
                                    elif response.status_code == 200:
                                        data = response.json()
                                        properties = data['data']['features'][0]['properties']
                                        category_name = properties.get('categoryName', None)
                                        description = properties.get('options', {}).get('description', None)
                                        name_district = properties.get('options', {}).get('name_district', None)
                                        name_locality = properties.get('options', {}).get('name_locality', None)
                                        name_by_doc = properties.get('options', {}).get('name_by_doc', None)
                                        legal_act_document_issuer = properties.get('options', {}).get('legal_act_document_issuer', None)
                                        type_zone = properties.get('options', {}).get('type_zone', None)
                                        name = properties.get('options', {}).get('name', None)
                                        await mix.update_kpt(reg_numb, category_name, description, name_district, name_locality, name_by_doc, legal_act_document_issuer, type_zone, name)
                            except Exception as e:
                                pass
                        else:
                            print(f"Число {number} не найдено в строке '{reg_numb}'.")
    finally:
        await mix.close_connection()

asyncio.run(nspd_zone())

Тут часть кода, из БД result = await mix.select_kpt() берутся номера зон, для примера "52:51-3.103 52:51-4.159 52:00-15.9 52:51-7.29", которые разделяются на 4,3,15,7 и т.д., в дальнейшем проверяются по словарю value_dict и подставляется в ссылку &thematicSearchId={value}

БЛОГ НА HUSL

API НСПД. Проблемы с выводом данных

Ответы (1 шт):