from typing import BinaryIO, Protocol from abc import abstractmethod from glob import glob import json from pathlib import Path from . import SoloTool import requests class SessionManager(Protocol): @abstractmethod def getSessions(self) -> list[str]: raise NotImplementedError @abstractmethod def loadSession(self, key: str, player=None) -> SoloTool: raise NotImplementedError @abstractmethod def saveSession(self, soloTool: SoloTool, key: str) -> None: raise NotImplementedError @abstractmethod def addSong(self, name: str, content: BinaryIO) -> None: raise NotImplementedError @staticmethod def _dict2st(session: dict, songPool: str, player) -> SoloTool: st = SoloTool(player=player) for i, entry in enumerate(session): songPath = entry["path"] keyPoints = entry["key_points"] st.addSong(songPath) st._keyPoints[i] = keyPoints return st @staticmethod def _st2dict(soloTool: SoloTool) -> dict: session = [] for i, song in enumerate(soloTool.songs): entry = { "path": song, "key_points" : soloTool._keyPoints[i] } session.append(entry) return session class _FileSystemSessionManager(SessionManager): def __init__(self, songPool: str, sessionPath: str): self._songPool = Path(songPool) self._sessionPath = Path(sessionPath) def getSessions(self) -> list[str]: return [Path(f).stem for f in glob(f"{self._sessionPath}/*.json")] def loadSession(self, key: str, player=None) -> SoloTool: with open(self._sessionPath / f"{key}.json", "r") as f: session = json.load(f) return SessionManager._dict2st(session, self._songPool, player) def saveSession(self, soloTool: SoloTool, key: str) -> None: session = SessionManager._st2dict(soloTool) with open(self._sessionPath / f"{key}.json", "w") as f: json.dump(session, f) def addSong(self, name: str, content: BinaryIO) -> None: from shutil import copyfileobj newSong = self._songPool / name with open(newSong, "wb") as f: copyfileobj(content, f) class _FileBrowserSessionManager(SessionManager): def __init__(self, songPoolUrl: str, sessionUrl: str): self._baseUrl = "https://files.0xf7.com" self._songPool = songPoolUrl self._username = "solo-tool" self._password = "mwC0ML8vLpJLPCLHKuxkiOxtIaE" self._apiKey = self._getApiKey() def getSessions(self) -> list[str]: url = f"{self._baseUrl}/api/resources/sessions" response = requests.get(url, headers={"X-Auth":self._apiKey}) response.raise_for_status() return [item["name"][0:-5] for item in response.json()["items"] if item["extension"] == ".json"] def loadSession(self, key: str, player=None) -> SoloTool: url = f"{self._baseUrl}/api/raw/sessions/{key}.json" response = requests.get(url, headers={"X-Auth":self._apiKey}) response.raise_for_status() return SessionManager._dict2st(json.loads(response.content), self._songPool, player=player) def saveSession(self, soloTool: SoloTool, key: str) -> None: pass def addSong(self, name: str, content: BinaryIO) -> None: pass def _getApiKey(self) -> str: response = requests.post(f"{self._baseUrl}/api/login", json={"username":self._username, "password":self._password}) return response.content def getSessionManager(songPool: str, sessionPath: str) -> SessionManager: from re import search match = search(r"^([a-z0-9]+://)", sessionPath) if not match or match.group(0) == "file://": return _FileSystemSessionManager(songPool, sessionPath) elif match.group(0) in ["http://", "https://"]: return _FileBrowserSessionManager(songPool, sessionPath)