From a07ee3bd520d8dde7b69225345ef17788dee7151 Mon Sep 17 00:00:00 2001 From: JISAUAY Date: Thu, 4 Sep 2025 15:30:01 -0500 Subject: [PATCH] went back to working on the first verison, fixed queue updating on main screen + weird footer bug --- textual-take2/app.py | 3 +- textual.log | 34 +- SubSonic.py => textual1/SubSonic.py | 874 ++++++------- screens-test.py => textual1/app.py | 1084 ++++++++--------- main.tcss => textual1/main.tcss | 8 +- requirements.txt => textual1/requirements.txt | 8 +- 6 files changed, 1018 insertions(+), 993 deletions(-) rename SubSonic.py => textual1/SubSonic.py (96%) rename screens-test.py => textual1/app.py (83%) rename main.tcss => textual1/main.tcss (92%) rename requirements.txt => textual1/requirements.txt (95%) diff --git a/textual-take2/app.py b/textual-take2/app.py index b8d679e..43548c0 100644 --- a/textual-take2/app.py +++ b/textual-take2/app.py @@ -157,12 +157,13 @@ class MusicApp(App): if song_id: self.play_song(song_id) + # Needs to be in the main thread so pause / play work + # TODO @0x01fe put stream https call into thread? @work(exclusive=True) async def play_song(self, song_id: str) -> None: try: song = conn.getSong(song_id) now_playing = self.query_one(NowPlaying) - # self.call_from_thread(NowPlaying.update_song, now_playing, song) now_playing.update_song(song) stream_data = conn.stream(song_id) diff --git a/textual.log b/textual.log index 3459617..37d79eb 100644 --- a/textual.log +++ b/textual.log @@ -1,3 +1,31 @@ -[2025-09-03 13:34:47,213] DEBUG in proactor_events: Using proactor: IocpProactor -[2025-09-03 13:34:55,201] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 -[2025-09-03 13:34:55,423] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getArtists?u=pseudo&t=e920f5780d30ae16dc5e7f65087de8e4&s=LHHshoee&v=1.16.1&c=testing&f=json HTTP/1.1" 200 None +[2025-09-04 14:49:01,708] DEBUG in proactor_events: Using proactor: IocpProactor +[2025-09-04 14:49:04,142] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:04,379] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getArtists?u=pseudo&t=2b8f5742d39c788122e9175e065483e0&s=e4mfyt4T&v=1.16.1&c=testing&f=json HTTP/1.1" 200 None +[2025-09-04 14:49:05,321] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:05,494] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=def2a86e92d32523f99a0747c6741047&s=BJ3TK656&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=0 HTTP/1.1" 200 None +[2025-09-04 14:49:05,499] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:05,651] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=0d05edbddb083b938bdf2e0543c8e44e&s=q39xsQJN&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=50 HTTP/1.1" 200 None +[2025-09-04 14:49:05,654] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:05,822] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=0296cab8954310b2b6d1af418fab6744&s=1qbZbaat&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=100 HTTP/1.1" 200 None +[2025-09-04 14:49:05,828] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:05,978] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=3b405a5c09c5259c8b2db318be52a577&s=MV94hD8j&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=150 HTTP/1.1" 200 None +[2025-09-04 14:49:05,981] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:06,134] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=c7b992e3538ed9e92f158b825eb6c3c8&s=VTK0roQ0&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=200 HTTP/1.1" 200 None +[2025-09-04 14:49:06,137] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:06,353] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=64fe71d647a2d62db5be7b84ccbae798&s=dEoTBmsf&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=250 HTTP/1.1" 200 None +[2025-09-04 14:49:06,356] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:06,501] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=0665f656905f113e49c2c96ac73fb164&s=Zw9rzvYM&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=300 HTTP/1.1" 200 None +[2025-09-04 14:49:06,503] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:06,645] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getAlbumList?u=pseudo&t=a96115e891c5315e1ae6118407b6504e&s=9o6Jmnpf&v=1.16.1&c=testing&f=json&type=alphabeticalByName&size=50&offset=350 HTTP/1.1" 200 145 +[2025-09-04 14:49:08,083] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:08,408] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getStarred?u=pseudo&t=f7f7bc6d80eb625439aab5a8567dfc79&s=eHkyJQuw&v=1.16.1&c=testing&f=json HTTP/1.1" 200 None +[2025-09-04 14:49:12,809] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:13,026] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getPlaylists?u=pseudo&t=509133849b592141690226ff28d1ca76&s=UMJx5VbS&v=1.16.1&c=testing&f=json HTTP/1.1" 200 1348 +[2025-09-04 14:49:14,583] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:14,818] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/getPlaylist?u=pseudo&t=01048946d8acff208aeea45da130e0d6&s=aQgAJaQL&v=1.16.1&c=testing&f=json&id=e6e8ff27-2a60-4e33-8aa1-09586191166a HTTP/1.1" 200 None +[2025-09-04 14:49:16,421] DEBUG in screens-test: Playing song I'm in Your Mind by King Gizzard & the Lizard Wizard. +[2025-09-04 14:49:16,423] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:16,619] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/stream?u=pseudo&t=03ab421b3afbad3b4439c0274989b6b1&s=PbHKnSOb&v=1.16.1&c=testing&f=json&id=2dcd70c3954594c7ea60eb5afe38a476 HTTP/1.1" 200 3418948 +[2025-09-04 14:49:34,597] DEBUG in screens-test: Playing song I'm Not in Your Mind by King Gizzard & the Lizard Wizard. +[2025-09-04 14:49:34,598] DEBUG in connectionpool: Starting new HTTPS connection (1): music.0x01fe.net:443 +[2025-09-04 14:49:34,913] DEBUG in connectionpool: https://music.0x01fe.net:443 "GET /rest/stream?u=pseudo&t=8303d34b33fd11d14072307afcdfd724&s=WNJQhtAu&v=1.16.1&c=testing&f=json&id=4f3b5f7c7c146695db01ccfb36bdcb45 HTTP/1.1" 200 None diff --git a/SubSonic.py b/textual1/SubSonic.py similarity index 96% rename from SubSonic.py rename to textual1/SubSonic.py index 57878dc..8ec0a0b 100644 --- a/SubSonic.py +++ b/textual1/SubSonic.py @@ -1,437 +1,437 @@ -import time -import random -import string -import hashlib -import logging -import urllib.parse - -import requests - - -logger = logging.getLogger(__name__) - -class Playlist: - id: str - name: str - songCount: int - duration: str - owner: str - coverArtId: str - - def __init__(self, info: dict): - self.id = info['id'] - self.name = info['name'] - self.songCount = int(info['songCount']) - self.duration = time.strftime('%H:%M:%S', time.gmtime(int(info['duration']))) - self.owner = info['owner'] - self.coverArtId = info['coverArt'] - -class Album: - id: str - title: str - name: str - album: str - coverArtId: str - duration: int - artistId: str - songCount: int - playCount: int - year: int | None - duration: str - - def __init__(self, album_info: dict): - self.id = album_info['id'] - self.title = album_info['title'] - self.name = album_info['name'] - self.album = album_info['album'] - self.coverArtId = album_info['coverArt'] - self.duration = int(album_info['duration']) - self.artistId = album_info['artistId'] - self.songCount = int(album_info['songCount']) - self.playCount = int(album_info['playCount']) if 'playCount' in album_info else 0 - self.year = int(album_info['year']) if 'year' in album_info else None - self.duration = self._secondToMinutes(int(album_info['duration'])) - - def _secondToMinutes(self, s: int) -> str: - return f'{s // 60}:{s % 60 if s % 60 > 9 else "0" + str(s % 60)}' - -class Song: - id: str - title: str - album: str - artist: str - trackNumber: int - contentType: str - suffix: str - duration: int - bitRate: int - path: str - discNumber: int - albumId: str - artistId: str - comment: str - channelCount: int - samplingRate: int - playCount: int | None - - def __init__(self, info: dict | None = {}): - self.id = info['id'] if 'id' in info else "" - self.title = info['title'] if 'title' in info else "" - self.album = info['album'] if 'album' in info else "" - self.artist = info['artist'] if 'artist' in info else "" - self.trackNumber = int(info['track']) if 'track' in info else "" - self.contentType = info['contentType'] if 'contentType' in info else "" - self.suffix = info['suffix'] if 'suffix' in info else "" - self.duration = int(info['duration']) if 'duration' in info else "" - self.bitRate = int(info['bitRate']) if 'bitRate' in info else "" - self.path = info['path'] if 'path' in info else "" - self.discNumber = int(info['discNumber']) if 'discNumber' in info else "" - self.albumId = info['albumId'] if 'albumId' in info else "" - self.artistId = info['artistId'] if 'artistId' in info else "" - self.comment = info['comment'] if 'comment' in info else "" - self.channelCount = int(info['channelCount']) if 'channelCount' in info else "" - self.samplingRate = int(info['samplingRate']) if 'samplingRate' in info else "" - self.playCount = int(info['playCount']) if 'playCount' in info else "" - -class Artist: - id: str - name: str - albumCount: int - coverArt: str - artistImageUrl: str - - def __init__(self, id: str, name: str, albumCount: str | int, coverArt: str, artistImageUrl: str): - self.id = id - self.name = name - self.albumCount = int(albumCount) - self.coverArt = coverArt - self.artistImageUrl = artistImageUrl - - def from_dict(info: dict): - return Artist( - info['id'] if 'id' in info else None, - info['name'] if 'name' in info else None, - info['albumCount'] if 'albumCount' in info else None, - info['coverArt'] if 'coverArt' in info else None, - info['artistImageUrl'] if 'artistImageUrl' in info else None - ) - -class BadSubsonicResponse(Exception): - pass - -class SubpyConn: - username: str - password: str - base_url: str - salt: str - salt_len: int = 8 - version: str = "1.16.1" - client_name: str = "testing" - - def __init__(self, u: str, p: str, base_url: str, salt_len: int | None = 8): - self.username = u - self.password = p - self.base_url = base_url - - self.salt_len = salt_len - - def _saltGenerator(self) -> None: - self.salt = ''.join(random.SystemRandom().choice(string.ascii_letters + string.digits) for _ in range(self.salt_len)) - - def _hashPass(self) -> str: - self._saltGenerator() - return hashlib.md5((self.password + self.salt).encode('utf-8')).hexdigest() - - def _getBasePayload(self) -> dict: - return { - "u" : self.username, - "t" : self._hashPass(), - "s" : self.salt, - "v" : self.version, - "c" : self.client_name, - "f": "json" - } - - def _encodePayload(self, payload: dict) -> str: - return urllib.parse.urlencode(payload, quote_via=urllib.parse.quote_plus) - - def _getMusicDirectory(self, id: str): - url = self.base_url + '/rest/getMusicDirectory?' - - payload = self._getBasePayload() - payload['id'] = id - - return requests.get(url + self._encodePayload(payload)) - - def getArtistAlbums(self, artist_id: str) -> list[Album]: - response: requests.Response = self._getMusicDirectory(artist_id).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - albums: list[Album] = [] - for item in response['directory']['child']: - albums.append(Album(item)) - - return albums - - def getAlbumSongs(self, album_id: str) -> list[Song]: - response: requests.Response = self._getMusicDirectory(album_id).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - songs: list[Song] = [] - for item in response['directory']['child']: - songs.append(Song(item)) - - return songs - - def getArtists(self) -> list[Artist]: - url = self.base_url + '/rest/getArtists?' - - payload = self._getBasePayload() - - response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - artists: list[Artist] = [] - - for artist_index in response['artists']['index']: - for artist in artist_index['artist']: - artists.append(Artist.from_dict(artist)) - - return artists - - def getAlbums(self, size: int | None = 25, offset: int | None = 0) -> list[Album]: - url = self.base_url + '/rest/getAlbumList?' - - payload = self._getBasePayload() - payload['type'] = 'alphabeticalByName' - payload['size'] = size - payload['offset'] = offset - - r = requests.get(url + self._encodePayload(payload)) - json_r = r.json() - - albums: list[Album] = [] - - if 'album' not in json_r['subsonic-response']['albumList']: - return [] - - for album in json_r['subsonic-response']['albumList']['album']: - albums.append(Album(album)) - - return albums - - def getAllAlbums(self) -> list[Album]: - all_albums: list[Album] = [] - - offset = 0 - while albums := self.getAlbums(50, offset): - all_albums.extend(albums) - offset += 50 - - return all_albums - - def getPlaylists(self) -> list[Playlist]: - url = self.base_url + "/rest/getPlaylists?" - - payload = self._getBasePayload() - - r = requests.get(url + self._encodePayload(payload)) - - json_r = r.json() - - playlists: list[Playlist] = [] - for playlist in json_r['subsonic-response']['playlists']['playlist']: - playlists.append(Playlist(playlist)) - - return playlists - - def getPlaylist(self, playlist_id: str) -> list[Song]: - url = self.base_url + "/rest/getPlaylist?" - - payload = self._getBasePayload() - payload['id'] = playlist_id - - r = requests.get(url + self._encodePayload(payload)) - - json_r = r.json() - - songs: list[Song] = [] - for song in json_r['subsonic-response']['playlist']['entry']: - songs.append(Song(song)) - - return songs - - def getStarred(self) -> list[Album | Song | Artist]: - url = self.base_url + "/rest/getStarred?" - - payload = self._getBasePayload() - - r = requests.get(url + self._encodePayload(payload)) - json_r = r.json() - - starred: list[Album | Song | Artist] = [] - for album in json_r['subsonic-response']['starred']['artist']: - starred.append(Album(album)) - - for artist in json_r['subsonic-response']['starred']['artist']: - starred.append(Artist.from_dict(artist)) - - for song in json_r['subsonic-response']['starred']['song']: - starred.append(Song(song)) - - return starred - - def getStarredSongs(self) -> list[Song]: - url = self.base_url + "/rest/getStarred?" - - payload = self._getBasePayload() - - r = requests.get(url + self._encodePayload(payload)) - json_r = r.json() - - starred: list[Song] = [] - for song in json_r['subsonic-response']['starred']['song']: - starred.append(Song(song)) - - return starred - - # Returns a requests Response object, the data stream can be found in Response.content - def stream(self, song_id: str) -> requests.Response: - url = self.base_url + '/rest/stream?' - - payload = self._getBasePayload() - - payload['id'] = song_id - - return requests.get(url + self._encodePayload(payload), stream=True) - - # Saves cover art id - def getCoverArt(self, cover_art_id: str, path: str | None = None) -> None: - url = self.base_url + '/rest/getCoverArt?' - - payload = self._getBasePayload() - payload['id'] = cover_art_id - - r = requests.get(url + self._encodePayload(payload)) - - with open(path if path else 'cover.png', 'wb') as file: - file.write(r.content) - - def getCoverArtRaw(self, cover_art_id: str) -> bytes: - url = self.base_url + '/rest/getCoverArt?' - - payload = self._getBasePayload() - payload['id'] = cover_art_id - - r = requests.get(url + self._encodePayload(payload)) - - return r.content - - def getSong(self, song_id: str) -> Song: - url = self.base_url + '/rest/getSong?' - - payload = self._getBasePayload() - payload['id'] = song_id - - response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - return Song(response['song']) - - def star(self, id: str) -> None: - url = self.base_url + '/rest/star?' - - payload = self._getBasePayload() - payload['id'] = id - - response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - def unstar(self, id: str) -> None: - url = self.base_url + '/rest/unstar?' - - payload = self._getBasePayload() - payload['id'] = id - - response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - """ - https://opensubsonic.netlify.app/docs/endpoints/scrobble/ - - id: str - unique identifier of the media to scrobble - OPTIONAL time: int - posix time of when the song was listened to - OPTIONAL submission: bool - Whether this is a "submission" or a "now playing" notification. Default: True - """ - def scrobble(self, id: str, time: int | None = None, submission: bool | None = True) -> None: - url = self.base_url + '/rest/scrobble?' - - payload = self._getBasePayload() - payload['id'] = id - - if time: - payload['time'] = time - - payload['submission'] = submission - - response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - """ - https://opensubsonic.netlify.app/docs/endpoints/getrandomsongs/ - """ - def getRandomSongs(self, - size: int | None = 10, - genre: str | None = None, - fromYear: str | int | None = None, - toYear: str | int | None = None, - musicFolderId: str | int | None = None - ) -> list[Song]: - url = self.base_url + '/rest/getRandomSongs?' - - payload = self._getBasePayload() - - # The default being 10 is only to make the method match the internal docs - if size and size != 10: - payload['size'] = size - - if genre: - payload['genre'] = genre - - if fromYear: - payload['fromYear'] = fromYear - - if toYear: - payload['toYear'] = toYear - - if musicFolderId: - payload['musicFolderId'] = musicFolderId - - response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] - - if response['status'] != 'ok': - raise BadSubsonicResponse() - - random_songs: list[Song] = [] - - for song in response['randomSongs']['song']: - random_songs.append(Song(song)) - - return random_songs - - - +import time +import random +import string +import hashlib +import logging +import urllib.parse + +import requests + + +logger = logging.getLogger(__name__) + +class Playlist: + id: str + name: str + songCount: int + duration: str + owner: str + coverArtId: str + + def __init__(self, info: dict): + self.id = info['id'] + self.name = info['name'] + self.songCount = int(info['songCount']) + self.duration = time.strftime('%H:%M:%S', time.gmtime(int(info['duration']))) + self.owner = info['owner'] + self.coverArtId = info['coverArt'] + +class Album: + id: str + title: str + name: str + album: str + coverArtId: str + duration: int + artistId: str + songCount: int + playCount: int + year: int | None + duration: str + + def __init__(self, album_info: dict): + self.id = album_info['id'] + self.title = album_info['title'] + self.name = album_info['name'] + self.album = album_info['album'] + self.coverArtId = album_info['coverArt'] + self.duration = int(album_info['duration']) + self.artistId = album_info['artistId'] + self.songCount = int(album_info['songCount']) + self.playCount = int(album_info['playCount']) if 'playCount' in album_info else 0 + self.year = int(album_info['year']) if 'year' in album_info else None + self.duration = self._secondToMinutes(int(album_info['duration'])) + + def _secondToMinutes(self, s: int) -> str: + return f'{s // 60}:{s % 60 if s % 60 > 9 else "0" + str(s % 60)}' + +class Song: + id: str + title: str + album: str + artist: str + trackNumber: int + contentType: str + suffix: str + duration: int + bitRate: int + path: str + discNumber: int + albumId: str + artistId: str + comment: str + channelCount: int + samplingRate: int + playCount: int | None + + def __init__(self, info: dict | None = {}): + self.id = info['id'] if 'id' in info else "" + self.title = info['title'] if 'title' in info else "" + self.album = info['album'] if 'album' in info else "" + self.artist = info['artist'] if 'artist' in info else "" + self.trackNumber = int(info['track']) if 'track' in info else "" + self.contentType = info['contentType'] if 'contentType' in info else "" + self.suffix = info['suffix'] if 'suffix' in info else "" + self.duration = int(info['duration']) if 'duration' in info else "" + self.bitRate = int(info['bitRate']) if 'bitRate' in info else "" + self.path = info['path'] if 'path' in info else "" + self.discNumber = int(info['discNumber']) if 'discNumber' in info else "" + self.albumId = info['albumId'] if 'albumId' in info else "" + self.artistId = info['artistId'] if 'artistId' in info else "" + self.comment = info['comment'] if 'comment' in info else "" + self.channelCount = int(info['channelCount']) if 'channelCount' in info else "" + self.samplingRate = int(info['samplingRate']) if 'samplingRate' in info else "" + self.playCount = int(info['playCount']) if 'playCount' in info else "" + +class Artist: + id: str + name: str + albumCount: int + coverArt: str + artistImageUrl: str + + def __init__(self, id: str, name: str, albumCount: str | int, coverArt: str, artistImageUrl: str): + self.id = id + self.name = name + self.albumCount = int(albumCount) + self.coverArt = coverArt + self.artistImageUrl = artistImageUrl + + def from_dict(info: dict): + return Artist( + info['id'] if 'id' in info else None, + info['name'] if 'name' in info else None, + info['albumCount'] if 'albumCount' in info else None, + info['coverArt'] if 'coverArt' in info else None, + info['artistImageUrl'] if 'artistImageUrl' in info else None + ) + +class BadSubsonicResponse(Exception): + pass + +class SubpyConn: + username: str + password: str + base_url: str + salt: str + salt_len: int = 8 + version: str = "1.16.1" + client_name: str = "testing" + + def __init__(self, u: str, p: str, base_url: str, salt_len: int | None = 8): + self.username = u + self.password = p + self.base_url = base_url + + self.salt_len = salt_len + + def _saltGenerator(self) -> None: + self.salt = ''.join(random.SystemRandom().choice(string.ascii_letters + string.digits) for _ in range(self.salt_len)) + + def _hashPass(self) -> str: + self._saltGenerator() + return hashlib.md5((self.password + self.salt).encode('utf-8')).hexdigest() + + def _getBasePayload(self) -> dict: + return { + "u" : self.username, + "t" : self._hashPass(), + "s" : self.salt, + "v" : self.version, + "c" : self.client_name, + "f": "json" + } + + def _encodePayload(self, payload: dict) -> str: + return urllib.parse.urlencode(payload, quote_via=urllib.parse.quote_plus) + + def _getMusicDirectory(self, id: str): + url = self.base_url + '/rest/getMusicDirectory?' + + payload = self._getBasePayload() + payload['id'] = id + + return requests.get(url + self._encodePayload(payload)) + + def getArtistAlbums(self, artist_id: str) -> list[Album]: + response: requests.Response = self._getMusicDirectory(artist_id).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + albums: list[Album] = [] + for item in response['directory']['child']: + albums.append(Album(item)) + + return albums + + def getAlbumSongs(self, album_id: str) -> list[Song]: + response: requests.Response = self._getMusicDirectory(album_id).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + songs: list[Song] = [] + for item in response['directory']['child']: + songs.append(Song(item)) + + return songs + + def getArtists(self) -> list[Artist]: + url = self.base_url + '/rest/getArtists?' + + payload = self._getBasePayload() + + response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + artists: list[Artist] = [] + + for artist_index in response['artists']['index']: + for artist in artist_index['artist']: + artists.append(Artist.from_dict(artist)) + + return artists + + def getAlbums(self, size: int | None = 25, offset: int | None = 0) -> list[Album]: + url = self.base_url + '/rest/getAlbumList?' + + payload = self._getBasePayload() + payload['type'] = 'alphabeticalByName' + payload['size'] = size + payload['offset'] = offset + + r = requests.get(url + self._encodePayload(payload)) + json_r = r.json() + + albums: list[Album] = [] + + if 'album' not in json_r['subsonic-response']['albumList']: + return [] + + for album in json_r['subsonic-response']['albumList']['album']: + albums.append(Album(album)) + + return albums + + def getAllAlbums(self) -> list[Album]: + all_albums: list[Album] = [] + + offset = 0 + while albums := self.getAlbums(50, offset): + all_albums.extend(albums) + offset += 50 + + return all_albums + + def getPlaylists(self) -> list[Playlist]: + url = self.base_url + "/rest/getPlaylists?" + + payload = self._getBasePayload() + + r = requests.get(url + self._encodePayload(payload)) + + json_r = r.json() + + playlists: list[Playlist] = [] + for playlist in json_r['subsonic-response']['playlists']['playlist']: + playlists.append(Playlist(playlist)) + + return playlists + + def getPlaylist(self, playlist_id: str) -> list[Song]: + url = self.base_url + "/rest/getPlaylist?" + + payload = self._getBasePayload() + payload['id'] = playlist_id + + r = requests.get(url + self._encodePayload(payload)) + + json_r = r.json() + + songs: list[Song] = [] + for song in json_r['subsonic-response']['playlist']['entry']: + songs.append(Song(song)) + + return songs + + def getStarred(self) -> list[Album | Song | Artist]: + url = self.base_url + "/rest/getStarred?" + + payload = self._getBasePayload() + + r = requests.get(url + self._encodePayload(payload)) + json_r = r.json() + + starred: list[Album | Song | Artist] = [] + for album in json_r['subsonic-response']['starred']['artist']: + starred.append(Album(album)) + + for artist in json_r['subsonic-response']['starred']['artist']: + starred.append(Artist.from_dict(artist)) + + for song in json_r['subsonic-response']['starred']['song']: + starred.append(Song(song)) + + return starred + + def getStarredSongs(self) -> list[Song]: + url = self.base_url + "/rest/getStarred?" + + payload = self._getBasePayload() + + r = requests.get(url + self._encodePayload(payload)) + json_r = r.json() + + starred: list[Song] = [] + for song in json_r['subsonic-response']['starred']['song']: + starred.append(Song(song)) + + return starred + + # Returns a requests Response object, the data stream can be found in Response.content + def stream(self, song_id: str) -> requests.Response: + url = self.base_url + '/rest/stream?' + + payload = self._getBasePayload() + + payload['id'] = song_id + + return requests.get(url + self._encodePayload(payload), stream=True) + + # Saves cover art id + def getCoverArt(self, cover_art_id: str, path: str | None = None) -> None: + url = self.base_url + '/rest/getCoverArt?' + + payload = self._getBasePayload() + payload['id'] = cover_art_id + + r = requests.get(url + self._encodePayload(payload)) + + with open(path if path else 'cover.png', 'wb') as file: + file.write(r.content) + + def getCoverArtRaw(self, cover_art_id: str) -> bytes: + url = self.base_url + '/rest/getCoverArt?' + + payload = self._getBasePayload() + payload['id'] = cover_art_id + + r = requests.get(url + self._encodePayload(payload)) + + return r.content + + def getSong(self, song_id: str) -> Song: + url = self.base_url + '/rest/getSong?' + + payload = self._getBasePayload() + payload['id'] = song_id + + response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + return Song(response['song']) + + def star(self, id: str) -> None: + url = self.base_url + '/rest/star?' + + payload = self._getBasePayload() + payload['id'] = id + + response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + def unstar(self, id: str) -> None: + url = self.base_url + '/rest/unstar?' + + payload = self._getBasePayload() + payload['id'] = id + + response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + """ + https://opensubsonic.netlify.app/docs/endpoints/scrobble/ + + id: str - unique identifier of the media to scrobble + OPTIONAL time: int - posix time of when the song was listened to + OPTIONAL submission: bool - Whether this is a "submission" or a "now playing" notification. Default: True + """ + def scrobble(self, id: str, time: int | None = None, submission: bool | None = True) -> None: + url = self.base_url + '/rest/scrobble?' + + payload = self._getBasePayload() + payload['id'] = id + + if time: + payload['time'] = time + + payload['submission'] = submission + + response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + """ + https://opensubsonic.netlify.app/docs/endpoints/getrandomsongs/ + """ + def getRandomSongs(self, + size: int | None = 10, + genre: str | None = None, + fromYear: str | int | None = None, + toYear: str | int | None = None, + musicFolderId: str | int | None = None + ) -> list[Song]: + url = self.base_url + '/rest/getRandomSongs?' + + payload = self._getBasePayload() + + # The default being 10 is only to make the method match the internal docs + if size and size != 10: + payload['size'] = size + + if genre: + payload['genre'] = genre + + if fromYear: + payload['fromYear'] = fromYear + + if toYear: + payload['toYear'] = toYear + + if musicFolderId: + payload['musicFolderId'] = musicFolderId + + response = requests.get(url + self._encodePayload(payload)).json()['subsonic-response'] + + if response['status'] != 'ok': + raise BadSubsonicResponse() + + random_songs: list[Song] = [] + + for song in response['randomSongs']['song']: + random_songs.append(Song(song)) + + return random_songs + + + diff --git a/screens-test.py b/textual1/app.py similarity index 83% rename from screens-test.py rename to textual1/app.py index 552750f..57977a4 100644 --- a/screens-test.py +++ b/textual1/app.py @@ -1,544 +1,540 @@ -import logging -import os -import io - -LOG_FILE = 'textual.log' -LOGGING_LEVEL = logging.DEBUG -LOG_FORMAT = '[%(asctime)s] %(levelname)s in %(module)s: %(message)s' - -# Clear Log File -os.remove(LOG_FILE) - -logging.basicConfig(filename=LOG_FILE, level=LOGGING_LEVEL, format=LOG_FORMAT) - -import SubSonic - -os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide" -import pygame - -from rich.table import Table, Column - -from textual import on, events, work -from textual.reactive import reactive -from textual.app import App, ComposeResult, RenderResult -from textual.widgets import Footer, Header, OptionList, DataTable, ProgressBar, Label -from textual.widget import Widget -from textual.containers import Horizontal, Vertical, Container, VerticalScroll -from textual.screen import Screen -from textual.widgets.option_list import Option -from textual.timer import Timer - -from rich_pixels import Pixels -from PIL import Image - -u = 'pseudo' -p = 'telegram9dock-MICHEL' - -conn = SubSonic.SubpyConn(u, p, "https://music.0x01fe.net") - -pygame.mixer.init() - -class AlbumSongs(Screen): - BINDINGS = [ - ('escape', 'die()', "Back"), - ('p', 'play()', 'Play Song'), - ('q' 'queue()', 'Queue Song') - ] - - selected_song_id: str | None = None - songs = None - - def __init__(self, album_id: str): - self.album_id = album_id - super().__init__() - - def compose(self) -> ComposeResult: - yield DataTable(id="albumSongs") - yield Footer() - - def on_mount(self) -> None: - self.songs = conn.getAlbumSongs(self.album_id) - songsTable: DataTable = self.query_one(DataTable) - - songsTable.cursor_type = 'row' - songsTable.add_columns("#", "Title", "Artist", "Duration", "Bitrate") - - for song in self.songs: - songsTable.add_row(song.trackNumber, song.title, song.artist, song.duration, song.bitRate, key=song.id) - - @on(DataTable.RowHighlighted) - def row_highlighted(self, event: DataTable.RowHighlighted) -> None: - self.selected_song_id = event.row_key.value - - def action_play(self) -> None: - song_index = self._find_selected_song() - app.play_song(self.songs[song_index]) - app.song_queue = self.songs[song_index + 1:] - - def action_queue(self) -> None: - app.queue_song(self._find_selected_song()) - - # Returns index of selected song - def _find_selected_song(self) -> int: - for i, song in enumerate(self.songs): - if song.id == self.selected_song_id: - return i - - # Pops and uninstalls self - def action_die(self) -> None: - app.pop_screen() - app.uninstall_screen("albumSongs") - -class ArtistAlbums(Screen): - BINDINGS = [ - ('escape', 'die()', "Back") - ] - - def __init__(self, artist_id: str): - self.artist_id = artist_id - super().__init__() - - def compose(self) -> ComposeResult: - yield DataTable(id="Albums") - yield Footer() - - def on_mount(self) -> None: - albums = conn.getArtistAlbums(self.artist_id) - albumTable: DataTable = self.query_one(DataTable) - - albumTable.cursor_type = 'row' - - albumTable.add_columns("Title", "Songs", "Plays", "Year", "Time") - - for album in albums: - albumTable.add_row(album.title, album.songCount, album.playCount, album.year, album.duration, key=album.id) - - @on(DataTable.RowSelected) - async def album_selected(self, event: DataTable.RowSelected) -> None: - app.install_screen(AlbumSongs(event.row_key.value), "albumSongs") - app.push_screen("albumSongs") - - # Pops and uninstalls self - def action_die(self) -> None: - app.pop_screen() - app.uninstall_screen("artistAlbums") - -class Artists(Screen): - - def compose(self) -> ComposeResult: - yield OptionList(id="Artists") - yield Footer() - - def on_mount(self) -> None: - artists = conn.getArtists() - artistList = self.query_one(OptionList) - - for artist in artists: - artistList.add_option(Option(artist.name, artist.id)) - - @on(OptionList.OptionSelected) - def artist_selected(self, event: OptionList.OptionSelected) -> None: - app.install_screen(ArtistAlbums(event.option_id), 'artistAlbums') - app.push_screen('artistAlbums') - -class Albums(Screen): - - def compose(self) -> ComposeResult: - yield DataTable() - yield Footer() - - def on_mount(self) -> None: - albums = conn.getAllAlbums() - albumTable = self.query_one(DataTable) - - albumTable.cursor_type = 'row' - albumTable.add_columns("Title", "Songs", "Plays", "Year", "Time") - - for album in albums: - albumTable.add_row(album.title, album.songCount, album.playCount, album.year, album.duration, key=album.id) - - @on(DataTable.RowSelected) - def album_selected(self, event: DataTable.RowSelected) -> None: - app.install_screen(AlbumSongs(event.row_key.value), 'albumSongs') - app.push_screen('albumSongs') - -class Playlists(Screen): - playlists: list[SubSonic.Playlist] | None = None - selected_playlist_id: str = None - - def compose(self) -> ComposeResult: - yield Header(icon="") - yield DataTable() - yield Footer() - - def on_mount(self) -> None: - app.sub_title = "Playlists" - self.playlists = conn.getPlaylists() - playlistTable = self.query_one(DataTable) - - playlistTable.cursor_type = "row" - playlistTable.add_columns("Name", "Songs", "Time", "Owner") - - for playlist in self.playlists: - playlistTable.add_row(playlist.name, playlist.songCount, playlist.duration, playlist.owner, key=playlist.id) - - @on(DataTable.RowHighlighted) - def row_highlighted(self, event: DataTable.RowHighlighted) -> None: - self.selected_playlist_id = event.row_key.value - - @on(DataTable.RowSelected) - def playlist_selected(self, event: DataTable.RowSelected) -> None: - app.install_screen(PlaylistSongs(self._find_selected_playlist()), "playlistSongs") - app.push_screen('playlistSongs') - - def _find_selected_playlist(self) -> SubSonic.Playlist: - for playlist in self.playlists: - if playlist.id == self.selected_playlist_id: - return playlist - -class PlaylistSongs(Screen): - BINDINGS = [ - ('escape', 'die()', "Back"), - ('p', 'play()', 'Play Song'), - ('q', 'queue()', 'Queue Song') - ] - - selected_song_id: str | None = None - - def __init__(self, playlist: SubSonic.Playlist): - self.playlist = playlist - self.songs = conn.getPlaylist(playlist.id) - super().__init__() - - def compose(self) -> ComposeResult: - self.sub_title = self.playlist.name - yield Header(icon="") - yield DataTable() - yield Footer() - - def on_mount(self) -> None: - songTable: DataTable = self.query_one(DataTable) - - songTable.cursor_type = "row" - songTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") - - for song in self.songs: - songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=song.id) - - # Pops and uninstalls self - def action_die(self) -> None: - app.pop_screen() - app.uninstall_screen("playlistSongs") - - @on(DataTable.RowHighlighted) - def row_highlighted(self, event: DataTable.RowHighlighted) -> None: - self.selected_song_id = event.row_key.value - - def action_play(self) -> None: - song_index = self._find_selected_song() - app.play_song(self.songs[song_index]) - app.song_queue = self.songs[song_index + 1:] - - def action_queue(self) -> None: - app.queue_song(self._find_selected_song()) - - # Returns index of selected song - def _find_selected_song(self) -> int: - for i, song in enumerate(self.songs): - if song.id == self.selected_song_id: - return i - -class Favourites(Screen): - BINDINGS = [ - ('p', 'play()', 'Play Song'), - ('q', 'queue()', 'Queue Song') - ] - - songs: list[SubSonic.Song] | None = None - selected_song_id: str | None = None - - def compose(self) -> ComposeResult: - yield Header(icon="") - yield DataTable() - yield Footer() - - def on_mount(self) -> None: - app.sub_title = "Favourites" - self.songs = conn.getStarredSongs() - songTable = self.query_one(DataTable) - - songTable.cursor_type = "row" - songTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") - - for song in self.songs: - songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=song.id) - - @on(DataTable.RowHighlighted) - def row_highlighted(self, event: DataTable.RowHighlighted) -> None: - self.selected_song_id = event.row_key.value - - def action_play(self) -> None: - song_index = self._find_selected_song() - app.play_song(self.songs[song_index]) - app.song_queue = self.songs[song_index + 1:] - - def action_queue(self) -> None: - app.queue_song(self._find_selected_song()) - - # Returns index of selected song - def _find_selected_song(self) -> int: - for i, song in enumerate(self.songs): - if song.id == self.selected_song_id: - return i - -class CoverArtPixels(Container): - - cover_art_id: str | None = None - - def set_cover_art(self, cover_art_id: str) -> None: - self.cover_art_id = cover_art_id - - def render(self) -> RenderResult: - - if not self.cover_art_id: - return "" - - # Make the size square - width, height = self.size - - if width > height: - width = height - elif height > width: - height = width - - image = Image.open(io.BytesIO(conn.getCoverArtRaw(self.cover_art_id))) - return Pixels().from_image(image, resize=(width, height)) - -class QueueTable(Widget): - - def render(self) -> RenderResult: - table = Table() - table.add_column("Title") - table.add_column("Album") - table.add_column("Artist") - table.add_column("Plays") - table.add_column("Duration") - table.add_column("Bitrate") - - queue: list[SubSonic.Song] = app.user_queue - queue.extend(app.song_queue) - - for song in queue: - table.add_row(song.title, song.album, song.artist, str(song.playCount), str(song.duration), str(song.bitRate)) - - return table - -# Fine to be a static renderable because it's not big enough to need a scroll -class PlayingInfo(Widget): - - def render(self) -> RenderResult: - table = Table() - - table.add_column("") - table.add_column("", min_width=10) - table.show_header = False - - if app.currently_playing: - song = app.currently_playing - table.add_row("Title", song.title) - table.add_row("Artist", song.artist) - table.add_row("Album", song.album) - table.add_row("Plays", str(song.playCount) if song.playCount else "0") - table.add_row("Duration", str(song.duration)) - table.add_row("Track Number", str(song.trackNumber)) - table.add_row("Bit Rate", str(song.bitRate)) - - return table - -# https://textual.textualize.io/widgets/data_table/#labeled-rows -# Add this ^ to queue sometime plz -class MainView(Screen): - CSS_PATH = 'main.tcss' - - def compose(self) -> ComposeResult: - yield Header(icon="") - with Horizontal(): - with Vertical(classes="info", id='info'): - yield Label("Playing") - yield PlayingInfo() - yield ProgressBar() - yield CoverArtPixels() - with Vertical(classes="column"): - yield Label("Queue") - yield DataTable(id="queue") - # yield VerticalScroll(QueueTable()) - # yield Footer() - - def on_mount(self) -> None: - app.sub_title = "Overview" - queueTable: DataTable = self.query_one("#queue") - queueTable.cursor_type = "row" - queueTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") - - # currentlyPlaying: DataTable = self.query_one('#currently_playing') - # currentlyPlaying.cursor_type = "row" - # currentlyPlaying.add_columns("", "") - - @on(events.ScreenResume) - def screen_resumed(self, event: events.ScreenResume) -> None: - app.sub_title = "Overview" - - # Update Queue Table - queueTable: DataTable = self.query_one('#queue') - self._clear_table('#queue') - - queue: list[SubSonic.Song] = [] - queue.extend(app.user_queue) - queue.extend(app.song_queue) - for song in queue: - queueTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate) - - # Update Currently Playing Table - # self._clear_table('#currently_playing') - if app.currently_playing: - # song = app.currently_playing - # currentlyPlaying: DataTable = self.query_one('#currently_playing') - # currentlyPlaying.add_row("Title", song.title) - # currentlyPlaying.add_row("Artist", song.artist) - # currentlyPlaying.add_row("Album", song.album) - # currentlyPlaying.add_row("Plays", song.playCount if song.playCount else 0) - # currentlyPlaying.add_row("Duration", song.duration) - # currentlyPlaying.add_row("Track Number", song.trackNumber) - # currentlyPlaying.add_row("Bit Rate", song.bitRate) - - self.query_one(CoverArtPixels).set_cover_art(app.currently_playing.id) - - def _clear_table(self, table_id: str) -> None: - table: DataTable = self.query_one(table_id) - - row_keys = [] - for row_key, row in table.rows.items(): - row_keys.append(row_key) - - for row_key in row_keys: - table.remove_row(row_key) - - -class TextualSubpyPlayer(App): - BINDINGS = [ - ("r", "rewind()", "⏪"), - ("space", "play_pause_song()", "⏯️"), - ("s", "skip()", "⏭️"), - ("f1", "switch_mode('main')", "Overview"), - ("f2", "switch_mode('artists')", "Artists"), - ("f3", "switch_mode('albums')", "Albums"), - ("f4", "switch_mode('favourites')", "Favourites"), - ("f5", "switch_mode('playlists')", "Playlists") - ] - - MODES = { - "main" : MainView, - "artists" : Artists, - "albums" : Albums, - "favourites" : Favourites, - "playlists" : Playlists - } - - currently_playing: SubSonic.Song | None = reactive(SubSonic.Song()) - song_queue: list[SubSonic.Song] = reactive([]) - user_queue: list[SubSonic.Song] = reactive([]) - play_history: list[SubSonic.Song] = [] - is_paused: bool = False - place: int = 0 - - check_playing_timer: Timer | None = None - - def on_mount(self) -> None: - self.title = "SubPy" - self.switch_mode('main') - - def on_key(self, event: events.Key) -> None: - # logging.debug(f'Key Pressed {event.name}') - pass - - @work(exclusive=True) - async def play_song(self, song: SubSonic.Song) -> None: - self.currently_playing = song - self.is_paused = False - logging.debug(f"Playing song {song.title} by {song.artist}.") - - self.notify(f"Playing \"{song.title}\" by {song.artist}.") - - r = conn.stream(song.id) - audio_data = io.BytesIO(r.content) - - pygame.mixer.music.load(audio_data) - pygame.mixer.music.play() - - self.check_playing_timer = self.set_interval(1 / 5, self.check_done_playing, pause=False) - - # playingLabel: Label = self.query_one(Label) - # playingLabel. - - # Set callback loop for progress bar - # self.progress_timer = self.set_interval(1 / 10, self.update_progress, pause=False) - - def check_done_playing(self) -> None: - if not pygame.mixer.music.get_busy() and not self.is_paused: - self.play_history.append(self.currently_playing) - queue = self.user_queue - queue.extend(self.song_queue) - - next_song = queue.pop(0) - - self.play_song(next_song) - - self.check_playing_timer.stop() - - def action_skip(self) -> None: - - if self.check_playing_timer: - self.check_playing_timer.stop() - - if self.currently_playing: - self.play_history.append(self.currently_playing) - - queue = self.user_queue - queue.extend(self.song_queue) - - self.play_song(queue.pop(0)) - - def action_rewind(self) -> None: - if self.check_playing_timer: - self.check_playing_timer.stop() - - self.play_song(self.play_history.pop()) - - # def update_progress(self) -> None: - # self.query_one(ProgressBar).advance(1) - - def queue_song(self, song: SubSonic.Song) -> None: - logging.debug(f'Queue\'d song "{song.title}" by {song.artist}.') - self.notify(f'Queue\'d song "{song.title}" by {song.artist}.') - self.user_queue.append(song) - - def action_play_pause_song(self) -> None: - self.play_pause_song() - - def play_pause_song(self) -> None: - if not self.currently_playing: - return None - - if not self.is_paused: - self.notify("Playback paused") - self.place = pygame.mixer.music.get_pos() - pygame.mixer.music.pause() - self.is_paused = True - else: - self.notify("Resuming Playback") - pygame.mixer.music.unpause() - self.is_paused = False - -if __name__ == '__main__': - app = TextualSubpyPlayer() - app.run() +import logging +import os +import io + +LOG_FILE = 'textual.log' +LOGGING_LEVEL = logging.DEBUG +LOG_FORMAT = '[%(asctime)s] %(levelname)s in %(module)s: %(message)s' + +# Clear Log File +if os.path.exists(LOG_FILE): + os.remove(LOG_FILE) + +logging.basicConfig(filename=LOG_FILE, level=LOGGING_LEVEL, format=LOG_FORMAT) + +import SubSonic + +os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide" +import pygame + +from rich.table import Table, Column + +from textual import on, events, work +from textual.reactive import reactive +from textual.app import App, ComposeResult, RenderResult +from textual.widgets import Footer, Header, OptionList, DataTable, ProgressBar, Label +from textual.widget import Widget +from textual.containers import Horizontal, Vertical, Container, VerticalScroll +from textual.screen import Screen +from textual.widgets.option_list import Option +from textual.timer import Timer + +from rich_pixels import Pixels +from PIL import Image + +u, p = [s.strip() for s in open('auth', 'r').read().split(',')] + +conn = SubSonic.SubpyConn(u, p, "https://music.0x01fe.net") + +pygame.mixer.init() + +class AlbumSongs(Screen): + BINDINGS = [ + ('escape', 'die()', "Back"), + ('p', 'play()', 'Play Song'), + ('q' 'queue()', 'Queue Song') + ] + + selected_song_id: str | None = None + songs = None + + def __init__(self, album_id: str): + self.album_id = album_id + super().__init__() + + def compose(self) -> ComposeResult: + yield DataTable(id="albumSongs") + yield Footer() + + def on_mount(self) -> None: + self.songs = conn.getAlbumSongs(self.album_id) + songsTable: DataTable = self.query_one(DataTable) + + songsTable.cursor_type = 'row' + songsTable.add_columns("#", "Title", "Artist", "Duration", "Bitrate") + + for song in self.songs: + songsTable.add_row(song.trackNumber, song.title, song.artist, song.duration, song.bitRate, key=song.id) + + @on(DataTable.RowHighlighted) + def row_highlighted(self, event: DataTable.RowHighlighted) -> None: + self.selected_song_id = event.row_key.value + + def action_play(self) -> None: + song_index = self._find_selected_song() + app.play_song(self.songs[song_index]) + app.song_queue = self.songs[song_index + 1:] + app.update_queue_table() + + def action_queue(self) -> None: + app.queue_song(self._find_selected_song()) + + # Returns index of selected song + def _find_selected_song(self) -> int: + for i, song in enumerate(self.songs): + if song.id == self.selected_song_id: + return i + + # Pops and uninstalls self + def action_die(self) -> None: + app.pop_screen() + app.uninstall_screen("albumSongs") + +class ArtistAlbums(Screen): + BINDINGS = [ + ('escape', 'die()', "Back") + ] + + def __init__(self, artist_id: str): + self.artist_id = artist_id + super().__init__() + + def compose(self) -> ComposeResult: + yield DataTable(id="Albums") + yield Footer() + + def on_mount(self) -> None: + albums = conn.getArtistAlbums(self.artist_id) + albumTable: DataTable = self.query_one(DataTable) + + albumTable.cursor_type = 'row' + + albumTable.add_columns("Title", "Songs", "Plays", "Year", "Time") + + for album in albums: + albumTable.add_row(album.title, album.songCount, album.playCount, album.year, album.duration, key=album.id) + + @on(DataTable.RowSelected) + async def album_selected(self, event: DataTable.RowSelected) -> None: + app.install_screen(AlbumSongs(event.row_key.value), "albumSongs") + app.push_screen("albumSongs") + + # Pops and uninstalls self + def action_die(self) -> None: + app.pop_screen() + app.uninstall_screen("artistAlbums") + +class Artists(Screen): + + def compose(self) -> ComposeResult: + yield OptionList(id="Artists") + yield Footer() + + def on_mount(self) -> None: + artists = conn.getArtists() + artistList = self.query_one(OptionList) + + for artist in artists: + artistList.add_option(Option(artist.name, artist.id)) + + @on(OptionList.OptionSelected) + def artist_selected(self, event: OptionList.OptionSelected) -> None: + app.install_screen(ArtistAlbums(event.option_id), 'artistAlbums') + app.push_screen('artistAlbums') + +class Albums(Screen): + + def compose(self) -> ComposeResult: + yield DataTable() + yield Footer() + + def on_mount(self) -> None: + albums = conn.getAllAlbums() + albumTable = self.query_one(DataTable) + + albumTable.cursor_type = 'row' + albumTable.add_columns("Title", "Songs", "Plays", "Year", "Time") + + for album in albums: + albumTable.add_row(album.title, album.songCount, album.playCount, album.year, album.duration, key=album.id) + + @on(DataTable.RowSelected) + def album_selected(self, event: DataTable.RowSelected) -> None: + app.install_screen(AlbumSongs(event.row_key.value), 'albumSongs') + app.push_screen('albumSongs') + +class Playlists(Screen): + playlists: list[SubSonic.Playlist] | None = None + selected_playlist_id: str = None + + def compose(self) -> ComposeResult: + yield Header(icon="") + yield DataTable() + yield Footer() + + def on_mount(self) -> None: + app.sub_title = "Playlists" + self.playlists = conn.getPlaylists() + playlistTable = self.query_one(DataTable) + + playlistTable.cursor_type = "row" + playlistTable.add_columns("Name", "Songs", "Time", "Owner") + + for playlist in self.playlists: + playlistTable.add_row(playlist.name, playlist.songCount, playlist.duration, playlist.owner, key=playlist.id) + + @on(DataTable.RowHighlighted) + def row_highlighted(self, event: DataTable.RowHighlighted) -> None: + self.selected_playlist_id = event.row_key.value + + @on(DataTable.RowSelected) + def playlist_selected(self, event: DataTable.RowSelected) -> None: + app.install_screen(PlaylistSongs(self._find_selected_playlist()), "playlistSongs") + app.push_screen('playlistSongs') + + def _find_selected_playlist(self) -> SubSonic.Playlist: + for playlist in self.playlists: + if playlist.id == self.selected_playlist_id: + return playlist + +class PlaylistSongs(Screen): + BINDINGS = [ + ('escape', 'die()', "Back"), + ('p', 'play()', 'Play Song'), + ('q', 'queue()', 'Queue Song') + ] + + selected_song_id: str | None = None + + def __init__(self, playlist: SubSonic.Playlist): + self.playlist = playlist + self.songs = conn.getPlaylist(playlist.id) + super().__init__() + + def compose(self) -> ComposeResult: + self.sub_title = self.playlist.name + yield Header(icon="") + yield DataTable() + yield Footer() + + def on_mount(self) -> None: + songTable: DataTable = self.query_one(DataTable) + + songTable.cursor_type = "row" + songTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") + + for song in self.songs: + songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=song.id) + + # Pops and uninstalls self + def action_die(self) -> None: + app.pop_screen() + app.uninstall_screen("playlistSongs") + + @on(DataTable.RowHighlighted) + def row_highlighted(self, event: DataTable.RowHighlighted) -> None: + self.selected_song_id = event.row_key.value + + def action_play(self) -> None: + song_index = self._find_selected_song() + app.play_song(self.songs[song_index]) + app.song_queue = self.songs[song_index + 1:] + app.update_queue_table() + + def action_queue(self) -> None: + app.queue_song(self._find_selected_song()) + + # Returns index of selected song + def _find_selected_song(self) -> int: + for i, song in enumerate(self.songs): + if song.id == self.selected_song_id: + return i + +class Favourites(Screen): + BINDINGS = [ + ('p', 'play()', 'Play Song'), + ('q', 'queue()', 'Queue Song') + ] + + songs: list[SubSonic.Song] | None = None + selected_song_id: str | None = None + + def compose(self) -> ComposeResult: + yield Header(icon="") + yield DataTable() + yield Footer() + + def on_mount(self) -> None: + app.sub_title = "Favourites" + self.songs = conn.getStarredSongs() + songTable = self.query_one(DataTable) + + songTable.cursor_type = "row" + songTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") + + for song in self.songs: + songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=song.id) + + @on(DataTable.RowHighlighted) + def row_highlighted(self, event: DataTable.RowHighlighted) -> None: + self.selected_song_id = event.row_key.value + + def action_play(self) -> None: + song_index = self._find_selected_song() + app.play_song(self.songs[song_index]) + app.song_queue = self.songs[song_index + 1:] + app.update_queue_table() + + def action_queue(self) -> None: + app.queue_song(self._find_selected_song()) + + # Returns index of selected song + def _find_selected_song(self) -> int: + for i, song in enumerate(self.songs): + if song.id == self.selected_song_id: + return i + +# TODO fix coverart +class CoverArtPixels(Container): + + cover = reactive[Image.Image | None](None) + + def render(self) -> RenderResult: + if self.cover: + width, height = self.size + width -= 4 + height -= 4 + + if width > height: + width = height + elif height > width: + height = width + + return Pixels.from_image(self.cover, resize=(width, height)) + + return '' + + @work(exclusive=True, thread=True) + def get_cover_art(self, cover_art_id: str) -> None: + try: + cover_art_data = conn.getCoverArtRaw(cover_art_id) + image = Image.open(io.BytesIO(cover_art_data)) + self.cover = image + except Exception as e: + logging.error(f"Failed to get cover art: {e}") + self.cover = None + +class QueueTable(Widget): + + def render(self) -> RenderResult: + table = Table() + table.add_column("Title") + table.add_column("Album") + table.add_column("Artist") + table.add_column("Plays") + table.add_column("Duration") + table.add_column("Bitrate") + + queue: list[SubSonic.Song] = app.user_queue + queue.extend(app.song_queue) + + for song in queue: + table.add_row(song.title, song.album, song.artist, str(song.playCount), str(song.duration), str(song.bitRate)) + + return table + +# Fine to be a static renderable because it's not big enough to need a scroll +class PlayingInfo(Widget): + + def render(self) -> RenderResult: + table = Table() + + table.add_column("") + table.add_column("", min_width=10) + table.show_header = False + + if app.currently_playing: + song = app.currently_playing + table.add_row("Title", song.title) + table.add_row("Artist", song.artist) + table.add_row("Album", song.album) + table.add_row("Plays", str(song.playCount) if song.playCount else "0") + table.add_row("Duration", str(song.duration)) + table.add_row("Track Number", str(song.trackNumber)) + table.add_row("Bit Rate", str(song.bitRate)) + + return table + +# https://textual.textualize.io/widgets/data_table/#labeled-rows +# Add this ^ to queue sometime plz +class MainView(Screen): + CSS_PATH = 'main.tcss' + + def compose(self) -> ComposeResult: + yield Header(icon="") + with Horizontal(): + with Vertical(classes="info", id='info'): + yield Label("Playing") + yield PlayingInfo() + yield ProgressBar() + yield CoverArtPixels(id="coverart") + with Vertical(classes="column"): + yield Label("Queue") + yield DataTable(id="queue", cursor_type='row') + yield Footer() + + def on_mount(self) -> None: + app.sub_title = "Overview" + queueTable: DataTable = self.query_one(DataTable) + queueTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") + + @on(events.ScreenResume) + def screen_resumed(self, event: events.ScreenResume) -> None: + app.sub_title = "Overview" + self.update_queue() + + def update_queue(self) -> None: + table = self.query_one(DataTable) + table.clear() + + queue: list[SubSonic.Song] = [] + queue.extend(app.user_queue) + queue.extend(app.song_queue) + + for song in queue: + table.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate) + + +class TextualSubpyPlayer(App): + BINDINGS = [ + ("r", "rewind()", "⏪"), + ("space", "play_pause_song()", "⏯"), + ("s", "skip()", "⏭"), + ("f1", "switch_mode('main')", "Overview"), + ("f2", "switch_mode('artists')", "Artists"), + ("f3", "switch_mode('albums')", "Albums"), + ("f4", "switch_mode('favourites')", "Favourites"), + ("f5", "switch_mode('playlists')", "Playlists") + ] + + MODES = { + "main" : MainView, + "artists" : Artists, + "albums" : Albums, + "favourites" : Favourites, + "playlists" : Playlists + } + + currently_playing: SubSonic.Song | None = reactive(SubSonic.Song()) + song_queue: list[SubSonic.Song] = reactive([]) + user_queue: list[SubSonic.Song] = reactive([]) + play_history: list[SubSonic.Song] = [] + is_paused: bool = False + place: int = 0 + + check_playing_timer: Timer | None = None + + def on_mount(self) -> None: + self.title = "SubPy" + self.switch_mode('main') + + def on_key(self, event: events.Key) -> None: + # logging.debug(f'Key Pressed {event.name}') + pass + + @work(exclusive=True) + async def play_song(self, song: SubSonic.Song) -> None: + self.currently_playing = song + self.is_paused = False + logging.debug(f"Playing song {song.title} by {song.artist}.") + + self.notify(f"Playing \"{song.title}\" by {song.artist}.") + + r = conn.stream(song.id) + audio_data = io.BytesIO(r.content) + + pygame.mixer.music.load(audio_data) + pygame.mixer.music.play() + + self.check_playing_timer = self.set_interval(1 / 5, self.check_done_playing, pause=False) + + self.update_queue_table() + + # playingLabel: Label = self.query_one(Label) + # playingLabel. + + # Set callback loop for progress bar + # self.progress_timer = self.set_interval(1 / 10, self.update_progress, pause=False) + + def check_done_playing(self) -> None: + if not pygame.mixer.music.get_busy() and not self.is_paused: + self.play_history.append(self.currently_playing) + queue = self.user_queue + queue.extend(self.song_queue) + + next_song = queue.pop(0) + + self.play_song(next_song) + + self.check_playing_timer.stop() + + def action_skip(self) -> None: + + if self.check_playing_timer: + self.check_playing_timer.stop() + + if self.currently_playing: + self.play_history.append(self.currently_playing) + + queue = self.user_queue + queue.extend(self.song_queue) + + self.play_song(queue.pop(0)) + + def action_rewind(self) -> None: + if self.check_playing_timer: + self.check_playing_timer.stop() + + self.play_song(self.play_history.pop()) + + # def update_progress(self) -> None: + # self.query_one(ProgressBar).advance(1) + + def queue_song(self, song: SubSonic.Song) -> None: + logging.debug(f'Queue\'d song "{song.title}" by {song.artist}.') + self.notify(f'Queue\'d song "{song.title}" by {song.artist}.') + self.update_user_queue(song) + + def action_play_pause_song(self) -> None: + self.play_pause_song() + + def play_pause_song(self) -> None: + if not self.currently_playing: + return None + + if not self.is_paused: + self.notify("Playback paused") + self.place = pygame.mixer.music.get_pos() + pygame.mixer.music.pause() + self.is_paused = True + else: + self.notify("Resuming Playback") + pygame.mixer.music.unpause() + self.is_paused = False + + def update_queue_table(self) -> None: + if isinstance(self.screen, MainView): + logging.debug("Updating queue through app") + self.screen.update_queue() + + def update_user_queue(self, song: SubSonic.Song) -> None: + self.user_queue.append(song) + self.update_queue_table() + + def update_song_queue(self, song: SubSonic.Song) -> None: + self.song_queue.append(song) + self.update_queue_table() + +if __name__ == '__main__': + app = TextualSubpyPlayer() + app.run() diff --git a/main.tcss b/textual1/main.tcss similarity index 92% rename from main.tcss rename to textual1/main.tcss index 2c55b1c..9c063e7 100644 --- a/main.tcss +++ b/textual1/main.tcss @@ -1,4 +1,4 @@ -.info { - width: 30%; - padding-right: 2; -} +.info { + width: 30%; + padding-right: 2; +} diff --git a/requirements.txt b/textual1/requirements.txt similarity index 95% rename from requirements.txt rename to textual1/requirements.txt index 16ad453..9895f89 100644 --- a/requirements.txt +++ b/textual1/requirements.txt @@ -1,5 +1,5 @@ -requests==2.31.0 -textual==1.0.0 -pygame==2.6.0 -Pillow==10.0.0 +requests==2.31.0 +textual==1.0.0 +pygame==2.6.0 +Pillow==10.0.0 rich-pixels==3.0.1 \ No newline at end of file