92 lines
4.2 KiB
Python
92 lines
4.2 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:
|
||
cadastral_numbers = xml_file.extract_cadastral_numbers()
|
||
|
||
for number in cadastral_numbers:
|
||
record = CadastralRecord(number, str(xml_file.file_path))
|
||
all_records.append(record)
|
||
|
||
self.cadastral_records = all_records
|
||
print(f"Всего найдено кадастровых номеров: {len(all_records)}")
|
||
return all_records
|
||
|
||
def get_unique_cadastral_numbers(self) -> Set[str]:
|
||
"""Возвращает уникальные кадастровые номера"""
|
||
return {record.number for record in self.cadastral_records}
|
||
|
||
def move_files_to_temp(self, cadastral_numbers_to_move: Set[str]) -> Dict[str, str]:
|
||
"""Перемещает файлы с указанными кадастровыми номерами во временную папку"""
|
||
moved_files = {}
|
||
|
||
for record in self.cadastral_records:
|
||
if record.number in cadastral_numbers_to_move:
|
||
source_path = Path(record.file_path)
|
||
if source_path.exists():
|
||
# Создаем уникальное имя для файла во временной папке
|
||
dest_name = f"{record.number.replace(':', '_')}_{source_path.name}"
|
||
dest_path = self.temp_directory / dest_name
|
||
|
||
try:
|
||
shutil.copy2(source_path, dest_path)
|
||
moved_files[record.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]:
|
||
"""Возвращает статистику обработки"""
|
||
return {
|
||
'total_xml_files': len(self.xml_files),
|
||
'files_with_numbers': len([f for f in self.xml_files if len(f.cadastral_numbers) > 0]),
|
||
'total_cadastral_numbers': len(self.cadastral_records),
|
||
'unique_cadastral_numbers': len(self.get_unique_cadastral_numbers())
|
||
}
|
||
|
||
def __str__(self):
|
||
stats = self.get_statistics()
|
||
return (f"FileSeeker: {stats['total_xml_files']} XML файлов, "
|
||
f"{stats['unique_cadastral_numbers']} уникальных номеров")
|