From 909908c7895f8db29c87e28979bafed3d6f6650a Mon Sep 17 00:00:00 2001 From: Whyverum Date: Mon, 8 Dec 2025 16:47:47 +0700 Subject: [PATCH] =?UTF-8?q?=D0=9C=D0=BE=D0=B4=D1=83=D0=BB=D1=8C=20=D0=BD?= =?UTF-8?q?=D0=B0=D0=BF=D0=BE=D0=BC=D0=B8=D0=BD=D0=B0=D0=BD=D0=B8=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bot/storage.py | 141 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 141 insertions(+) create mode 100644 bot/storage.py diff --git a/bot/storage.py b/bot/storage.py new file mode 100644 index 0000000..23a1330 --- /dev/null +++ b/bot/storage.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List + +from configs import settings + + +@dataclass +class Reminder: + """ + Модель напоминания. + + Attributes: + time: Unix-время, когда нужно отправить напоминание. + channel_id: ID текстового канала. + user_mention: mention пользователя. + text: Текст напоминания. + """ + time: float + channel_id: int + user_mention: str + text: str + + +@dataclass +class Storage: + """ + Класс для работы с локальными JSON‑хранилищами: + предупреждения, напоминания, чёрный список. + """ + + warnings_file: Path = settings.WARNINGS_FILE + reminders_file: Path = settings.REMINDERS_FILE + blacklist_file: Path = settings.BLACKLIST_FILE + + reminders: List[Reminder] = field(default_factory=list) + user_warnings: Dict[str, List[Dict[str, Any]]] = field(default_factory=dict) + blacklist: List[str] = field(default_factory=list) + + def load_all(self) -> None: + """ + Загрузить все данные из файлов. + """ + self.load_warnings() + self.load_reminders() + self.load_blacklist() + + def load_warnings(self) -> None: + """ + Загрузить предупреждения пользователей из JSON‑файла. + """ + if self.warnings_file.is_file(): + try: + data = json.loads(self.warnings_file.read_text(encoding="utf-8")) + if isinstance(data, dict): + self.user_warnings = data + else: + self.user_warnings = {} + except json.JSONDecodeError: + self.user_warnings = {} + + def save_warnings(self) -> None: + """ + Сохранить предупреждения пользователей в JSON‑файл. + """ + self.warnings_file.write_text( + json.dumps(self.user_warnings, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + def load_reminders(self) -> None: + """ + Загрузить напоминания из JSON‑файла. + """ + self.reminders.clear() + if self.reminders_file.is_file(): + try: + data = json.loads(self.reminders_file.read_text(encoding="utf-8")) + if isinstance(data, list): + for item in data: + try: + self.reminders.append( + Reminder( + time=float(item["time"]), + channel_id=int(item["channel_id"]), + user_mention=str(item["user_mention"]), + text=str(item["text"]), + ) + ) + except (KeyError, ValueError, TypeError): + continue + except json.JSONDecodeError: + self.reminders = [] + + def save_reminders(self) -> None: + """ + Сохранить напоминания в JSON‑файл. + """ + data: List[Dict[str, Any]] = [ + { + "time": r.time, + "channel_id": r.channel_id, + "user_mention": r.user_mention, + "text": r.text, + } + for r in self.reminders + ] + self.reminders_file.write_text( + json.dumps(data, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + def load_blacklist(self) -> None: + """ + Загрузить чёрный список слов из JSON‑файла. + """ + self.blacklist.clear() + if self.blacklist_file.is_file(): + try: + data = json.loads(self.blacklist_file.read_text(encoding="utf-8")) + if isinstance(data, list): + self.blacklist = [str(w).lower() for w in data] + else: + self.blacklist = [] + except Exception: + self.blacklist = [] + + def save_blacklist(self) -> None: + """ + Сохранить чёрный список слов в JSON‑файл. + """ + self.blacklist_file.write_text( + json.dumps(self.blacklist, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + +storage: Storage = Storage()