xml
This commit is contained in:
505
seekers.py
Normal file
505
seekers.py
Normal file
@@ -0,0 +1,505 @@
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from lxml import etree
|
||||
import re
|
||||
|
||||
class CadastralRecord:
|
||||
"""Класс для хранения информации о кадастровом номере и связанных с ним данных."""
|
||||
|
||||
def __init__(self, cadastral_number, source_xml_info, temp_file_path=None):
|
||||
self.cadastral_number = cadastral_number # Основной кадастровый номер
|
||||
self.source_xml_info = source_xml_info # Ссылка на XMLFileInfo где найден
|
||||
self.temp_file_path = temp_file_path # Путь к файлу во временной папке
|
||||
self.quarter_number = self._extract_quarter_number() # Номер квартала
|
||||
self.related_numbers = [] # Смежные кадастровые номера из того же файла
|
||||
|
||||
def _extract_quarter_number(self):
|
||||
"""Извлекает номер квартала из кадастрового номера."""
|
||||
# Квартал - это первые три части номера (NN:NN:NNNNNNN)
|
||||
parts = self.cadastral_number.split(':')
|
||||
if len(parts) >= 3:
|
||||
return ':'.join(parts[:3])
|
||||
return None
|
||||
|
||||
def set_related_numbers(self, all_numbers_from_file):
|
||||
"""Устанавливает список смежных кадастровых номеров из того же файла."""
|
||||
# Убираем сам номер из списка смежных
|
||||
self.related_numbers = [num for num in all_numbers_from_file if num != self.cadastral_number]
|
||||
|
||||
def get_object_type(self):
|
||||
"""Определяет тип объекта по кадастровому номеру."""
|
||||
parts = self.cadastral_number.split(':')
|
||||
if len(parts) == 4:
|
||||
# Полный кадастровый номер объекта недвижимости
|
||||
return "Объект недвижимости"
|
||||
elif len(parts) == 3:
|
||||
# Номер квартала
|
||||
return "Квартал"
|
||||
else:
|
||||
return "Неизвестный тип"
|
||||
|
||||
def __str__(self):
|
||||
return f"CadastralRecord({self.cadastral_number}, квартал: {self.quarter_number})"
|
||||
|
||||
def __repr__(self):
|
||||
return self.__str__()
|
||||
|
||||
class XMLFileInfo:
|
||||
"""Класс для хранения информации о XML файле и его происхождении."""
|
||||
|
||||
def __init__(self, original_name, content, source_path, archive_name=None):
|
||||
self.original_name = original_name # Оригинальное имя файла
|
||||
self.content = content # Содержимое файла (bytes)
|
||||
self.source_path = source_path # Полный путь к источнику (файлу или архиву)
|
||||
self.archive_name = archive_name # Имя архива, если файл из архива
|
||||
self.is_from_archive = archive_name is not None
|
||||
self.cadastral_numbers = [] # Список найденных кадастровых номеров
|
||||
self.analysis_completed = False # Флаг выполненного анализа
|
||||
|
||||
def analyze_xml_content(self):
|
||||
"""Анализирует содержимое XML файла и извлекает кадастровые номера."""
|
||||
if self.analysis_completed:
|
||||
return self.cadastral_numbers
|
||||
|
||||
try:
|
||||
# Парсим XML из bytes
|
||||
root = etree.fromstring(self.content)
|
||||
self.cadastral_numbers = self._extract_cadastral_numbers(root)
|
||||
self.analysis_completed = True
|
||||
except Exception as e:
|
||||
print(f"Ошибка при анализе XML файла {self.original_name}: {e}")
|
||||
self.cadastral_numbers = []
|
||||
self.analysis_completed = True
|
||||
|
||||
return self.cadastral_numbers
|
||||
|
||||
def _extract_cadastral_numbers(self, element):
|
||||
"""Извлекает кадастровые номера из XML элемента."""
|
||||
cadastral_numbers = set()
|
||||
|
||||
# Ищем кадастровые номера в атрибутах
|
||||
if element.attrib:
|
||||
for attr_name, attr_value in element.attrib.items():
|
||||
if 'CadastralNumber' in attr_name or 'cadastralnumber' in attr_name.lower():
|
||||
if self._is_valid_cadastral_number(attr_value):
|
||||
cadastral_numbers.add(attr_value)
|
||||
elif 'Definition' in attr_name:
|
||||
if self._is_valid_cadastral_number(attr_value):
|
||||
cadastral_numbers.add(attr_value)
|
||||
|
||||
# Ищем кадастровые номера в тексте элементов
|
||||
if element.text and element.text.strip():
|
||||
text = element.text.strip()
|
||||
if self._is_valid_cadastral_number(text):
|
||||
cadastral_numbers.add(text)
|
||||
|
||||
# Рекурсивно проходим по всем дочерним элементам
|
||||
for child in element:
|
||||
# Проверяем название тега на наличие ключевых слов
|
||||
tag_lower = child.tag.lower()
|
||||
if any(keyword in tag_lower for keyword in ['cad_number', 'cadastralnumber', 'quarter_cad_number', 'land_cad_number']):
|
||||
if child.text and child.text.strip():
|
||||
text = child.text.strip()
|
||||
if self._is_valid_cadastral_number(text):
|
||||
cadastral_numbers.add(text)
|
||||
|
||||
# Рекурсивно анализируем дочерние элементы
|
||||
child_numbers = self._extract_cadastral_numbers(child)
|
||||
cadastral_numbers.update(child_numbers)
|
||||
|
||||
return list(cadastral_numbers)
|
||||
|
||||
def _is_valid_cadastral_number(self, value):
|
||||
"""Проверяет, является ли строка валидным кадастровым номером."""
|
||||
if not value or not isinstance(value, str):
|
||||
return False
|
||||
|
||||
# Паттерн для кадастрового номера: NN:NN:NNNNNNN:NNNN
|
||||
# где N - цифра, количество цифр может варьироваться
|
||||
pattern = r'^\d{1,2}:\d{1,2}:\d{6,7}:\d{1,4}$'
|
||||
return bool(re.match(pattern, value.strip()))
|
||||
|
||||
def get_cadastral_numbers(self):
|
||||
"""Возвращает список кадастровых номеров (с анализом если не выполнен)."""
|
||||
if not self.analysis_completed:
|
||||
self.analyze_xml_content()
|
||||
return self.cadastral_numbers
|
||||
|
||||
def create_cadastral_records(self, temp_file_path=None):
|
||||
"""Создает объекты CadastralRecord для каждого найденного кадастрового номера."""
|
||||
cadastral_numbers = self.get_cadastral_numbers()
|
||||
records = []
|
||||
|
||||
for number in cadastral_numbers:
|
||||
record = CadastralRecord(number, self, temp_file_path)
|
||||
record.set_related_numbers(cadastral_numbers)
|
||||
records.append(record)
|
||||
|
||||
return records
|
||||
|
||||
def get_display_name(self):
|
||||
"""Возвращает удобное для отображения имя файла."""
|
||||
if self.is_from_archive:
|
||||
return f"{self.original_name} (из архива {self.archive_name})"
|
||||
else:
|
||||
return self.original_name
|
||||
|
||||
def get_safe_filename(self):
|
||||
"""Возвращает безопасное имя файла для сохранения."""
|
||||
if self.is_from_archive:
|
||||
archive_stem = Path(self.archive_name).stem
|
||||
return f"ARCHIVE_{archive_stem}_{Path(self.original_name).stem}.xml"
|
||||
else:
|
||||
return self.original_name
|
||||
|
||||
def save_to_temp(self, temp_dir):
|
||||
"""Сохраняет файл во временную папку."""
|
||||
safe_name = self.get_safe_filename()
|
||||
dest_path = temp_dir / safe_name
|
||||
|
||||
# Если файл уже существует, добавляем номер
|
||||
counter = 1
|
||||
while dest_path.exists():
|
||||
name_parts = dest_path.stem, counter, dest_path.suffix
|
||||
dest_path = temp_dir / f"{name_parts[0]}_{name_parts[1]}{name_parts[2]}"
|
||||
counter += 1
|
||||
|
||||
dest_path.write_bytes(self.content)
|
||||
return dest_path
|
||||
|
||||
class FileSeeker:
|
||||
def __init__(self, search_dir):
|
||||
self.search_dir = Path(search_dir)
|
||||
# Создаем временную папку рядом с программой
|
||||
program_dir = Path(__file__).parent
|
||||
self.temp_dir = program_dir / "kadastr14_temp"
|
||||
self.temp_dir.mkdir(exist_ok=True)
|
||||
print(f"Временная папка создана: {self.temp_dir}")
|
||||
|
||||
def find_xml_files(self):
|
||||
"""Ищет все XML-файлы рекурсивно в директории поиска и возвращает объекты XMLFileInfo."""
|
||||
xml_files = list(self.search_dir.rglob('*.xml'))
|
||||
xml_file_objects = []
|
||||
|
||||
print(f"Найдено XML файлов: {len(xml_files)}")
|
||||
for file_path in xml_files:
|
||||
print(f" - {file_path}")
|
||||
try:
|
||||
content = file_path.read_bytes()
|
||||
xml_info = XMLFileInfo(
|
||||
original_name=file_path.name,
|
||||
content=content,
|
||||
source_path=str(file_path)
|
||||
)
|
||||
xml_file_objects.append(xml_info)
|
||||
except Exception as e:
|
||||
print(f" Ошибка при чтении {file_path}: {e}")
|
||||
|
||||
return xml_file_objects
|
||||
|
||||
def find_archives(self):
|
||||
"""Ищет все архивы (ZIP) рекурсивно в директории поиска."""
|
||||
zip_files = list(self.search_dir.rglob('*.zip'))
|
||||
print(f"Найдено ZIP архивов: {len(zip_files)}")
|
||||
for file in zip_files:
|
||||
print(f" - {file}")
|
||||
return zip_files
|
||||
|
||||
def extract_xml_from_archive(self, archive_path):
|
||||
"""Извлекает XML файлы из архива и возвращает объекты XMLFileInfo."""
|
||||
xml_file_objects = []
|
||||
archive_name = archive_path.name
|
||||
|
||||
try:
|
||||
with zipfile.ZipFile(archive_path, 'r') as zip_ref:
|
||||
# Получаем список файлов в архиве
|
||||
file_list = zip_ref.namelist()
|
||||
xml_files_in_archive = [f for f in file_list if f.lower().endswith('.xml')]
|
||||
|
||||
print(f" В архиве {archive_name} найдено XML файлов: {len(xml_files_in_archive)}")
|
||||
|
||||
for xml_file in xml_files_in_archive:
|
||||
try:
|
||||
# Читаем содержимое файла из архива
|
||||
content = zip_ref.read(xml_file)
|
||||
xml_info = XMLFileInfo(
|
||||
original_name=Path(xml_file).name,
|
||||
content=content,
|
||||
source_path=str(archive_path),
|
||||
archive_name=archive_name
|
||||
)
|
||||
xml_file_objects.append(xml_info)
|
||||
print(f" Найден в архиве: {xml_file}")
|
||||
except Exception as e:
|
||||
print(f" Ошибка при извлечении {xml_file}: {e}")
|
||||
|
||||
except Exception as e:
|
||||
print(f" Ошибка при обработке архива {archive_path}: {e}")
|
||||
|
||||
return xml_file_objects
|
||||
|
||||
def copy_to_temp(self):
|
||||
"""Собирает все XML файлы (обычные и из архивов) и сохраняет их во временную папку."""
|
||||
all_xml_objects = []
|
||||
|
||||
# Сначала собираем обычные XML файлы
|
||||
print(f"\nПоиск обычных XML файлов...")
|
||||
xml_objects = self.find_xml_files()
|
||||
all_xml_objects.extend(xml_objects)
|
||||
|
||||
# Теперь обрабатываем архивы
|
||||
print(f"\nПоиск и обработка архивов...")
|
||||
archives = self.find_archives()
|
||||
|
||||
for archive_path in archives:
|
||||
print(f"\nОбработка архива: {archive_path.name}")
|
||||
archive_xml_objects = self.extract_xml_from_archive(archive_path)
|
||||
all_xml_objects.extend(archive_xml_objects)
|
||||
|
||||
# Сохраняем все найденные XML файлы во временную папку
|
||||
print(f"\nСохранение XML файлов в: {self.temp_dir}")
|
||||
saved_files = []
|
||||
|
||||
for xml_obj in all_xml_objects:
|
||||
try:
|
||||
saved_path = xml_obj.save_to_temp(self.temp_dir)
|
||||
saved_files.append(saved_path)
|
||||
print(f" Сохранен: {xml_obj.get_display_name()} -> {saved_path.name}")
|
||||
except Exception as e:
|
||||
print(f" Ошибка при сохранении {xml_obj.get_display_name()}: {e}")
|
||||
|
||||
print(f"\nВсего найдено XML файлов: {len(all_xml_objects)}")
|
||||
print(f"Успешно сохранено: {len(saved_files)}")
|
||||
|
||||
return all_xml_objects, saved_files
|
||||
|
||||
def collect_all_cadastral_records(self, xml_objects, saved_files):
|
||||
"""Собирает все кадастровые записи из всех XML файлов."""
|
||||
print(f"\n{'='*80}")
|
||||
print(f"СБОР КАДАСТРОВЫХ ЗАПИСЕЙ")
|
||||
print(f"{'='*80}")
|
||||
|
||||
all_records = []
|
||||
file_path_mapping = {}
|
||||
|
||||
# Создаем маппинг XML объектов к путям сохраненных файлов
|
||||
for i, xml_obj in enumerate(xml_objects):
|
||||
if i < len(saved_files):
|
||||
file_path_mapping[xml_obj] = saved_files[i]
|
||||
|
||||
# Собираем записи из каждого файла
|
||||
for xml_obj in xml_objects:
|
||||
temp_file_path = file_path_mapping.get(xml_obj)
|
||||
records = xml_obj.create_cadastral_records(temp_file_path)
|
||||
all_records.extend(records)
|
||||
|
||||
if records:
|
||||
print(f"📄 {xml_obj.get_display_name()}: {len(records)} кадастровых записей")
|
||||
|
||||
print(f"\nВсего собрано кадастровых записей: {len(all_records)}")
|
||||
return all_records
|
||||
|
||||
def print_cadastral_records_summary(self, records):
|
||||
"""Выводит сводку по всем кадастровым записям."""
|
||||
print(f"\n{'='*80}")
|
||||
print(f"СВОДКА ПО КАДАСТРОВЫМ ЗАПИСЯМ")
|
||||
print(f"{'='*80}")
|
||||
|
||||
if not records:
|
||||
print("Кадастровые записи не найдены.")
|
||||
return
|
||||
|
||||
# Группируем по типам объектов
|
||||
by_type = {}
|
||||
by_quarter = {}
|
||||
unique_numbers = set()
|
||||
|
||||
for record in records:
|
||||
obj_type = record.get_object_type()
|
||||
if obj_type not in by_type:
|
||||
by_type[obj_type] = []
|
||||
by_type[obj_type].append(record)
|
||||
|
||||
if record.quarter_number:
|
||||
if record.quarter_number not in by_quarter:
|
||||
by_quarter[record.quarter_number] = []
|
||||
by_quarter[record.quarter_number].append(record)
|
||||
|
||||
unique_numbers.add(record.cadastral_number)
|
||||
|
||||
print(f"📊 Общая статистика:")
|
||||
print(f" Всего записей: {len(records)}")
|
||||
print(f" Уникальных номеров: {len(unique_numbers)}")
|
||||
print(f" Кварталов: {len(by_quarter)}")
|
||||
|
||||
print(f"\n📋 По типам объектов:")
|
||||
for obj_type, type_records in by_type.items():
|
||||
print(f" {obj_type}: {len(type_records)} записей")
|
||||
|
||||
print(f"\n🏘️ По кварталам (топ-10):")
|
||||
sorted_quarters = sorted(by_quarter.items(), key=lambda x: len(x[1]), reverse=True)
|
||||
for quarter, quarter_records in sorted_quarters[:10]:
|
||||
print(f" {quarter}: {len(quarter_records)} объектов")
|
||||
|
||||
def print_detailed_cadastral_records(self, records, max_records=20):
|
||||
"""Выводит детальную информацию о кадастровых записях."""
|
||||
print(f"\n{'='*80}")
|
||||
print(f"ДЕТАЛЬНАЯ ИНФОРМАЦИЯ О КАДАСТРОВЫХ ЗАПИСЯХ (первые {max_records})")
|
||||
print(f"{'='*80}")
|
||||
|
||||
for i, record in enumerate(records[:max_records], 1):
|
||||
print(f"\n{i}. 🏠 {record.cadastral_number}")
|
||||
print(f" 📍 Тип: {record.get_object_type()}")
|
||||
if record.quarter_number:
|
||||
print(f" 🏘️ Квартал: {record.quarter_number}")
|
||||
|
||||
print(f" 📁 Источник: {record.source_xml_info.get_display_name()}")
|
||||
print(f" 📂 Оригинальный файл: {record.source_xml_info.source_path}")
|
||||
|
||||
if record.temp_file_path:
|
||||
print(f" 💾 Временный файл: {record.temp_file_path}")
|
||||
|
||||
if record.related_numbers:
|
||||
print(f" 🔗 Смежные номера ({len(record.related_numbers)}):")
|
||||
for j, related in enumerate(record.related_numbers[:3], 1):
|
||||
print(f" {j}. {related}")
|
||||
if len(record.related_numbers) > 3:
|
||||
print(f" ... и еще {len(record.related_numbers) - 3} номеров")
|
||||
else:
|
||||
print(f" 🔗 Смежные номера: нет")
|
||||
|
||||
if len(records) > max_records:
|
||||
print(f"\n... и еще {len(records) - max_records} записей")
|
||||
|
||||
def print_all_objects_info(self, xml_objects):
|
||||
"""Выводит подробную информацию о всех найденных XML объектах."""
|
||||
print(f"\n{'='*80}")
|
||||
print(f"ПОДРОБНАЯ ИНФОРМАЦИЯ О ВСЕХ НАЙДЕННЫХ XML ОБЪЕКТАХ")
|
||||
print(f"{'='*80}")
|
||||
print(f"Всего объектов: {len(xml_objects)}")
|
||||
|
||||
if not xml_objects:
|
||||
print("Объекты не найдены.")
|
||||
return
|
||||
|
||||
# Статистика по источникам
|
||||
from_files = sum(1 for obj in xml_objects if not obj.is_from_archive)
|
||||
from_archives = sum(1 for obj in xml_objects if obj.is_from_archive)
|
||||
|
||||
print(f"\nСтатистика по источникам:")
|
||||
print(f" - Из обычных файлов: {from_files}")
|
||||
print(f" - Из архивов: {from_archives}")
|
||||
|
||||
# Группируем по архивам для удобства
|
||||
archive_groups = {}
|
||||
regular_files = []
|
||||
|
||||
for obj in xml_objects:
|
||||
if obj.is_from_archive:
|
||||
if obj.archive_name not in archive_groups:
|
||||
archive_groups[obj.archive_name] = []
|
||||
archive_groups[obj.archive_name].append(obj)
|
||||
else:
|
||||
regular_files.append(obj)
|
||||
|
||||
# Выводим обычные файлы
|
||||
if regular_files:
|
||||
print(f"\n{'-'*60}")
|
||||
print(f"ОБЫЧНЫЕ XML ФАЙЛЫ ({len(regular_files)} файлов)")
|
||||
print(f"{'-'*60}")
|
||||
|
||||
for i, obj in enumerate(regular_files, 1):
|
||||
self._print_object_details(obj, i)
|
||||
|
||||
# Выводим файлы из архивов
|
||||
if archive_groups:
|
||||
print(f"\n{'-'*60}")
|
||||
print(f"XML ФАЙЛЫ ИЗ АРХИВОВ ({len(archive_groups)} архивов, {from_archives} файлов)")
|
||||
print(f"{'-'*60}")
|
||||
|
||||
counter = 1
|
||||
for archive_name, objects in archive_groups.items():
|
||||
print(f"\n📦 АРХИВ: {archive_name}")
|
||||
print(f" Путь: {objects[0].source_path}")
|
||||
print(f" XML файлов в архиве: {len(objects)}")
|
||||
|
||||
for obj in objects:
|
||||
self._print_object_details(obj, counter, indent=" ")
|
||||
counter += 1
|
||||
|
||||
print(f"\n{'='*80}")
|
||||
print(f"ИТОГО: {len(xml_objects)} XML файлов обработано")
|
||||
print(f"{'='*80}")
|
||||
|
||||
def _print_object_details(self, obj, number, indent=""):
|
||||
"""Выводит детальную информацию об одном XML объекте."""
|
||||
print(f"\n{indent}{number}. {obj.get_display_name()}")
|
||||
print(f"{indent} 📄 Исходное имя: {obj.original_name}")
|
||||
print(f"{indent} 📁 Источник: {obj.source_path}")
|
||||
print(f"{indent} 💾 Размер содержимого: {len(obj.content):,} байт")
|
||||
print(f"{indent} 🗂️ Безопасное имя: {obj.get_safe_filename()}")
|
||||
|
||||
# Запускаем анализ кадастровых номеров если не выполнен
|
||||
cadastral_numbers = obj.get_cadastral_numbers() # Это запустит анализ если нужно
|
||||
cadastral_count = len(cadastral_numbers)
|
||||
|
||||
print(f"{indent} 🔍 Анализ завершен: найдено {cadastral_count} кадастровых номеров")
|
||||
if cadastral_count > 0:
|
||||
# Показываем первые 3 номера
|
||||
for i, num in enumerate(cadastral_numbers[:3], 1):
|
||||
print(f"{indent} {i}. {num}")
|
||||
if cadastral_count > 3:
|
||||
print(f"{indent} ... и еще {cadastral_count - 3} номеров")
|
||||
else:
|
||||
print(f"{indent} ❌ Кадастровые номера не найдены")
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Пример использования - ищем во всей папке с примерами
|
||||
search_path = r"C:\Users\USER\Downloads\14. выборка выписок из ЕГРН по списку КН"
|
||||
print(f"Начинаем поиск XML файлов в: {search_path}")
|
||||
print("Поиск будет выполняться рекурсивно во всех подпапках...")
|
||||
|
||||
seeker = FileSeeker(search_path)
|
||||
xml_objects, saved_files = seeker.copy_to_temp()
|
||||
|
||||
print(f"\nГотово! Временная папка: {seeker.temp_dir}")
|
||||
print(f"Найдено XML объектов: {len(xml_objects)}")
|
||||
print(f"Сохранено файлов: {len(saved_files)}")
|
||||
|
||||
# Собираем все кадастровые записи
|
||||
cadastral_records = seeker.collect_all_cadastral_records(xml_objects, saved_files)
|
||||
|
||||
# Выводим сводку
|
||||
seeker.print_cadastral_records_summary(cadastral_records)
|
||||
|
||||
# Выводим детальную информацию
|
||||
seeker.print_detailed_cadastral_records(cadastral_records, max_records=10)
|
||||
|
||||
# Дополнительный анализ - находим записи с наибольшим количеством смежных номеров
|
||||
if cadastral_records:
|
||||
print(f"\n{'='*80}")
|
||||
print(f"ЗАПИСИ С НАИБОЛЬШИМ КОЛИЧЕСТВОМ СМЕЖНЫХ НОМЕРОВ")
|
||||
print(f"{'='*80}")
|
||||
|
||||
# Сортируем по количеству смежных номеров
|
||||
sorted_by_related = sorted(cadastral_records, key=lambda x: len(x.related_numbers), reverse=True)
|
||||
|
||||
for i, record in enumerate(sorted_by_related[:5], 1):
|
||||
print(f"\n{i}. {record.cadastral_number}")
|
||||
print(f" Смежных номеров: {len(record.related_numbers)}")
|
||||
print(f" Источник: {record.source_xml_info.get_display_name()}")
|
||||
if record.related_numbers:
|
||||
print(f" Связанные номера: {', '.join(record.related_numbers[:5])}")
|
||||
if len(record.related_numbers) > 5:
|
||||
print(f" ... и еще {len(record.related_numbers) - 5} номеров")
|
||||
|
||||
print(f"\n{'='*80}")
|
||||
print(f"АНАЛИЗ ЗАВЕРШЕН")
|
||||
print(f"{'='*80}")
|
||||
print(f"📊 Всего кадастровых записей: {len(cadastral_records)}")
|
||||
print(f"📁 Временная папка: {seeker.temp_dir}")
|
||||
print(f"💾 Сохранено XML файлов: {len(saved_files)}")
|
||||
Reference in New Issue
Block a user