subpy/widgets.py

import asyncio
import random
import re

from dataclasses import dataclass

import pyperclip

from textual import on
from textual.widgets import DataTable, Input, Label, Button, RadioSet, RadioButton
from textual.widget import Widget
from textual.app import ComposeResult
from textual.binding import Binding
from textual.screen import ModalScreen
from textual.containers import Container, Horizontal, Vertical

import SubSonic

def fix_duration(seconds: int) -> str:
    min, sec = divmod(seconds, 60)

    min = max(0, min)
    sec = max(0, sec)

    formatted = f"{min}:{sec:02d}"

    return formatted

class RatingModal(ModalScreen[int]):
    """Modal for getting a numerical rating for an Album, Artist, or Song"""

    BINDINGS = [
        ('escape', 'die', 'Close')
    ]

    thing_title: str
    selected_rating: int = 0

    def __init__(self, thing_title: str):
        self.thing_title = thing_title
        super().__init__()

    def on_mount(self) -> None:
        label: Label = self.query_one(Label)
        label.update(f'Select a rating for {self.thing_title}')

    def compose(self) -> ComposeResult:
        with Vertical():
            yield Label()
            with Horizontal():
                with RadioSet():
                    yield RadioButton("0", value = True)
                    yield RadioButton("1")
                    yield RadioButton("2")
                    yield RadioButton("3")
                    yield RadioButton("4")
                    yield RadioButton("5")
            with Horizontal():
                yield Button("Submit")

    @on(Button.Pressed)
    def selection_submit(self, event: Button.Pressed) -> None:
        self.dismiss(self.selected_rating)

    @on(RadioSet.Changed)
    def selection_changed(self, event: RadioSet.Changed) -> None:
        self.selected_rating = event.radio_set.pressed_index

    def action_die(self) -> None:
        self.dismiss()

class ShareLinkModal(ModalScreen):
    """Modal for copying a generated share link"""

    BINDINGS = [
        ('escape', 'die', 'Close')
    ]

    link: str

    def __init__(self, link: str):
        self.link = link
        super().__init__()

    def on_mount(self) -> None:
        label: Label = self.query_one(Label)
        label.update(self.link)

    def compose(self) -> ComposeResult:
        with Horizontal():
            yield Label()
            yield Button("Copy Link")

    @on(Button.Pressed)
    def copy_link(self) -> None:
        pyperclip.copy(self.link)

    def action_die(self) -> None:
        self.dismiss()

class PlaylistSelectionModal(ModalScreen[str]):
    """Modal for selecting a playlist"""

    playlists: list[SubSonic.Playlist] = []
    selected_playlist_id: str = ""
    BINDINGS = [
        ('escape', 'die', 'Back')
    ]

    def compose(self) -> ComposeResult:
        with Container():
            yield Label("Select a playlist")
            yield DataTable()

    def on_mount(self) -> None:
        self.playlists = self.app.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:
        self.dismiss(self.selected_playlist_id)

    def action_die(self) -> None:
        self.dismiss("")

@dataclass
class SongTableColumnConfig:
    track_number: bool = False
    title: bool = True
    album: bool = True
    artist: bool = True
    plays: bool = True
    duration: bool = True
    rating: bool = False
    bitrate: bool = False

# Maps config field -> (column header, song attribute)
COLUMN_MAPPING: list[tuple[str, str, str]] = [
    ('track_number', '#', 'trackNumber'),
    ('title', 'Title', 'title'),
    ('album', 'Album', 'album'),
    ('artist', 'Artist', 'artist'),
    ('plays', 'Plays', 'playCount'),
    ('duration', 'Duration', 'duration'),
    ('rating', 'Rating', 'rating'),
    ('bitrate', 'Bitrate', 'bitRate')
]

