""" Сервис для поиска и обработки XML файлов. """ import os import shutil import tempfile from pathlib import Path from typing import List, Set, Dict from models.xml_file_info import XMLFileInfo from models.cadastral_record import CadastralRecord class FileSeeker: def __init__(self, base_directory: str, temp_directory: str = None): self.base_directory = Path(base_directory) self.temp_directory = Path(temp_directory) if temp_directory else Path(tempfile.mkdtemp(prefix="kadastr14_")) self.xml_files: List[XMLFileInfo] = [] self.cadastral_records: List[CadastralRecord] = [] # Создаем временную директорию если её нет self.temp_directory.mkdir(parents=True, exist_ok=True) def find_xml_files(self) -> List[XMLFileInfo]: """Находит все XML файлы в базовой директории""" xml_files = [] if not self.base_directory.exists(): print(f"Директория {self.base_directory} не существует") return xml_files for file_path in self.base_directory.rglob("*.xml"): if file_path.is_file(): xml_info = XMLFileInfo(str(file_path)) xml_files.append(xml_info) self.xml_files = xml_files print(f"Найдено XML файлов: {len(xml_files)}") return xml_files def process_files(self) -> List[CadastralRecord]: """Обрабатывает все найденные XML файлы и создает записи с основными и всеми номерами""" all_records = [] for xml_file in self.xml_files: main_number, all_numbers = xml_file.extract_cadastral_data() if main_number and all_numbers: record = CadastralRecord(main_number, str(xml_file.file_path), all_numbers) all_records.append(record) self.cadastral_records = all_records print(f"Создано записей объектов: {len(all_records)}") # Выводим статистику total_numbers = sum(len(record.all_numbers) for record in all_records) print(f"Всего номеров во всех объектах: {total_numbers}") return all_records def get_unique_cadastral_numbers(self) -> Set[str]: """Возвращает уникальные основные кадастровые номера""" return {record.main_number for record in self.cadastral_records} def get_all_cadastral_numbers(self) -> Set[str]: """Возвращает все уникальные кадастровые номера из всех объектов""" all_numbers = set() for record in self.cadastral_records: all_numbers.update(record.all_numbers) return all_numbers def move_files_to_temp(self, cadastral_numbers_to_move: Set[str]) -> Dict[str, str]: """Перемещает файлы с указанными кадастровыми номерами во временную папку""" moved_files = {} for record in self.cadastral_records: # Проверяем, есть ли какой-либо из номеров объекта в списке для перемещения if any(num in cadastral_numbers_to_move for num in record.all_numbers): source_path = Path(record.file_path) if source_path.exists(): # Создаем уникальное имя для файла во временной папке dest_name = f"{record.main_number.replace(':', '_')}_{source_path.name}" dest_path = self.temp_directory / dest_name try: shutil.copy2(source_path, dest_path) moved_files[record.main_number] = str(dest_path) print(f"Файл скопирован: {source_path.name} -> {dest_path}") except Exception as e: print(f"Ошибка при копировании файла {source_path}: {e}") return moved_files def get_statistics(self) -> Dict[str, int]: """Возвращает статистику обработки""" total_all_numbers = sum(len(record.all_numbers) for record in self.cadastral_records) return { 'total_xml_files': len(self.xml_files), 'files_with_numbers': len([f for f in self.xml_files if f.main_cadastral_number]), 'total_cadastral_objects': len(self.cadastral_records), 'unique_main_numbers': len(self.get_unique_cadastral_numbers()), 'total_all_numbers': total_all_numbers, 'unique_all_numbers': len(self.get_all_cadastral_numbers()) } def __str__(self): stats = self.get_statistics() return (f"FileSeeker: {stats['total_xml_files']} XML файлов, " f"{stats['total_cadastral_objects']} объектов, " f"{stats['unique_main_numbers']} уникальных основных номеров, " f"{stats['unique_all_numbers']} всего уникальных номеров")