import time import threading from typing import Dict, Tuple, Any class TranscriptionDict: def __init__(self, max_size=1000): self.max_size = max_size self._dict: Dict[str, Tuple[str, float]] = {} # Start a thread to clean expired items every 3 minutes self._clean_thread = threading.Thread(target=self._clean_expired_items_periodically) self._clean_thread.daemon = True self._clean_thread.start() def __len__(self): return len(self._dict) def get(self, key, default=None): return self._dict.get(key, {'value': default})['value'] def set(self, key, value): self._dict[key] = {'value': value, 'timestamp': time.time()} if len(self._dict) > self.max_size: self._clean_expired_items() def __contains__(self, key): return key in self._dict def _clean_expired_items(self): now = time.time() expired_items = [k for k, v in self._dict.items() if now - v['timestamp'] >= 180] # 180 seconds = 3 minutes for key in expired_items: del self._dict[key] # Ensure the cache size does not exceed max_size while len(self._dict) > self.max_size: oldest_key = min(self._dict, key=lambda k: self._dict[k]['timestamp']) del self._dict[oldest_key] print(f"Cleaned {len(expired_items)} expired items from the cache") def _clean_expired_items_periodically(self): while True: time.sleep(180) # Check for expired items every 3 minutes self._clean_expired_items()