500 lines
27 KiB
Python
500 lines
27 KiB
Python
from lxml import etree
|
||
from difflib import SequenceMatcher
|
||
import openpyxl
|
||
import os
|
||
import datetime
|
||
|
||
class xlsx_filer():
|
||
|
||
def save_file(self):
|
||
if os.path.isfile(self.path):
|
||
try:
|
||
self.workbook.save(self.path)
|
||
except PermissionError:
|
||
return False
|
||
else:
|
||
current_datetime = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
||
file_name = f"/xml_{current_datetime}.xlsx"
|
||
self.workbook.save(self.path + file_name)
|
||
|
||
self.workbook.close()
|
||
|
||
def update_data(self):
|
||
self.sheet.cell(row=1, column=1).value = "№ п/п"
|
||
self.sheet.cell(row=1, column=2).value = "Раздел КПТР"
|
||
self.sheet.cell(row=1, column=3).value = "Кадастровый номер"
|
||
self.sheet.cell(row=1, column=4).value = "Кадастровый квартал"
|
||
self.sheet.cell(row=1, column=5).value = "Вид объекта"
|
||
self.sheet.cell(row=1, column=6).value = "Площадь ЗУ"
|
||
self.sheet.cell(row=1, column=7).value = "Погрешность определения площади"
|
||
self.sheet.cell(row=1, column=8).value = "Наличие координат"
|
||
self.sheet.cell(row=1, column=9).value = "Пояснения"
|
||
row = self.sheet.max_row+1
|
||
for object in self.data:
|
||
self.sheet.cell(row=row, column=1).value = row-1
|
||
self.sheet.cell(row=row, column=2).value = object.section_KPTR
|
||
self.sheet.cell(row=row, column=3).value = object.cadastral_number
|
||
self.sheet.cell(row=row, column=4).value = object.cadastral_quarter
|
||
self.sheet.cell(row=row, column=5).value = object.object_type
|
||
self.sheet.cell(row=row, column=6).value = object.land_plot_area
|
||
self.sheet.cell(row=row, column=7).value = object.error_in_determining_area
|
||
self.sheet.cell(row=row, column=8).value = object.availability_coordinates
|
||
self.sheet.cell(row=row, column=9).value = object.explanations
|
||
row += 1
|
||
|
||
def update_connections_data(self):
|
||
"""Обновляет данные на листе Связи"""
|
||
self.connections_sheet.cell(row=1, column=1).value = "№ п/п"
|
||
self.connections_sheet.cell(row=1, column=2).value = "Раздел КПТР"
|
||
self.connections_sheet.cell(row=1, column=3).value = "Кадастровый номер"
|
||
self.connections_sheet.cell(row=1, column=4).value = "Кадастровый квартал"
|
||
self.connections_sheet.cell(row=1, column=5).value = "Вид объекта"
|
||
self.connections_sheet.cell(row=1, column=6).value = "Связь"
|
||
|
||
row = 2
|
||
counter = 1
|
||
for connection in self.connections_data:
|
||
self.connections_sheet.cell(row=row, column=1).value = counter
|
||
self.connections_sheet.cell(row=row, column=2).value = connection.section_KPTR
|
||
self.connections_sheet.cell(row=row, column=3).value = connection.cadastral_number
|
||
self.connections_sheet.cell(row=row, column=4).value = connection.cadastral_quarter
|
||
self.connections_sheet.cell(row=row, column=5).value = connection.object_type
|
||
self.connections_sheet.cell(row=row, column=6).value = connection.connection
|
||
row += 1
|
||
counter += 1
|
||
|
||
def update_access_data(self):
|
||
"""Обновляет данные на листе Сведения о доступе"""
|
||
self.access_sheet.cell(row=1, column=1).value = "№ п/п"
|
||
self.access_sheet.cell(row=1, column=2).value = "Раздел КПТР"
|
||
self.access_sheet.cell(row=1, column=3).value = "Кадастровый номер"
|
||
self.access_sheet.cell(row=1, column=4).value = "Кадастровый квартал"
|
||
self.access_sheet.cell(row=1, column=5).value = "Сведения о доступе"
|
||
|
||
row = 2
|
||
counter = 1
|
||
for access in self.access_data:
|
||
self.access_sheet.cell(row=row, column=1).value = counter
|
||
self.access_sheet.cell(row=row, column=2).value = access.section_KPTR
|
||
self.access_sheet.cell(row=row, column=3).value = access.cadastral_number
|
||
self.access_sheet.cell(row=row, column=4).value = access.cadastral_quarter
|
||
self.access_sheet.cell(row=row, column=5).value = access.access_info
|
||
row += 1
|
||
counter += 1
|
||
|
||
def __init__(self,data=[], connections_data=[], access_data=[], path=""):
|
||
self.data = data
|
||
self.connections_data = connections_data
|
||
self.access_data = access_data
|
||
self.path = path
|
||
if os.path.isdir(path):
|
||
self.workbook = openpyxl.Workbook()
|
||
self.sheet = self.workbook.create_sheet("Основные данные")
|
||
self.connections_sheet = self.workbook.create_sheet("Связи")
|
||
self.access_sheet = self.workbook.create_sheet("Сведения о доступе")
|
||
# Удаляем стандартный лист
|
||
if "Sheet" in self.workbook.sheetnames:
|
||
self.workbook.remove(self.workbook["Sheet"])
|
||
|
||
|
||
class Data_farame():
|
||
def __init__(self,section_KPTR,cadastral_number,cadastral_quarter,object_type,land_plot_area,error_in_determining_area,availability_coordinates,explanations):
|
||
self.section_KPTR = section_KPTR #Раздел КПТР
|
||
self.cadastral_number = cadastral_number #Кадастровый номер
|
||
self.cadastral_quarter = cadastral_quarter #Кадастровый квартал
|
||
self.object_type = object_type #Вид объекта
|
||
self.land_plot_area = land_plot_area #Площадь ЗУ
|
||
self.error_in_determining_area = error_in_determining_area #Погрешность определения площади
|
||
self.availability_coordinates = availability_coordinates #наличие координат
|
||
self.explanations = explanations #Пояснения
|
||
|
||
|
||
class Connection_Data():
|
||
def __init__(self, section_KPTR, cadastral_number, cadastral_quarter, object_type, connection):
|
||
self.section_KPTR = section_KPTR
|
||
self.cadastral_number = cadastral_number
|
||
self.cadastral_quarter = cadastral_quarter
|
||
self.object_type = object_type
|
||
self.connection = connection
|
||
|
||
|
||
class Access_Data():
|
||
def __init__(self, section_KPTR, cadastral_number, cadastral_quarter, access_info):
|
||
self.section_KPTR = section_KPTR
|
||
self.cadastral_number = cadastral_number
|
||
self.cadastral_quarter = cadastral_quarter
|
||
self.access_info = access_info
|
||
|
||
|
||
class super_visor():
|
||
|
||
def find_best_match(self,input_str: str) -> str:
|
||
keywords = {
|
||
"FormParcels":"Образование ЗУ",
|
||
"SpecifyParcels":"Уточнение ЗУ",
|
||
"CadastralErrorsParcels":"Исправление ЗУ",
|
||
"ExistObjectsRealty":"Уточнение ОКС",
|
||
"ExistNewObjectsRealty":"Уточнение ОКС",
|
||
"CadastralErrorsOKS":"Исправление ОКС"
|
||
}
|
||
|
||
# Сравниваем коэффициент схожести (0.0 - 1.0) строки с каждым ключевым словом
|
||
best_match = max(keywords, key=lambda word: SequenceMatcher(None, input_str, word.lower()).ratio())
|
||
return keywords[best_match]
|
||
|
||
def chec_kad_number(self,atrebut):
|
||
if '@attributes' in atrebut:
|
||
if 'CadastralNumber' in atrebut['@attributes']:
|
||
return atrebut['@attributes']['CadastralNumber']
|
||
if 'Definition' in atrebut['@attributes']:
|
||
return atrebut['@attributes']['Definition']
|
||
if 'CadastralNumber' in atrebut:
|
||
if '#text' in atrebut['CadastralNumber'][0]:
|
||
return atrebut['CadastralNumber'][0]['#text']
|
||
return "нет"
|
||
|
||
def get_cad_block(self,cad_block):
|
||
temp = ""
|
||
for item in cad_block:
|
||
temp += f"{item['#text']}, "
|
||
return temp
|
||
|
||
def chec_cad_block(self, atrebut):
|
||
if 'CadastralBlocks' in atrebut:
|
||
if 'CadastralBlock' in atrebut['CadastralBlocks'][0]:
|
||
return self.get_cad_block(atrebut['CadastralBlocks'][0]['CadastralBlock'])
|
||
if 'CadastralBlock' in atrebut:
|
||
return atrebut['CadastralBlock'][0]['#text']
|
||
return "нет"
|
||
|
||
def get_object_type(self,type:str):
|
||
if type == "002001002000":
|
||
return "Здание"
|
||
if type == "002001004000":
|
||
return "Сооружение"
|
||
if type == "002001005000":
|
||
return "Объект"
|
||
|
||
def chec_object_type(self, atrebut):
|
||
if 'ObjectType' in atrebut:
|
||
return self.get_object_type(atrebut['ObjectType'][0]['#text'])
|
||
return "зу"
|
||
|
||
def chec_area(self,atrebut):
|
||
if "Area" in atrebut:
|
||
if "Area" in atrebut['Area'][0]:
|
||
if '#text' in atrebut['Area'][0]['Area'][0]:
|
||
return atrebut['Area'][0]['Area'][0]['#text']
|
||
return "нет"
|
||
|
||
def chec_inaccuracy(self, atrebut):
|
||
if "Area" in atrebut:
|
||
if "Inaccuracy" in atrebut['Area'][0]:
|
||
if '#text' in atrebut['Area'][0]['Inaccuracy'][0]:
|
||
return atrebut['Area'][0]['Inaccuracy'][0]['#text']
|
||
return "нет"
|
||
|
||
def chec_koordinat(self,atrebut):
|
||
if 'EntitySpatial' in atrebut:
|
||
if 'SpatialElement' in atrebut['EntitySpatial'][0]:
|
||
return "да"
|
||
else:
|
||
return "нет"
|
||
else:
|
||
return "нет"
|
||
|
||
def chec_explanations(self,atrebut):
|
||
if 'Explanations' in atrebut:
|
||
return atrebut['Explanations'][0]['#text']
|
||
if 'Explanation' in atrebut:
|
||
return atrebut['Explanation'][0]['#text']
|
||
return "нет"
|
||
|
||
def get_connections(self, atrebut):
|
||
"""Получает связи объекта"""
|
||
connections = []
|
||
|
||
# Для ЗУ ищем ObjectsRealty -> ObjectRealty -> InnerCadastralNumbers -> CadastralNumber
|
||
if 'ObjectsRealty' in atrebut:
|
||
objects_realty = atrebut['ObjectsRealty']
|
||
for obj_realty in objects_realty:
|
||
if 'ObjectRealty' in obj_realty:
|
||
for obj in obj_realty['ObjectRealty']:
|
||
if 'InnerCadastralNumbers' in obj:
|
||
for inner_cad in obj['InnerCadastralNumbers']:
|
||
if 'CadastralNumber' in inner_cad:
|
||
for cad_num in inner_cad['CadastralNumber']:
|
||
if '#text' in cad_num:
|
||
connections.append(cad_num['#text'])
|
||
|
||
# Для ОКС ищем ParentCadastralNumbers -> CadastralNumber
|
||
if 'ParentCadastralNumbers' in atrebut:
|
||
parent_cad = atrebut['ParentCadastralNumbers']
|
||
for parent in parent_cad:
|
||
if 'CadastralNumber' in parent:
|
||
for cad_num in parent['CadastralNumber']:
|
||
if '#text' in cad_num:
|
||
connections.append(cad_num['#text'])
|
||
|
||
return connections
|
||
|
||
def get_access_info(self, atrebut):
|
||
"""Получает сведения о доступе"""
|
||
access_info = []
|
||
|
||
if 'ProvidingPassCadastralNumbers' in atrebut:
|
||
providing_pass = atrebut['ProvidingPassCadastralNumbers']
|
||
for provide in providing_pass:
|
||
# CadastralNumber
|
||
if 'CadastralNumber' in provide:
|
||
for cad_num in provide['CadastralNumber']:
|
||
if '#text' in cad_num:
|
||
access_info.append(f"Кадастровый номер: {cad_num['#text']}")
|
||
|
||
# Definition
|
||
if 'Definition' in provide:
|
||
for definition in provide['Definition']:
|
||
if '#text' in definition:
|
||
access_info.append(f"Обозначение: {definition['#text']}")
|
||
|
||
# Other
|
||
if 'Other' in provide:
|
||
for other in provide['Other']:
|
||
if '#text' in other:
|
||
access_info.append(f"Прочее: {other['#text']}")
|
||
|
||
return access_info
|
||
|
||
def chec_object(self, atrebut):
|
||
if '@attributes' in atrebut:
|
||
if 'CadastralNumber' in atrebut['@attributes']:
|
||
return True
|
||
if 'Definition' in atrebut['@attributes']:
|
||
return True
|
||
if 'CadastralNumber' in atrebut:
|
||
if '#text' in atrebut['CadastralNumber'][0]:
|
||
return True
|
||
return False
|
||
|
||
def return_data(self):
|
||
return self.data
|
||
|
||
def return_connections_data(self):
|
||
return self.connections_data
|
||
|
||
def return_access_data(self):
|
||
return self.access_data
|
||
|
||
def __init__(self, xml):
|
||
self.data = []
|
||
self.connections_data = []
|
||
self.access_data = []
|
||
i = 0
|
||
for section_name, section_data in xml.get('Package', [{}])[0].items():
|
||
for group in section_data:
|
||
if isinstance(group, dict):
|
||
for item_type, item_list in group.items():
|
||
for obj in item_list:
|
||
if isinstance(obj, dict):
|
||
if self.chec_object(obj):
|
||
# Основные данные
|
||
self.data.append(
|
||
Data_farame(
|
||
section_KPTR=self.find_best_match(section_name),
|
||
cadastral_number=self.chec_kad_number(obj),
|
||
cadastral_quarter=self.chec_cad_block(obj),
|
||
object_type=self.chec_object_type(obj),
|
||
land_plot_area=self.chec_area(obj),
|
||
error_in_determining_area=self.chec_inaccuracy(obj),
|
||
availability_coordinates=self.chec_koordinat(obj),
|
||
explanations=self.chec_explanations(obj)
|
||
)
|
||
)
|
||
|
||
# Обработка связей
|
||
connections = self.get_connections(obj)
|
||
for connection in connections:
|
||
self.connections_data.append(
|
||
Connection_Data(
|
||
section_KPTR=self.find_best_match(section_name),
|
||
cadastral_number=self.chec_kad_number(obj),
|
||
cadastral_quarter=self.chec_cad_block(obj),
|
||
object_type=self.chec_object_type(obj),
|
||
connection=connection
|
||
)
|
||
)
|
||
|
||
# Обработка сведений о доступе (только для ЗУ)
|
||
if self.chec_object_type(obj) == "зу":
|
||
access_infos = self.get_access_info(obj)
|
||
for access_info in access_infos:
|
||
self.access_data.append(
|
||
Access_Data(
|
||
section_KPTR=self.find_best_match(section_name),
|
||
cadastral_number=self.chec_kad_number(obj),
|
||
cadastral_quarter=self.chec_cad_block(obj),
|
||
access_info=access_info
|
||
)
|
||
)
|
||
|
||
i += 1
|
||
if i == 785:
|
||
pass
|
||
else:
|
||
for inner_key, inner_items in obj.items():
|
||
for real_obj in inner_items:
|
||
if isinstance(real_obj, dict):
|
||
if self.chec_object(real_obj):
|
||
# Основные данные
|
||
self.data.append(
|
||
Data_farame(
|
||
section_KPTR=self.find_best_match(section_name),
|
||
cadastral_number=self.chec_kad_number(real_obj),
|
||
cadastral_quarter=self.chec_cad_block(real_obj),
|
||
object_type=self.chec_object_type(real_obj),
|
||
land_plot_area=self.chec_area(real_obj),
|
||
error_in_determining_area=self.chec_inaccuracy(real_obj),
|
||
availability_coordinates=self.chec_koordinat(real_obj),
|
||
explanations=self.chec_explanations(real_obj)
|
||
)
|
||
)
|
||
|
||
# Обработка связей
|
||
connections = self.get_connections(real_obj)
|
||
for connection in connections:
|
||
self.connections_data.append(
|
||
Connection_Data(
|
||
section_KPTR=self.find_best_match(section_name),
|
||
cadastral_number=self.chec_kad_number(real_obj),
|
||
cadastral_quarter=self.chec_cad_block(real_obj),
|
||
object_type=self.chec_object_type(real_obj),
|
||
connection=connection
|
||
)
|
||
)
|
||
|
||
# Обработка сведений о доступе (только для ЗУ)
|
||
if self.chec_object_type(real_obj) == "зу":
|
||
access_infos = self.get_access_info(real_obj)
|
||
for access_info in access_infos:
|
||
self.access_data.append(
|
||
Access_Data(
|
||
section_KPTR=self.find_best_match(section_name),
|
||
cadastral_number=self.chec_kad_number(real_obj),
|
||
cadastral_quarter=self.chec_cad_block(real_obj),
|
||
access_info=access_info
|
||
)
|
||
)
|
||
else:
|
||
for inner_key, inner_items in obj.items():
|
||
for real_obj_real in inner_items:
|
||
for real_wtf in real_obj_real:
|
||
|
||
test_item = real_obj_real[real_wtf][0]
|
||
if self.chec_object(test_item):
|
||
# Основные данные
|
||
self.data.append(
|
||
Data_farame(
|
||
section_KPTR=self.find_best_match(real_wtf),
|
||
cadastral_number=self.chec_kad_number(test_item),
|
||
cadastral_quarter=self.chec_cad_block(test_item),
|
||
object_type=self.chec_object_type(test_item),
|
||
land_plot_area=self.chec_area(test_item),
|
||
error_in_determining_area=self.chec_inaccuracy(test_item),
|
||
availability_coordinates=self.chec_koordinat(test_item),
|
||
explanations=self.chec_explanations(test_item)
|
||
)
|
||
)
|
||
|
||
# Обработка связей
|
||
connections = self.get_connections(test_item)
|
||
for connection in connections:
|
||
self.connections_data.append(
|
||
Connection_Data(
|
||
section_KPTR=self.find_best_match(real_wtf),
|
||
cadastral_number=self.chec_kad_number(test_item),
|
||
cadastral_quarter=self.chec_cad_block(test_item),
|
||
object_type=self.chec_object_type(test_item),
|
||
connection=connection
|
||
)
|
||
)
|
||
|
||
# Обработка сведений о доступе (только для ЗУ)
|
||
if self.chec_object_type(test_item) == "зу":
|
||
access_infos = self.get_access_info(test_item)
|
||
for access_info in access_infos:
|
||
self.access_data.append(
|
||
Access_Data(
|
||
section_KPTR=self.find_best_match(real_wtf),
|
||
cadastral_number=self.chec_kad_number(test_item),
|
||
cadastral_quarter=self.chec_cad_block(test_item),
|
||
access_info=access_info
|
||
)
|
||
)
|
||
i += 1
|
||
if i == 785:
|
||
pass
|
||
pass
|
||
|
||
|
||
|
||
|
||
|
||
class Parser():
|
||
|
||
def lxml_to_dict(self, element):
|
||
result = {}
|
||
|
||
# Обрабатываем атрибуты (если есть)
|
||
if element.attrib:
|
||
result["@attributes"] = element.attrib
|
||
|
||
# Обрабатываем текстовое содержимое (если есть)
|
||
if element.text and element.text.strip():
|
||
result["#text"] = element.text.strip()
|
||
|
||
# Группируем все дочерние элементы в списки
|
||
children_by_tag = {}
|
||
for child in element:
|
||
child_tag = child.tag
|
||
if child_tag not in children_by_tag:
|
||
children_by_tag[child_tag] = []
|
||
children_by_tag[child_tag].append(self.lxml_to_dict(child))
|
||
|
||
# Добавляем в результат (всегда списками)
|
||
for tag, children in children_by_tag.items():
|
||
result[tag] = children # Всегда массив, даже если элемент один
|
||
|
||
return result
|
||
|
||
def __init__(self, dir_xml=[], save_dir=""):
|
||
xlsx_file = xlsx_filer(path=save_dir)
|
||
all_data = []
|
||
all_connections_data = []
|
||
all_access_data = []
|
||
|
||
for item in dir_xml:
|
||
try:
|
||
tree = etree.parse(item)
|
||
temp = self.lxml_to_dict(tree.getroot())
|
||
new_data = super_visor(temp)
|
||
all_data.extend(new_data.return_data())
|
||
all_connections_data.extend(new_data.return_connections_data())
|
||
all_access_data.extend(new_data.return_access_data())
|
||
except Exception as e:
|
||
print(f"Ошибка при обработке файла {item}: {e}")
|
||
continue
|
||
|
||
xlsx_file.data = all_data
|
||
xlsx_file.connections_data = all_connections_data
|
||
xlsx_file.access_data = all_access_data
|
||
xlsx_file.path = save_dir
|
||
|
||
if all_data:
|
||
xlsx_file.update_data()
|
||
if all_connections_data:
|
||
xlsx_file.update_connections_data()
|
||
if all_access_data:
|
||
xlsx_file.update_access_data()
|
||
|
||
xlsx_file.save_file()
|