94 lines
2.3 KiB
Python
94 lines
2.3 KiB
Python
import time
|
|
from typing import Dict
|
|
from services.runtime_state import get_state, set_state
|
|
|
|
# category -> unix timestamp until muted
|
|
|
|
|
|
def _mutes() -> Dict[str, float]:
|
|
return get_state().get("mutes", {})
|
|
|
|
|
|
def _save(mutes: Dict[str, float]):
|
|
set_state("mutes", mutes)
|
|
|
|
|
|
def _cleanup() -> None:
|
|
mutes = _mutes()
|
|
now = time.time()
|
|
expired = [k for k, until in mutes.items() if until <= now]
|
|
for k in expired:
|
|
mutes.pop(k, None)
|
|
_save(mutes)
|
|
|
|
|
|
def set_mute(category: str, seconds: int) -> float:
|
|
_cleanup()
|
|
mutes = _mutes()
|
|
until = time.time() + max(0, seconds)
|
|
mutes[category] = until
|
|
_save(mutes)
|
|
return until
|
|
|
|
|
|
def clear_mute(category: str) -> None:
|
|
mutes = _mutes()
|
|
mutes.pop(category, None)
|
|
_save(mutes)
|
|
|
|
|
|
def is_muted(category: str | None) -> bool:
|
|
if not category:
|
|
return False
|
|
_cleanup()
|
|
mutes = _mutes()
|
|
until = mutes.get(category)
|
|
if until is None:
|
|
return False
|
|
if until <= time.time():
|
|
mutes.pop(category, None)
|
|
_save(mutes)
|
|
return False
|
|
return True
|
|
|
|
|
|
def list_mutes() -> dict[str, int]:
|
|
_cleanup()
|
|
now = time.time()
|
|
mutes = _mutes()
|
|
return {k: int(until - now) for k, until in mutes.items()}
|
|
|
|
|
|
def is_auto_muted(cfg: dict, category: str | None) -> bool:
|
|
if not category:
|
|
return False
|
|
auto_list = cfg.get("alerts", {}).get("auto_mute", [])
|
|
if not isinstance(auto_list, list):
|
|
return False
|
|
now = time.localtime()
|
|
now_minutes = now.tm_hour * 60 + now.tm_min
|
|
for item in auto_list:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
cat = item.get("category")
|
|
if cat != category:
|
|
continue
|
|
start = item.get("start", "00:00")
|
|
end = item.get("end", "00:00")
|
|
try:
|
|
sh, sm = [int(x) for x in start.split(":")]
|
|
eh, em = [int(x) for x in end.split(":")]
|
|
except Exception:
|
|
continue
|
|
start_min = sh * 60 + sm
|
|
end_min = eh * 60 + em
|
|
if start_min == end_min:
|
|
continue
|
|
if start_min < end_min:
|
|
if start_min <= now_minutes < end_min:
|
|
return True
|
|
else:
|
|
if now_minutes >= start_min or now_minutes < end_min:
|
|
return True
|
|
return False
|