这是代码
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}®istersId=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}®istersId=36441')
objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&objdocId={feature_id}®istersId=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}®istersId=36453')
buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&objdocId={feature_id}®istersId=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}®istersId=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}®istersId=36441')
objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&objdocId={feature_id}®istersId=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}®istersId=36453')
buildParts = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-values-data?tabClass=buildParts&objdocId={feature_id}®istersId=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}®istersId=36441')
objectsList = await get_tab_values(f'https://nspd.gov.ru/api/geoportal/v1/tab-group-data?tabClass=objectsList&objdocId={feature_id}®istersId=36441')
并通过引用objectsList应该提供与建筑物相关的处所的信息(它们位于NSPD本身中),但它在代码中给出了它
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': '停车位(列表)','值':['']}]}。这个 None 破坏了一切,我通过明确的等待、通过反复的重新连接来处理它,但它仍然先给出 None 然后再给出数据。
None 如何处理?请帮忙
后来我制作了一个带有内联按钮和回调的机器人,当请求时,它不会显示带有房间的按钮(因为这个 None )
感谢@gord1402 的回答,我得到了这个代码,我还没有在所有类型的物体上尝试过它,我会单独检查:
然后我也会按区域发布它(按领土,安全等),就这样,也许有人会需要它(但我会在那里提取某些数据(我需要的):
以下是部分代码,从 DB 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}