Add admin dashboard, API endpoints, CLI provider

xnetcat · xnetcat · commit 6cc9e6ffe25f · 2026-02-24T14:34:15.000+01:00
Add admin UI and backend integrations for the CLI, plus a command-palette provider and deployment configs. New admin screen (cli/src/spotdl_cli/screens/admin.py) implements dashboard, matches/reports/users tables and actions; account screen shows an Admin Dashboard button and pushes the admin screen. API client (cli/src/spotdl_cli/core/api_client.py) gains admin endpoints: get_admin_stats, get_admin_matches, approve_match, reject_match, get_admin_reports, resolve_report, get_admin_users, update_user_role, and ban_user. Add a SpotDLCommandProvider (cli/src/spotdl_cli/providers.py) to surface universal search results in the command palette and register it with the app. Small test update: test_api_client now calls get_metadata_sources with an id. Also add railway.toml files for backend and frontend deployment configuration.
diff --git a/backend/railway.toml b/backend/railway.toml
@@ -0,0 +1,9 @@
+[build]
+builder = "DOCKERFILE"
+dockerfilePath = "Dockerfile"
+watchPatterns = ["backend/**", "core/**"]
+
+[deploy]
+healthcheckPath = "/api/v1/health"
+healthcheckTimeout = 100
+restartPolicyType = "ON_FAILURE"
diff --git a/cli/src/spotdl_cli/app.py b/cli/src/spotdl_cli/app.py
@@ -25,6 +25,7 @@
 from spotdl_cli.screens.onboarding import OnboardingScreen, should_show_onboarding
 from spotdl_cli.screens.queue import QueueScreen
 from spotdl_cli.screens.settings import SettingsScreen
+from spotdl_cli.providers import SpotDLCommandProvider
 
 if TYPE_CHECKING:
     from textual.screen import Screen
@@ -38,6 +39,8 @@ class SpotDLApp(App[None]):
     TITLE = "SpotDL"
     SUB_TITLE = f"v{__version__}"
     CSS_PATH = "app.tcss"
