76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
from zyrarbot.music.history_entry import HistoryEntry
|
|
from zyrarbot.storage.json_file_store import JsonFileStore
|
|
|
|
|
|
class HistoryRepository:
|
|
"""Historia odtwarzania w osobnym pliku dla każdego serwera."""
|
|
|
|
def __init__(self, store: JsonFileStore, limit: int = 50):
|
|
self._store = store
|
|
self._limit = limit
|
|
self._cache: dict[int, list[HistoryEntry]] = {}
|
|
self._guild_stores: dict[int, JsonFileStore] = {}
|
|
self._legacy_migrated = False
|
|
|
|
def _guild_store(self, guild_id: int) -> JsonFileStore:
|
|
if guild_id not in self._guild_stores:
|
|
directory = os.path.join(os.path.dirname(self._store.path), "history")
|
|
path = os.path.join(directory, f"{guild_id}.json")
|
|
self._guild_stores[guild_id] = JsonFileStore(path)
|
|
return self._guild_stores[guild_id]
|
|
|
|
def migrate_legacy(self, first_guild_id: int) -> None:
|
|
"""Move the old shared history file into one file per guild."""
|
|
if self._legacy_migrated:
|
|
return
|
|
|
|
legacy_data = self._store.read()
|
|
for key, entries in legacy_data.items():
|
|
if key == "history" and isinstance(entries, list):
|
|
self._write_entries(first_guild_id, entries)
|
|
elif key.isdigit() and isinstance(entries, list):
|
|
self._write_entries(int(key), entries)
|
|
self._legacy_migrated = True
|
|
|
|
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._guild_store(guild_id).get("entries", [])
|
|
raw_entries = raw_entries 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._guild_store(guild_id).set(
|
|
"entries", [entry.to_dict() for entry in self._cache.get(guild_id, [])]
|
|
)
|
|
|
|
def _write_entries(self, guild_id: int, entries: list[dict]) -> None:
|
|
self._guild_store(guild_id).set("entries", entries[: self._limit])
|