Bot become a Community Guard & Post send manager
added: dictionary support for censore message/user management with dict triggers
This commit is contained in:
42
app/moderation/cache.py
Normal file
42
app/moderation/cache.py
Normal file
@@ -0,0 +1,42 @@
|
||||
import time
|
||||
from typing import Any, Optional, Dict
|
||||
|
||||
class TTLCache:
|
||||
"""
|
||||
Простой in-memory кеш с TTL и грубой политикой вытеснения (удаляем самый старый ключ).
|
||||
Не потокобезопасен — вызывайте из одного потока/процесса.
|
||||
"""
|
||||
def __init__(self, ttl_seconds: int = 60, max_size: int = 1024):
|
||||
self.ttl = ttl_seconds
|
||||
self.max = max_size
|
||||
self._data: Dict[Any, Any] = {}
|
||||
self._ts: Dict[Any, float] = {}
|
||||
|
||||
def get(self, key: Any) -> Optional[Any]:
|
||||
now = time.time()
|
||||
ts = self._ts.get(key)
|
||||
if ts is None:
|
||||
return None
|
||||
if now - ts > self.ttl:
|
||||
# истёк TTL
|
||||
self._data.pop(key, None)
|
||||
self._ts.pop(key, None)
|
||||
return None
|
||||
return self._data.get(key)
|
||||
|
||||
def set(self, key: Any, value: Any):
|
||||
# простое вытеснение — удаляем самый старый
|
||||
if len(self._data) >= self.max and self._ts:
|
||||
oldest_key = min(self._ts.items(), key=lambda kv: kv[1])[0]
|
||||
self._data.pop(oldest_key, None)
|
||||
self._ts.pop(oldest_key, None)
|
||||
self._data[key] = value
|
||||
self._ts[key] = time.time()
|
||||
|
||||
def invalidate(self, key: Any):
|
||||
self._data.pop(key, None)
|
||||
self._ts.pop(key, None)
|
||||
|
||||
def clear(self):
|
||||
self._data.clear()
|
||||
self._ts.clear()
|
||||
Reference in New Issue
Block a user