core/scanner.py - master - O.S.Prog/DiScan

core/scanner.py
513 строк21 KB
O.S.Prog
Initial commit: DiScan v1.1.0
18 июл 2026, 15:51
18 июл 2026, 15:51238a6ec
# core/scanner.py
"""
Главный сканер DICOM-файлов
"""

import os
import re
import warnings
from pathlib import Path
from typing import List, Optional, Set, Tuple, Dict
from datetime import datetime
import pydicom

# Подавляем предупреждения pydicom
warnings.filterwarnings("ignore", category=UserWarning, module="pydicom")
warnings.filterwarnings("ignore", category=UserWarning, module="pydicom.charset")
warnings.filterwarnings("ignore", category=UserWarning, module="pydicom.valuerep")

from core.models import PatientInfo, ScanResult


class DICOMScanner:
    """Главный сканер DICOM-файлов"""
    
    def __init__(self, root_path: str, encoding: str = 'cp1251', max_files: int = 10000):
        self.root_path = Path(root_path)
        self.supported_extensions = {'.dcm', '.dicom'}
        self.encoding = encoding
        self.max_files = max_files  # Максимальное количество файлов для сканирования
        self.scanned_files = 0
        self.found_dicom = 0
        self._stop_scanning = False
        
    def set_encoding(self, encoding: str):
        """Устанавливает кодировку для декодирования"""
        self.encoding = encoding             
        
    def _group_files_by_patient_id(self, dicom_files: List[Path], progress_callback=None) -> Dict[str, List[Path]]:
        """
        Группирует файлы по PatientID из DICOM-тегов
        """
        groups = {}
        processed = 0
        total = len(dicom_files)
        last_update_percent = -1
        last_update_count = 0
        
        for file_path in dicom_files:
            # ← ПРОВЕРКА ОСТАНОВКИ ВНУТРИ ЦИКЛА
            if self._stop_scanning:
                break
                
            processed += 1
            
            percent = int((processed / total) * 100)
            
            if (percent - last_update_percent >= 3) or (processed - last_update_count >= 200) or (processed == total):
                if progress_callback:
                    progress_callback(
                        f"🔄 Группировка: {processed}/{total} файлов ({percent}%)",
                        20 + int(percent * 0.1)
                    )
                last_update_percent = percent
                last_update_count = processed
            
            try:
                ds = pydicom.dcmread(str(file_path), force=True, stop_before_pixels=True)
                
                patient_id = None
                patient_name = None
                
                if hasattr(ds, 'PatientID') and ds.PatientID:
                    patient_id = str(ds.PatientID).strip()
                
                if not patient_id and hasattr(ds, 'PatientName') and ds.PatientName:
                    patient_name = self._decode_name(str(ds.PatientName))
                    patient_id = patient_name if patient_name else None
                
                if not patient_id:
                    parent_name = file_path.parent.name
                    if parent_name and parent_name not in ['DX', 'dx', 'Images', 'images']:
                        patient_id = parent_name
                    else:
                        grand_parent = file_path.parent.parent.name
                        patient_id = grand_parent if grand_parent else file_path.stem
                
                if patient_id and len(str(patient_id)) > 50:
                    patient_id = str(patient_id)[:50]
                
                if patient_id not in groups:
                    groups[patient_id] = []
                groups[patient_id].append(file_path)
                
            except Exception as e:
                parent_name = file_path.parent.name
                if parent_name and parent_name not in ['DX', 'dx']:
                    patient_id = parent_name
                else:
                    patient_id = file_path.parent.parent.name or file_path.stem
                
                if patient_id not in groups:
                    groups[patient_id] = []
                groups[patient_id].append(file_path)
        
        return groups
    
    def _process_patient_group(self, patient_id: str, files: List[Path]) -> Optional[PatientInfo]:
        """
        Обрабатывает группу файлов одного пациента
        """
        if not files:
            return None
        
        files.sort()
        first_file = files[0]
        
        try:
            # Читаем метаданные из первого файла
            patient_name, study_date, study_uid = self._extract_metadata(first_file)
            
            # Если имя не найдено, ищем в других файлах
            if not patient_name or patient_name == "" or patient_name == "❌ Имя не найдено":
                patient_name = self._find_patient_name_in_group(files)
            
            # Если имя всё ещё не найдено - используем ID
            if not patient_name or patient_name == "" or patient_name == "❌ Имя не найдено":
                patient_name = f"Пациент {patient_id}"
            
            # Определяем количество снимков
            image_count, series_uids = self._count_images(files)
            
            # Если дата не найдена, извлекаем из пути или из имени файла
            if not study_date or study_date == "Неизвестно":
                study_date = self._extract_date_from_files(files) or self._extract_date_from_path(self.root_path)
            
            # Путь к папке - определяем общий родительский путь
            folder_path = self._get_common_parent(files) or self.root_path
            
            return PatientInfo(
                patient_id=patient_id,
                patient_name=patient_name,
                folder_path=folder_path,
                study_date=study_date or "Неизвестно",
                total_files=len(files),
                image_count=image_count,
                sample_file=first_file,
                series_uids=series_uids,
                study_uid=study_uid
            )
            
        except Exception as e:
            return PatientInfo(
                patient_id=patient_id,
                patient_name="❌ Ошибка",
                folder_path=self.root_path,
                study_date="Ошибка",
                error=str(e)
            )
    
    def _get_common_parent(self, files: List[Path]) -> Optional[Path]:
        """Находит общий родительский путь для всех файлов"""
        if not files:
            return None
        
        if len(files) == 1:
            return files[0].parent
        
        # Находим общий префикс путей
        paths = [str(f.parent) for f in files]
        common = Path(os.path.commonpath(paths)) if paths else None
        return common
    
    def _extract_date_from_files(self, files: List[Path]) -> Optional[str]:
        """Извлекает дату из метаданных файлов"""
        for file_path in files[:10]:  # Проверяем первые 10 файлов
            try:
                ds = pydicom.dcmread(str(file_path), force=True, stop_before_pixels=True)
                if hasattr(ds, 'StudyDate') and ds.StudyDate:
                    study_date = str(ds.StudyDate)
                    if len(study_date) == 8:
                        return f"{study_date[:4]}-{study_date[4:6]}-{study_date[6:8]}"
            except:
                continue
        return None
          
    def _extract_metadata(self, dicom_path: Path) -> Tuple[Optional[str], Optional[str], Optional[str]]:
        """Извлекает метаданные из DICOM файла"""
        try:
            ds = pydicom.dcmread(str(dicom_path), force=True, stop_before_pixels=True)
            
            patient_name = None
            if hasattr(ds, 'PatientName') and ds.PatientName:
                raw_name = str(ds.PatientName)
                patient_name = self._decode_name(raw_name)
            
            study_date = None
            if hasattr(ds, 'StudyDate') and ds.StudyDate:
                study_date = str(ds.StudyDate)
                if len(study_date) == 8:
                    study_date = f"{study_date[:4]}-{study_date[4:6]}-{study_date[6:8]}"
            
            study_uid = None
            if hasattr(ds, 'StudyInstanceUID'):
                study_uid = str(ds.StudyInstanceUID)
            
            return patient_name, study_date, study_uid
            
        except Exception as e:
            print(f"⚠️ Ошибка чтения метаданных: {e}")
            return None, None, None
    
    def _decode_name(self, raw_name: str) -> str:
        """Декодирует имя пациента с текущей кодировкой"""
        if not raw_name:
            return raw_name
        
        try:
            decoded = raw_name.encode('latin1').decode(self.encoding)
            if any(('\u0400' <= c <= '\u04FF') for c in decoded):
                if decoded.startswith('^^'):
                    decoded = decoded[2:]
                return decoded.strip()
        except:
            pass
        
        for enc in ['cp1251', 'mac-cyrillic', 'koi8-r', 'iso-8859-5', 'cp866']:
            if enc == self.encoding:
                continue
            try:
                decoded = raw_name.encode('latin1').decode(enc)
                if any(('\u0400' <= c <= '\u04FF') for c in decoded):
                    if decoded.startswith('^^'):
                        decoded = decoded[2:]
                    return decoded.strip()
            except:
                continue
        
        if raw_name.startswith('^^'):
            raw_name = raw_name[2:]
        return raw_name.strip()
    
    def _find_patient_name_in_group(self, dicom_files: List[Path]) -> Optional[str]:
        """Ищет имя пациента во всех файлах группы"""
        for file_path in dicom_files[:50]:  # Проверяем первые 50 файлов
            try:
                ds = pydicom.dcmread(str(file_path), force=True, stop_before_pixels=True)
                if hasattr(ds, 'PatientName') and ds.PatientName:
                    raw_name = str(ds.PatientName)
                    decoded = self._decode_name(raw_name)
                    if decoded and decoded != "❌ Имя не найдено" and len(decoded) > 1:
                        return decoded
            except:
                continue
        return None
    
    def _count_images(self, dicom_files: List[Path]) -> Tuple[int, Set[str]]:
        """
        Определяет количество снимков.
        
        Логика:
        1. Если в папке есть BMP-файлы — считаем уникальные имена BMP (1 BMP = 1 снимок)
        2. Если BMP нет — считаем уникальные SeriesInstanceUID
        3. Если ничего не найдено — считаем количество DICOM-файлов
        """
        if not dicom_files:
            return 0, set()
        
        first_file = dicom_files[0]
        parent_folder = first_file.parent
        
        # ===== 1. ПРОВЕРЯЕМ НАЛИЧИЕ РАСТРОВЫХ ФАЙЛОВ (BMP) =====
        raster_extensions = ['.bmp', '.BMP', '.bpm', '.BPM']
        raster_files = []
        
        for ext in raster_extensions:
            raster_files.extend([f for f in parent_folder.glob(f'*{ext}') if f.is_file()])
        
        # Если есть BMP — считаем уникальные имена (1 BMP = 1 снимок)
        if raster_files:
            unique_names = set()
            for f in raster_files:
                name_without_ext = f.stem
                unique_names.add(name_without_ext)
            
            return len(unique_names), set()
        
        # ===== 2. НЕТ BMP — ИСПОЛЬЗУЕМ SeriesInstanceUID =====
        series_uids = set()
        limit = min(len(dicom_files), 200)
        
        for file_path in dicom_files[:limit]:
            try:
                ds = pydicom.dcmread(str(file_path), force=True, stop_before_pixels=True)
                if hasattr(ds, 'SeriesInstanceUID'):
                    series_uids.add(str(ds.SeriesInstanceUID))
            except:
                continue
        
        if series_uids:
            return len(series_uids), series_uids
        
        # ===== 3. НИЧЕГО НЕ НАШЛИ =====
        return len(dicom_files), set()
    
    def _extract_date_from_path(self, folder: Path) -> str:
        """Извлекает дату из пути"""
        for part in folder.parts:
            try:
                datetime.strptime(part, '%Y-%m-%d')
                return part
            except ValueError:
                pass
            
            try:
                if len(part) == 8 and part.isdigit():
                    dt = datetime.strptime(part, '%Y%m%d')
                    return dt.strftime('%Y-%m-%d')
            except ValueError:
                pass
        
        return "Неизвестно"
    
    def reencode_patients(self, patients: List[PatientInfo], new_encoding: str) -> List[PatientInfo]:
        """Перекодирует имена пациентов с новой кодировкой"""
        self.encoding = new_encoding
        result = []
        
        for patient in patients:
            if patient.error or patient.patient_name == "❌ Имя не найдено":
                result.append(patient)
                continue
            
            try:
                if patient.sample_file:
                    ds = pydicom.dcmread(str(patient.sample_file), force=True, stop_before_pixels=True)
                    if hasattr(ds, 'PatientName') and ds.PatientName:
                        raw_name = str(ds.PatientName)
                        new_name = self._decode_name(raw_name)
                        if new_name:
                            patient.patient_name = new_name
            except:
                pass
            
            result.append(patient)
        
        return result
    
    def stop(self):
        """Останавливает сканирование"""
        self._stop_scanning = True
        print("⏹️ Остановка сканирования...")
        
    def scan(self, progress_callback=None, patient_callback=None) -> ScanResult:
        """
        Сканирует папку и возвращает результат
        """
        if not self.root_path.exists():
            raise FileNotFoundError(f"Папка не найдена: {self.root_path}")
        
        self.scanned_files = 0
        self.found_dicom = 0
        self._stop_scanning = False
        
        patients = []
        errors = []
        
        if progress_callback:
            progress_callback("🔍 Поиск DICOM-файлов...", 0)
        
        all_dicom_files = self._find_all_dicom_files(self.root_path, progress_callback)
        
        if self._stop_scanning:
            return ScanResult(
                root_path=self.root_path,
                patients=[],
                scan_time=datetime.now(),
                errors=["Сканирование прервано пользователем"]
            )
        
        if not all_dicom_files:
            return ScanResult(
                root_path=self.root_path,
                patients=[],
                scan_time=datetime.now(),
                errors=["DICOM-файлы не найдены"]
            )
        
        # ← УБИРАЕМ ЛИШНЕЕ ОБНОВЛЕНИЕ ЗДЕСЬ
        # if progress_callback:
        #     progress_callback(f"📊 Найдено {len(all_dicom_files)} DICOM-файлов", 20)
        
        # Группировка сама покажет прогресс от 20% до ~30%
        patient_groups = self._group_files_by_patient_id(all_dicom_files, progress_callback)
        
        if self._stop_scanning:
            return ScanResult(
                root_path=self.root_path,
                patients=[],
                scan_time=datetime.now(),
                errors=["Сканирование прервано пользователем"]
            )
        
        if progress_callback:
            progress_callback(f"👤 Найдено {len(patient_groups)} пациентов", 30)
        
        total_groups = len(patient_groups)
        processed = 0
        
        for patient_id, files in patient_groups.items():
            if self._stop_scanning:
                break
                
            try:
                patient_info = self._process_patient_group(patient_id, files)
                if patient_info:
                    patients.append(patient_info)
                    processed += 1
                    
                    if patient_callback:
                        patient_callback(patient_info)
                    
                    if processed % 5 == 0 and progress_callback:
                        percent = 30 + int((processed / total_groups) * 60)
                        progress_callback(
                            f"👤 Пациентов: {processed}/{total_groups}",
                            percent
                        )
            except Exception as e:
                error_msg = f"Ошибка обработки пациента {patient_id}: {e}"
                errors.append(error_msg)
                print(f"⚠️ {error_msg}")
        
        if self._stop_scanning:
            return ScanResult(
                root_path=self.root_path,
                patients=patients,
                scan_time=datetime.now(),
                errors=[f"Сканирование прервано (найдено {len(patients)} пациентов)"]
            )
        
        patients.sort(key=lambda x: x.patient_name)
        
        if progress_callback:
            progress_callback("✅ Готово!", 100)
        
        return ScanResult(
            root_path=self.root_path,
            patients=patients,
            scan_time=datetime.now(),
            errors=errors
        )

    def _get_eta(self, processed: int, total: int) -> str:
        """Рассчитывает примерное оставшееся время"""
        if processed == 0 or total == 0:
            return "расчёт..."
        
        # Время на одного пациента
        time_per_patient = 0.5  # примерное время в секундах на одного пациента
        
        remaining = (total - processed) * time_per_patient
        
        if remaining > 60:
            return f"~{int(remaining/60)} мин"
        elif remaining > 10:
            return f"~{int(remaining)} сек"
        else:
            return f"~{int(remaining)} сек"
    
    def _find_all_dicom_files(self, folder: Path, progress_callback=None) -> List[Path]:
        """
        Рекурсивно находит все DICOM-файлы в папке и подпапках
        """
        dicom_files = []
        total_processed = 0
        last_update = 0
        
        try:
            for file_path in folder.rglob('*'):
                if self._stop_scanning:
                    break
                    
                self.scanned_files += 1
                total_processed += 1
                
                # Проверяем, является ли файл DICOM
                is_dicom = file_path.is_file() and file_path.suffix.lower() in self.supported_extensions
                if is_dicom:
                    dicom_files.append(file_path)
                    self.found_dicom += 1
                
                # Обновляем прогресс каждые 200 файлов
                if total_processed - last_update >= 200:
                    if progress_callback:
                        progress_callback(
                            f"📁 Найдено DICOM: {self.found_dicom} из {self.scanned_files} файлов",
                            0
                        )
                    last_update = total_processed
                
                if self.scanned_files > self.max_files:
                    print(f"⚠️ Достигнут лимит файлов ({self.max_files})")
                    break
                        
        except PermissionError:
            pass
        except Exception as e:
            print(f"⚠️ Ошибка при сканировании: {e}")
        
        if progress_callback:
            progress_callback(f"📁 Найдено DICOM: {self.found_dicom} из {self.scanned_files} файлов", 20)
        
        return dicom_files