111 lines
5.4 KiB
Python
111 lines
5.4 KiB
Python
"""
|
||
Сервис для поиска и обработки 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']} всего уникальных номеров")
|