src/description_cache_manager.py - master - IvanMysin/Topics

src/description_cache_manager.py
75 строк3 KB
ivan
Add detail saving of results, summarization cache. Some function need be tested.
16 окт 2025, 15:37
16 окт 2025, 15:37660cacb
import json
import os
from typing import Dict, List
from config import RESULTS_DIR


class DescriptionCacheManager:
    """Менеджер для работы с кэшем описаний кластеров"""

    def __init__(self):
        self.cache_file = RESULTS_DIR / "cluster_descriptions_cache.json"
        self.cache = self._load_cache()

    def _load_cache(self) -> Dict:
        """Загружает кэш описаний"""
        if self.cache_file.exists():
            try:
                with open(self.cache_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                print(f"⚠️ Не удалось загрузить кэш описаний: {e}")
        return {}

    def save_cache(self):
        """Сохраняет кэш описаний"""
        try:
            with open(self.cache_file, 'w', encoding='utf-8') as f:
                json.dump(self.cache, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"❌ Не удалось сохранить кэш описаний: {e}")

    def get_description(self, cluster_signature: str) -> str:
        """Возвращает описание по сигнатуре кластера"""
        return self.cache.get(cluster_signature)

    def set_description(self, cluster_signature: str, description: str):
        """Сохраняет описание в кэш"""
        self.cache[cluster_signature] = description
        self.save_cache()

    def clear_cache(self):
        """Очищает кэш описаний"""
        self.cache = {}
        self.save_cache()
        print("✅ Кэш описаний очищен")

    def get_cache_stats(self) -> Dict:
        """Возвращает статистику кэша"""
        return {
            "total_entries": len(self.cache),
            "cache_file": str(self.cache_file),
            "cache_size_kb": os.path.getsize(self.cache_file) / 1024 if self.cache_file.exists() else 0
        }

    def find_similar_clusters(self, topic_words: List, threshold: int = 3) -> List[str]:
        """Находит похожие кластеры в кэше по ключевым словам"""
        similar = []

        for signature, description in self.cache.items():
            # Извлекаем ключевые слова из сигнатуры
            words_part = signature.split('_')[0]
            cached_words = words_part.split('|')

            # Считаем совпадения
            matches = len(set(topic_words) & set(cached_words))

            if matches >= threshold:
                similar.append({
                    'signature': signature,
                    'matches': matches,
                    'description': description
                })

        # Сортируем по количеству совпадений
        return sorted(similar, key=lambda x: x['matches'], reverse=True)