aboutsummaryrefslogtreecommitdiffstats
path: root/solo-tool-project/src/solo_tool/session_manager.py
blob: 4f831baf85c2c3a156bd98db336de1bd2ab1fb3d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
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(songPool, 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)