cache_manager.py - master - O.S.Prog/DiScan

cache_manager.py
237 строк9 KB
O.S.Prog
Initial commit: DiScan v1.1.0
18 июл 2026, 15:51
18 июл 2026, 15:51238a6ec
# cache_manager.py
"""
Управление кэшированием результатов сканирования
"""

import json
import hashlib
import shutil
from pathlib import Path
from datetime import datetime
from typing import Optional, List, Dict, Any
from core.models import PatientInfo, ScanResult


class CacheManager:
    """Управление кэшем"""
    
    CACHE_DIR_NAME = ".dicom_cache"
    CACHE_FILE_NAME = "cache.json"
    SUPPORTED_EXTENSIONS = {'.dcm', '.dicom'}
    
    @staticmethod
    def get_cache_path(folder_path: Path) -> Path:
        """
        Возвращает путь к файлу кэша для указанной папки
        """
        cache_dir = folder_path / CacheManager.CACHE_DIR_NAME
        return cache_dir / CacheManager.CACHE_FILE_NAME
    
    @staticmethod
    def get_cache_key(folder_path: Path, encoding: str) -> str:
        """
        Генерирует ключ кэша на основе пути и кодировки
        """
        key_string = f"{folder_path}_{encoding}"
        return hashlib.md5(key_string.encode('utf-8')).hexdigest()
    
    @staticmethod
    def count_dicom_files(folder_path: Path) -> int:
        """
        Подсчитывает количество DICOM-файлов в папке (рекурсивно)
        """
        count = 0
        try:
            for file_path in folder_path.rglob('*'):
                if file_path.is_file() and file_path.suffix.lower() in CacheManager.SUPPORTED_EXTENSIONS:
                    count += 1
        except PermissionError:
            pass
        except Exception as e:
            print(f"⚠️ Ошибка подсчёта файлов: {e}")
        return count
    
    @staticmethod
    def is_valid(folder_path: Path, encoding: str) -> bool:
        """
        Проверяет, актуален ли кэш для указанной папки
        Сравнивает количество DICOM-файлов в папке с сохранённым в кэше
        """
        try:
            cache_path = CacheManager.get_cache_path(folder_path)
            if not cache_path.exists():
                return False
            
            with open(cache_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            
            # Проверяем кодировку
            if data.get("encoding") != encoding:
                return False
            
            # Проверяем путь
            if data.get("root_path") != str(folder_path):
                return False
            
            # Считаем текущие DICOM-файлы
            current_count = CacheManager.count_dicom_files(folder_path)
            
            # Считаем файлы из кэша (сначала пробуем из file_count, потом суммируем по пациентам)
            cached_count = data.get("file_count", 0)
            if cached_count == 0:
                cached_count = sum(p.get("total_files", 0) for p in data.get("patients", []))
            
            # Если количество файлов изменилось — кэш не актуален
            return current_count == cached_count
            
        except Exception as e:
            print(f"⚠️ Ошибка проверки кэша: {e}")
            return False
    
    @staticmethod
    def save(folder_path: Path, scan_result: ScanResult, encoding: str) -> bool:
        """
        Сохраняет результат сканирования в кэш
        """
        try:
            cache_path = CacheManager.get_cache_path(folder_path)
            cache_path.parent.mkdir(parents=True, exist_ok=True)
            
            # Преобразуем данные в словарь
            data = {
                "root_path": str(folder_path),
                "scan_time": scan_result.scan_time.isoformat(),
                "encoding": encoding,
                "patients": [],
                "errors": scan_result.errors,
                "file_count": CacheManager.count_dicom_files(folder_path)  # ← ДОБАВЛЯЕМ
            }
            
            for patient in scan_result.patients:
                patient_dict = {
                    "patient_id": patient.patient_id,
                    "patient_name": patient.patient_name,
                    "folder_path": str(patient.folder_path),
                    "study_date": patient.study_date,
                    "total_files": patient.total_files,
                    "image_count": patient.image_count,
                    "sample_file": str(patient.sample_file) if patient.sample_file else None,
                    "series_uids": list(patient.series_uids) if patient.series_uids else [],
                    "study_uid": patient.study_uid,
                    "error": patient.error
                }
                data["patients"].append(patient_dict)
            
            # Сохраняем в файл
            with open(cache_path, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            
            return True
            
        except Exception as e:
            print(f"⚠️ Ошибка сохранения кэша: {e}")
            return False
    
    @staticmethod
    def load(folder_path: Path, encoding: str) -> Optional[ScanResult]:
        """
        Загружает результат сканирования из кэша
        """
        try:
            cache_path = CacheManager.get_cache_path(folder_path)
            if not cache_path.exists():
                return None
            
            with open(cache_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            
            # Проверяем, что кэш соответствует текущей папке и кодировке
            if data.get("root_path") != str(folder_path):
                return None
            
            if data.get("encoding") != encoding:
                return None
            
            # Восстанавливаем объекты PatientInfo
            patients = []
            for patient_data in data.get("patients", []):
                patient = PatientInfo(
                    patient_id=patient_data["patient_id"],
                    patient_name=patient_data["patient_name"],
                    folder_path=Path(patient_data["folder_path"]),
                    study_date=patient_data["study_date"],
                    total_files=patient_data["total_files"],
                    image_count=patient_data["image_count"],
                    sample_file=Path(patient_data["sample_file"]) if patient_data.get("sample_file") else None,
                    series_uids=set(patient_data.get("series_uids", [])),
                    study_uid=patient_data.get("study_uid"),
                    error=patient_data.get("error")
                )
                patients.append(patient)
            
            scan_time = datetime.fromisoformat(data["scan_time"])
            
            return ScanResult(
                root_path=Path(data["root_path"]),
                patients=patients,
                scan_time=scan_time,
                errors=data.get("errors", [])
            )
            
        except Exception as e:
            print(f"⚠️ Ошибка загрузки кэша: {e}")
            return None
    
    @staticmethod
    def clear(folder_path: Path) -> bool:
        """
        Удаляет кэш для указанной папки
        """
        try:
            cache_dir = folder_path / CacheManager.CACHE_DIR_NAME
            if cache_dir.exists():
                shutil.rmtree(cache_dir)
            return True
        except Exception as e:
            print(f"⚠️ Ошибка удаления кэша: {e}")
            return False
    
    @staticmethod
    def exists(folder_path: Path, encoding: str) -> bool:
        """
        Проверяет, существует ли актуальный кэш
        """
        cache_path = CacheManager.get_cache_path(folder_path)
        if not cache_path.exists():
            return False
        
        try:
            with open(cache_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            return data.get("root_path") == str(folder_path) and data.get("encoding") == encoding
        except:
            return False
    
    @staticmethod
    def get_cache_size(folder_path: Path) -> str:
        """
        Возвращает размер кэша в удобном формате
        """
        try:
            cache_dir = folder_path / CacheManager.CACHE_DIR_NAME
            if not cache_dir.exists():
                return "0 B"
            
            total_size = 0
            for file_path in cache_dir.rglob('*'):
                if file_path.is_file():
                    total_size += file_path.stat().st_size
            
            for unit in ['B', 'KB', 'MB', 'GB']:
                if total_size < 1024.0:
                    return f"{total_size:.1f} {unit}"
                total_size /= 1024.0
            return f"{total_size:.1f} TB"
            
        except:
            return "Неизвестно"