lgooseandClaude Bot
e59b4ad9f1
Dostęp do danych trwałych schodzi za trzy repozytoria, zamiast być rozsmarowany po metodach coga i endpointach: - GuildConfigRepository - kanały powiadomień per serwer i domyślna liczba odtworzeń. - HistoryRepository - historia per serwer z buforem w pamięci i limitem wpisów. Zasada odsiewania duplikatów (zgodny tytuł albo adres) siedzi teraz w HistoryEntry.is_same_track_as zamiast w warunku wbudowanym w listę składaną. - UploadsRepository - katalog uploadów. Każda nazwa przechodzi przez os.path.basename w jednym miejscu, co zamyka temat ścieżek względnych przychodzących z API. Wpisy historii i pliki opisują struktury HistoryEntry oraz UploadFileInfo zamiast luźnych słowników z kluczami sprawdzanymi przez .get(). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
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, [])])
|