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()