"""
A custom DataTable designed to hold and display songs

TODO @0x01fe : needs optional paging on init
"""
class SongTable(Widget):
    BINDINGS = [
        ('p', 'play()', 'Play Song'),
        ('q', 'queue()', 'Queue Song'),
        ('m', 'shuffle()', 'Shuffle Play'),
        ('A', 'add_to_playlist()', 'Add to Playlist'),
        ('f', 'star_song()', 'Star Song'),
        ('S', 'share_song()', 'Share Song'),
        ('R', 'rate_song()', 'Rate Song'),
        Binding('/', 'filter()', 'Search', show=False)
    ]

    columns: SongTableColumnConfig
    songs: list[SubSonic.Song] = []
    filter_active: bool = False
    selected_song_id: str | None = None
    max_col_width: int | None = None

    OVERFLOW_INDICATOR: str = "..."

    def __init__(self, columns: SongTableColumnConfig | None = SongTableColumnConfig(), songs: list[SubSonic.Song] | None = [], max_col_width: int | None = None, **kwargs):
        self.columns = columns
        self.songs = songs
        self.max_col_width = max_col_width
        super().__init__(**kwargs)

    def compose(self) -> ComposeResult:
        yield Input(placeholder="Search", id="filter")
        yield DataTable()

    def _format_cell(self, config_attr: str, value):
        """Format cell values for display in the DataTable."""
        if config_attr == "duration":
            try:
                return fix_duration(int(value or 0))
            except (TypeError, ValueError):
                return fix_duration(0)
        if self.max_col_width and type(value) == str and len(value) > self.max_col_width:
            value = value[:self.max_col_width - len(self.OVERFLOW_INDICATOR)]
            value += self.OVERFLOW_INDICATOR

        return value

    def _row_data_for_song(self, song: SubSonic.Song) -> list:
        """Build DataTable row values based on configured columns."""
        row_data = []
        for config_attr, _, song_attr in COLUMN_MAPPING:
            if getattr(self.columns, config_attr):
                value = getattr(song, song_attr)
                row_data.append(self._format_cell(config_attr, value))
        return row_data

    def on_mount(self) -> None:
        dt: DataTable = self.query_one(DataTable)
        dt.focus()
        dt.cursor_type = "row"

        # Add columns based on config
        for config_attr, header, _ in COLUMN_MAPPING:
            if getattr(self.columns, config_attr):
                dt.add_column(header)

        # Add rows for existing songs
        for i, song in enumerate(self.songs):
            dt.add_row(*self._row_data_for_song(song), key=str(i))

    def add_song(self, song: SubSonic.Song) -> None:
        self.songs.append(song)
        dt: DataTable = self.query_one(DataTable)
        dt.add_row(*self._row_data_for_song(song), key=str(len(self.songs) - 1))

    @on(Input.Submitted)
    def search_submit(self, event: Input.Submitted) -> None:
        search_value = event.value.lower()

        dt: DataTable = self.query_one(DataTable)
        dt.clear()

        for i, song in enumerate(self.songs):
            if (
                re.search(search_value, song.title.lower())
                or re.search(search_value, song.artist.lower())
                or re.search(search_value, song.album.lower())
            ):
                dt.add_row(*self._row_data_for_song(song), key=str(i))

    @on(DataTable.RowHighlighted)
    def row_highlighted(self, event: DataTable.RowHighlighted) -> None:
        if not self.songs:
            return
        self.selected_song_id = self.songs[int(event.row_key.value)].id

    def action_play(self) -> None:
        song_index: int = self._find_selected_song_index()
        self.app.play_song(self.songs[song_index])
        self.app.song_queue = self.songs[song_index + 1:]

    def action_queue(self) -> None:
        song_index: int = self._find_selected_song_index()
        self.app.queue_song(self.songs[song_index])

    def action_shuffle(self) -> None:
        q: list[SubSonic.Song] = self.songs.copy()
        random.shuffle(q)
        self.app.play_song(q.pop(0))
        self.app.song_queue = q

    def action_add_to_playlist(self) -> None:
        
        def process_modal_selection(playlist_id: str) -> None:
            self.app.conn.updatePlaylist(playlist_id, songIdToAdd=[self.selected_song_id])
            self.app.notify("Added song to playlist!")

        self.app.push_screen(PlaylistSelectionModal(), process_modal_selection)

    def action_star_song(self) -> None:
        song = self.songs[self._find_selected_song_index()]

        is_song_starred: bool = self.app.navidrome.isSongStarred(song.id)

        if not is_song_starred:
            asyncio.create_task(asyncio.to_thread(self.app.conn.star, self.selected_song_id))
            self.app.notify(f'Starred song "{song.title}"')
        else:
            asyncio.create_task(asyncio.to_thread(self.app.conn.unstar, self.selected_song_id))
            self.app.notify(f'Unstarred song "{song.title}"')

    def action_share_song(self) -> None:
        share_link: str = self.app.conn.createShare(self.selected_song_id)
        self.app.push_screen(ShareLinkModal(share_link))

    def action_rate_song(self) -> None:
        song: SubSonic.Song = self.songs[self._find_selected_song_index()]

        def process_modal_selection(rating: int | None) -> None:
            if rating == None:
                return

            self.app.conn.setRating(self.selected_song_id, rating)
            if rating != 0:
                self.app.notify(f'Rated "{song.title}" a {rating}.')
            else:
                self.app.notify(f'Removed rating for "{song.title}".')

        self.app.push_screen(RatingModal(f'"{song.title}" By {song.artist}'), process_modal_selection)

    # Just makes the filter appear
    def action_filter(self) -> None:
        search_bar: Input = self.query_one(Input)

        if not self.filter_active:
            search_bar.display = 'block'
        else:
            search_bar.display = 'none'

        self.filter_active = not self.filter_active
        search_bar.focus()

    # Returns index of selected song
    def _find_selected_song_index(self) -> int:
        for i, song in enumerate(self.songs):
            if song.id == self.selected_song_id:
                return i