diff --git a/zyrarbot/music/__init__.py b/zyrarbot/music/__init__.py new file mode 100644 index 0000000..2287f25 --- /dev/null +++ b/zyrarbot/music/__init__.py @@ -0,0 +1 @@ +"""Domena odtwarzania: stan kolejki, tryby zapętlania i silnik odtwarzacza.""" diff --git a/zyrarbot/music/history_entry.py b/zyrarbot/music/history_entry.py new file mode 100644 index 0000000..d573e4f --- /dev/null +++ b/zyrarbot/music/history_entry.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class HistoryEntry: + """Pojedynczy wpis historii odtwarzania zapisywany w history.json.""" + + id: str + title: str + url: str | None = None + duration: float | None = None + is_local: bool = False + filepath: str | None = None + + def is_same_track_as(self, other: HistoryEntry) -> bool: + """Wpisy uznajemy za ten sam utwór przy zgodnym tytule albo adresie.""" + return self.title == other.title or bool(other.url and self.url == other.url) + + @property + def replay_source(self) -> str | None: + """Ścieżka lub adres, z którego wpis można odtworzyć ponownie.""" + return (self.filepath or self.url) if self.is_local else self.url + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "title": self.title, + "url": self.url, + "duration": self.duration, + "is_local": self.is_local, + "filepath": self.filepath, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> HistoryEntry: + return cls( + id=str(data.get("id", "")), + title=data.get("title", ""), + url=data.get("url"), + duration=data.get("duration"), + is_local=bool(data.get("is_local", False)), + filepath=data.get("filepath"), + ) diff --git a/zyrarbot/storage/guild_config_repository.py b/zyrarbot/storage/guild_config_repository.py new file mode 100644 index 0000000..9a57072 --- /dev/null +++ b/zyrarbot/storage/guild_config_repository.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import logging + +from zyrarbot.storage.json_file_store import JsonFileStore + +logger = logging.getLogger(__name__) + + +class GuildConfigRepository: + """Dostęp do config.json: kanały powiadomień per serwer + globalne ustawienia.""" + + DEFAULT_PLAYS_KEY = "default_plays" + + def __init__(self, store: JsonFileStore, fallback_plays: int = 5): + self._store = store + self._fallback_plays = fallback_plays + + def get_notification_channel_id(self, guild_id: int) -> int | None: + value = self._store.get(str(guild_id)) + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + logger.error("Nieprawidłowe ID kanału w konfiguracji serwera %s: %r", guild_id, value) + return None + + def set_notification_channel_id(self, guild_id: int, channel_id: int | None) -> None: + self._store.set(str(guild_id), channel_id) + + def get_default_plays(self) -> int: + value = self._store.get(self.DEFAULT_PLAYS_KEY) + if value is None: + return self._fallback_plays + try: + return int(value) + except (TypeError, ValueError): + logger.error("Nieprawidłowa wartość %s: %r", self.DEFAULT_PLAYS_KEY, value) + return self._fallback_plays + + def set_default_plays(self, value: int) -> None: + self._store.set(self.DEFAULT_PLAYS_KEY, int(value)) diff --git a/zyrarbot/storage/history_repository.py b/zyrarbot/storage/history_repository.py new file mode 100644 index 0000000..8cc2236 --- /dev/null +++ b/zyrarbot/storage/history_repository.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from zyrarbot.music.history_entry import HistoryEntry +from zyrarbot.storage.json_file_store import JsonFileStore + + +class HistoryRepository: + """Historia odtwarzania per serwer, buforowana w pamięci i utrwalana w history.json.""" + + def __init__(self, store: JsonFileStore, limit: int = 50): + self._store = store + self._limit = limit + self._cache: dict[int, list[HistoryEntry]] = {} + + def list(self, guild_id: int) -> list[HistoryEntry]: + if guild_id not in self._cache: + self._cache[guild_id] = self._load(guild_id) + return self._cache[guild_id] + + def add(self, guild_id: int, entry: HistoryEntry) -> None: + if not entry.title: + return + entries = [item for item in self.list(guild_id) if not item.is_same_track_as(entry)] + entries.insert(0, entry) + self._cache[guild_id] = entries[: self._limit] + self._persist(guild_id) + + def find(self, guild_id: int, entry_id: str) -> HistoryEntry | None: + return next((item for item in self.list(guild_id) if item.id == entry_id), None) + + def _load(self, guild_id: int) -> list[HistoryEntry]: + raw_entries = self._store.get(str(guild_id), []) or [] + entries: list[HistoryEntry] = [] + seen: set[tuple[str, str | None]] = set() + for raw in raw_entries: + entry = HistoryEntry.from_dict(raw) + key = (entry.title, entry.url) + if key in seen: + continue + seen.add(key) + entries.append(entry) + return entries + + def _persist(self, guild_id: int) -> None: + self._store.set(str(guild_id), [entry.to_dict() for entry in self._cache.get(guild_id, [])]) diff --git a/zyrarbot/storage/upload_file_info.py b/zyrarbot/storage/upload_file_info.py new file mode 100644 index 0000000..a686b13 --- /dev/null +++ b/zyrarbot/storage/upload_file_info.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class UploadFileInfo: + """Metadane pliku znajdującego się w serwerowej bazie plików.""" + + filename: str + size_mb: float + mtime: float + + def to_dict(self) -> dict[str, Any]: + return {"filename": self.filename, "size_mb": self.size_mb, "mtime": self.mtime} diff --git a/zyrarbot/storage/uploads_repository.py b/zyrarbot/storage/uploads_repository.py new file mode 100644 index 0000000..8c259d3 --- /dev/null +++ b/zyrarbot/storage/uploads_repository.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import os +import shutil +from typing import BinaryIO + +from zyrarbot.storage.upload_file_info import UploadFileInfo + + +class UploadsRepository: + """Katalog z plikami wgranymi na serwer; nazwy zawsze sprowadzane do basename.""" + + def __init__(self, directory: str): + self._directory = directory + + @property + def directory(self) -> str: + return self._directory + + def resolve(self, filename: str) -> str: + return os.path.join(self._directory, os.path.basename(filename)) + + def exists(self, filename: str) -> bool: + return os.path.isfile(self.resolve(filename)) + + def list(self) -> list[UploadFileInfo]: + os.makedirs(self._directory, exist_ok=True) + files = [] + for name in os.listdir(self._directory): + path = os.path.join(self._directory, name) + if not os.path.isfile(path) or name.startswith("."): + continue + files.append( + UploadFileInfo( + filename=name, + size_mb=round(os.path.getsize(path) / (1024 * 1024), 2), + mtime=os.path.getmtime(path), + ) + ) + files.sort(key=lambda item: item.mtime, reverse=True) + return files + + def search(self, query: str, limit: int = 25) -> list[str]: + needle = query.lower() + return [item.filename for item in self.list() if needle in item.filename.lower()][:limit] + + def save(self, filename: str, source: BinaryIO) -> str: + os.makedirs(self._directory, exist_ok=True) + name = os.path.basename(filename) + with open(self.resolve(name), "wb") as target: + shutil.copyfileobj(source, target) + return name + + def delete(self, filename: str) -> bool: + path = self.resolve(filename) + if not os.path.isfile(path): + return False + os.remove(path) + return True + + def touch(self, filename: str) -> None: + path = self.resolve(filename) + if os.path.isfile(path): + os.utime(path, None)