+    
+    COMMANDS = App.COMMANDS | {SpotDLCommandProvider}
 
     BINDINGS: ClassVar[list[Binding]] = [
         Binding("q", "quit", "Quit", priority=True),
diff --git a/cli/src/spotdl_cli/core/api_client.py b/cli/src/spotdl_cli/core/api_client.py
@@ -1560,6 +1560,155 @@ async def enrich_song_all_sources(self, song_id: str) -> dict[str, Any]:
         except httpx.HTTPError as e:
             raise APIError(f"Request failed: {e}") from e
 
+    async def get_admin_stats(self) -> dict[str, Any]:
+        """Get admin dashboard statistics."""
+        try:
+            client = await self._get_client()
+            response = await client.get("/api/v1/admin/stats")
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def get_admin_matches(
+        self, status: str | None = None, limit: int = 50, offset: int = 0
+    ) -> dict[str, Any]:
+        """Get matches for admin review."""
+        try:
+            client = await self._get_client()
+            params = {"limit": limit, "offset": offset}
+            if status:
+                params["status"] = status
+            response = await client.get("/api/v1/admin/matches", params=params)
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def approve_match(self, match_id: str) -> dict[str, Any]:
+        """Approve a match submission."""
+        try:
+            client = await self._get_client()
+            response = await client.post(f"/api/v1/admin/matches/{match_id}/approve")
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def reject_match(self, match_id: str) -> dict[str, Any]:
+        """Reject a match submission."""
+        try:
+            client = await self._get_client()
+            response = await client.post(f"/api/v1/admin/matches/{match_id}/reject")
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def get_admin_reports(
+        self, status: str | None = None, limit: int = 50, offset: int = 0
+    ) -> dict[str, Any]:
+        """Get reports for admin review."""
+        try:
+            client = await self._get_client()
+            params = {"limit": limit, "offset": offset}
+            if status:
+                params["status"] = status
+            response = await client.get("/api/v1/admin/reports", params=params)
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def resolve_report(
+        self, report_id: str, action: str, details: str = ""
+    ) -> dict[str, Any]:
+        """Resolve a report."""
+        try:
+            client = await self._get_client()
+            response = await client.post(
+                f"/api/v1/admin/reports/{report_id}/resolve",
+                json={"action": action, "details": details},
+            )
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def get_admin_users(
+        self, limit: int = 50, offset: int = 0
+    ) -> dict[str, Any]:
+        """Get users for admin view."""
+        try:
+            client = await self._get_client()
+            response = await client.get(
+                "/api/v1/admin/users", params={"limit": limit, "offset": offset}
+            )
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def update_user_role(self, user_id: str, is_admin: bool) -> dict[str, Any]:
+        """Update a user's role."""
+        try:
+            client = await self._get_client()
+            response = await client.patch(
+                f"/api/v1/admin/users/{user_id}/role", json={"is_admin": is_admin}
+            )
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
+    async def ban_user(self, user_id: str, reason: str = "") -> dict[str, Any]:
+        """Ban or unban a user."""
+        try:
+            client = await self._get_client()
+            response = await client.post(
+                f"/api/v1/admin/users/{user_id}/ban", json={"reason": reason}
+            )
+            response.raise_for_status()
+            return response.json()
+        except httpx.ConnectError as e:
+            raise ConnectionError(f"Cannot connect to API: {e}") from e
+        except httpx.HTTPStatusError as e:
+            raise APIError(f"API error: {e.response.text}") from e
+        except httpx.HTTPError as e:
+            raise APIError(f"Request failed: {e}") from e
+
 
 # Global client instance
 _api_client: APIClient | None = None
diff --git a/cli/src/spotdl_cli/providers.py b/cli/src/spotdl_cli/providers.py
@@ -0,0 +1,166 @@
+"""Command palette providers for SpotDL CLI."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from textual.command import DiscoveryHit, Hit, Provider
+
+from spotdl_cli.core import APIError, get_api_client
+from spotdl_cli.theme import get_platform_icon
+
+if TYPE_CHECKING:
+    from spotdl_cli.app import SpotDLApp
+    from spotdl_cli.core.types import UniversalSearchResponse
+
+logger = logging.getLogger(__name__)
+
+
+class SpotDLCommandProvider(Provider):
+    """Universal search provider for the command palette."""
+
+    async def search(self, query: str) -> None:
+        """Search for songs, albums, artists, and playlists.
+
+        Args:
+            query: The search query from the command palette input.
+        """
+        if len(query) < 2:
+            return
+
+        matcher = self.matcher(query)
+        api_client = get_api_client()
+
+        try:
+            # We use universal_search returning 5 of each for quick palette lookups
+            results = await api_client.universal_search(query, limit=5)
+            
+            # Helper to create hits from entities
+            hits = []
+            
+            # Process artists
+            for artist in results.artists:
+                name = artist.get("name", "Unknown Artist")
+                platform = artist.get("platform", "spotify")
+                desc = "Artist \u2022 " + platform.title()
+                icon = get_platform_icon(platform)
+                
+                # Match against the artist name, and if it matches we show it
+                score = matcher.match(name)
+                if score > 0:
+                    hits.append(Hit(
+                        score,
+                        matcher.highlight(f"{icon} {name}"),
+                        self._make_navigate_callback("ArtistScreen", artist, "artist"),
+                        help=desc,
+                    ))
+
+            # Process albums
+            for album in results.albums:
+                name = album.get("name", "Unknown Album")
+                artist_name = album.get("artist_name", "") or album.get("artist", "")
+                platform = album.get("platform", "spotify")
+                desc = f"Album \u2022 {artist_name} \u2022 {platform.title()}"
+                icon = get_platform_icon(platform)
+                
+                # Match against album name or artist name
+                score = matcher.match(f"{name} {artist_name}")
+                if score > 0:
+                    hits.append(Hit(
+                        score,
+                        matcher.highlight(f"{icon} {name}"),
+                        self._make_navigate_callback("AlbumScreen", album, "album"),
+                        help=desc,
+                    ))
+
+            # Process playlists
+            for playlist in results.playlists:
+                name = playlist.get("name", "Unknown Playlist")
+                owner = playlist.get("owner_name", "")
+                platform = playlist.get("platform", "spotify")
+                desc = f"Playlist \u2022 {owner} \u2022 {platform.title()}"
+                icon = get_platform_icon(platform)
+                
+                score = matcher.match(f"{name} {owner}")
+                if score > 0:
+                    hits.append(Hit(
+                        score,
+                        matcher.highlight(f"{icon} {name}"),
+                        self._make_navigate_callback("PlaylistScreen", playlist, "playlist"),
+                        help=desc,
+                    ))
+
+            # Process tracks
+            for track in results.tracks:
+                name = track.get("name", "Unknown Track")
+                artist_name = track.get("artist_name", "") or track.get("artist", "")
+                platform = track.get("platform", "spotify")
+                desc = f"Track \u2022 {artist_name} \u2022 {platform.title()}"
+                icon = get_platform_icon(platform)
+                
+                score = matcher.match(f"{name} {artist_name}")
+                if score > 0:
+                    hits.append(Hit(
+                        score,
+                        matcher.highlight(f"{icon} {name}"),
+                        self._make_track_callback(track, "track"),
+                        help=desc,
+                    ))
+
+            # Sort by score and yield
+            hits.sort(key=lambda hit: hit.score, reverse=True)
+            for hit in hits:
+                yield hit
+
+        except APIError as e:
+            logger.error(f"Search failed in command palette: {e}")
+            yield Hit(1.0, f"Error: {e}", lambda: None, help="API Error")
+        except Exception as e:
+            logger.error(f"Error in command palette: {e}")
+
+    def _make_navigate_callback(self, screen_type: str, item: dict[str, Any], entity_type: str) -> Any:
+        """Create a callback that navigates to the given screen type."""
+        app = self.app
+        entity_id = item.get("id")
+        platform = item.get("platform", "spotify")
+        platform_id = item.get("platform_id", "")
+        
+        async def navigate() -> None:
+            if screen_type == "ArtistScreen":
+                from spotdl_cli.screens.artist import ArtistScreen
+                await app.push_screen(ArtistScreen(platform_id, platform, initial_data=item, entity_id=entity_id))
+            elif screen_type == "AlbumScreen":
+                from spotdl_cli.screens.album import AlbumScreen
+                await app.push_screen(AlbumScreen(platform_id, platform, initial_data=item, entity_id=entity_id))
+            elif screen_type == "PlaylistScreen":
+                from spotdl_cli.screens.playlist import PlaylistScreen
+                await app.push_screen(PlaylistScreen(platform_id, platform, initial_data=item, entity_id=entity_id))
+                
+        return navigate
+        
+    def _make_track_callback(self, track: dict[str, Any], entity_type: str) -> Any:
+        """Create a callback for track navigation."""
+        app = self.app
+        entity_id = track.get("id")
+        platform = track.get("platform", "spotify")
+        platform_id = track.get("platform_id", "")
+        
+        async def navigate() -> None:
+            from spotdl_cli.screens.track import TrackScreen
+            from spotdl_cli.core.types import Song, Platform
+            
+            song = Song(
+                name=track.get("name", "Unknown"),
+                artists=[track.get("artist_name") or track.get("artist", "Unknown")],
+                artist=track.get("artist_name") or track.get("artist", "Unknown"),
+                duration=track.get("duration", 0),
+                platform=Platform(platform),
+                platform_id=platform_id,
+                url=track.get("url", ""),
+                cover_url=track.get("cover_url"),
+            )
+            await app.push_screen(TrackScreen(song, platform_id, platform, initial_data=track, entity_id=entity_id))
+            
+        return navigate
+
diff --git a/cli/src/spotdl_cli/screens/__init__.py b/cli/src/spotdl_cli/screens/__init__.py
@@ -1,5 +1,7 @@
 """Textual screens for SpotDL CLI."""
 
+from spotdl_cli.screens.account import AccountScreen
+from spotdl_cli.screens.admin import AdminScreen
 from spotdl_cli.screens.album import AlbumScreen
 from spotdl_cli.screens.artist import ArtistScreen
 from spotdl_cli.screens.main import MainScreen
@@ -10,6 +12,8 @@
 from spotdl_cli.screens.track import TrackScreen
 
 __all__ = [
+    "AccountScreen",
+    "AdminScreen",
     "AlbumScreen",
     "ArtistScreen",
     "MainScreen",
diff --git a/cli/src/spotdl_cli/screens/account.py b/cli/src/spotdl_cli/screens/account.py
@@ -50,6 +50,7 @@ def compose(self) -> ComposeResult:
                 yield Static("", id="profile-role", classes="account-field")
                 yield Static("", id="profile-member-since", classes="account-field")
                 yield Static("", id="profile-status", classes="account-field")
+                yield Button("Admin Dashboard", id="admin-dashboard-btn", variant="success", classes="hidden")
 
             # Reputation card
             with Vertical(id="reputation-card", classes="card settings-group hidden"):
@@ -200,6 +201,12 @@ def _update_display(self) -> None:
         )
         role_text = "Admin" if is_admin else "User"
         self.query_one("#profile-role", Static).update(f"Role: {role_text}")
+        
+        admin_btn = self.query_one("#admin-dashboard-btn", Button)
+        if is_admin:
+            admin_btn.remove_class("hidden")
+        else:
+            admin_btn.add_class("hidden")
 
         if created_at:
             date_str = created_at[:10] if len(created_at) >= 10 else created_at
@@ -230,6 +237,9 @@ async def on_button_pressed(self, event: Button.Pressed) -> None:
         if button_id == "sign-in-btn":
             from spotdl_cli.screens.login import LoginScreen
             await self.app.push_screen(LoginScreen(), self._on_login_complete)
+        elif button_id == "admin-dashboard-btn":
+            from spotdl_cli.screens.admin import AdminScreen
+            await self.app.push_screen(AdminScreen())
         elif button_id == "change-password-btn":
             await self._change_password()
         elif button_id == "logout-btn":
diff --git a/cli/src/spotdl_cli/screens/admin.py b/cli/src/spotdl_cli/screens/admin.py
diff --git a/cli/tests/test_api_client.py b/cli/tests/test_api_client.py
diff --git a/frontend/railway.toml b/frontend/railway.toml
