99 lines
4.4 KiB
Python
99 lines
4.4 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class MessengerDBManager:
|
|
def __init__(self, base_path: str = "data") -> None:
|
|
self.base_path = Path(base_path)
|
|
self._initialize_structure()
|
|
|
|
def _initialize_structure(self) -> None:
|
|
(self.base_path / "dms").mkdir(parents=True, exist_ok=True)
|
|
(self.base_path / "servers").mkdir(parents=True, exist_ok=True)
|
|
(self.base_path / "users").mkdir(parents=True, exist_ok=True)
|
|
|
|
self._ensure_file(self.base_path / "users" / "tags.json", {})
|
|
self._ensure_file(self.base_path / "users" / "users.json", {})
|
|
|
|
def _ensure_file(self, path: Path, default_content: Any) -> None:
|
|
if not path.exists():
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._write_json(path, default_content)
|
|
|
|
def _read_json(self, path: Path) -> Any:
|
|
if not path.exists():
|
|
return None
|
|
with path.open("r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
def _write_json(self, path: Path, data: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=4, ensure_ascii=False)
|
|
|
|
def _get_dm_filename(self, user1: str, user2: str) -> str:
|
|
users = sorted([user1, user2])
|
|
return f"{users[0]}-{users[1]}_encrypted.json"
|
|
|
|
def read_dm(self, user1: str, user2: str) -> Optional[List[Dict[str, Any]]]:
|
|
path = self.base_path / "dms" / self._get_dm_filename(user1, user2)
|
|
return self._read_json(path)
|
|
|
|
def write_dm(self, user1: str, user2: str, messages: List[Dict[str, Any]]) -> None:
|
|
path = self.base_path / "dms" / self._get_dm_filename(user1, user2)
|
|
self._write_json(path, messages)
|
|
|
|
def read_users(self) -> Dict[str, Any]:
|
|
return self._read_json(self.base_path / "users" / "users.json") or {}
|
|
|
|
def write_users(self, data: Dict[str, Any]) -> None:
|
|
self._write_json(self.base_path / "users" / "users.json", data)
|
|
|
|
def read_tags(self) -> Dict[str, Any]:
|
|
return self._read_json(self.base_path / "users" / "tags.json") or {}
|
|
|
|
def write_tags(self, data: Dict[str, Any]) -> None:
|
|
self._write_json(self.base_path / "users" / "tags.json", data)
|
|
|
|
def _get_server_path(self, server_id: str) -> Path:
|
|
return self.base_path / "servers" / str(server_id)
|
|
|
|
def init_server(self, server_id: str) -> None:
|
|
server_path = self._get_server_path(server_id)
|
|
self._ensure_file(server_path / "banned_ppl.json", [])
|
|
self._ensure_file(server_path / "members.json", [])
|
|
self._ensure_file(server_path / "roles.json", {})
|
|
self._ensure_file(server_path / "server_settings.json", {})
|
|
|
|
def read_server_file(self, server_id: str, filename: str) -> Any:
|
|
path = self._get_server_path(server_id) / f"{filename}.json"
|
|
return self._read_json(path)
|
|
|
|
def write_server_file(self, server_id: str, filename: str, data: Any) -> None:
|
|
path = self._get_server_path(server_id) / f"{filename}.json"
|
|
self._write_json(path, data)
|
|
|
|
def _get_channel_path(self, server_id: str, channel_id: str) -> Path:
|
|
return self._get_server_path(server_id) / "channels" / str(channel_id)
|
|
|
|
def init_channel(self, server_id: str, channel_id: str) -> None:
|
|
channel_path = self._get_channel_path(server_id, channel_id)
|
|
self._ensure_file(channel_path / "messages.json", [])
|
|
self._ensure_file(channel_path / "channel_config.json", {})
|
|
|
|
def read_channel_messages(self, server_id: str, channel_id: str) -> Optional[List[Dict[str, Any]]]:
|
|
path = self._get_channel_path(server_id, channel_id) / "messages.json"
|
|
return self._read_json(path)
|
|
|
|
def write_channel_messages(self, server_id: str, channel_id: str, messages: List[Dict[str, Any]]) -> None:
|
|
path = self._get_channel_path(server_id, channel_id) / "messages.json"
|
|
self._write_json(path, messages)
|
|
|
|
def read_channel_config(self, server_id: str, channel_id: str) -> Optional[Dict[str, Any]]:
|
|
path = self._get_channel_path(server_id, channel_id) / "channel_config.json"
|
|
return self._read_json(path)
|
|
|
|
def write_channel_config(self, server_id: str, channel_id: str, config: Dict[str, Any]) -> None:
|
|
path = self._get_channel_path(server_id, channel_id) / "channel_config.json"
|
|
self._write_json(path, config) |