added basic subscribe/unsubscribe support; some refactoring to isolat…

…e functionality for reuse; rudimentary OPML reading/writing via it being the native subscription list format
mwhickson · Nov 23, 2024 · 0ee18ec · 0ee18ec
1 parent 13ad413
commit 0ee18ec
Show file tree

Hide file tree

Showing 6 changed files with 203 additions and 61 deletions.
diff --git a/.gitignore b/.gitignore
@@ -286,4 +286,6 @@ poetry.toml
 # LSP config files
 pyrightconfig.json
 
-# End of https://www.toptal.com/developers/gitignore/api/python,pycharm
+# End of https://www.toptal.com/developers/gitignore/api/python,pycharm
+
+debug.log
diff --git a/README.md b/README.md
@@ -15,6 +15,7 @@ This implementation utilizes Python and Textual -- but no ChatGPT.
 - discover podcasts through iTunes-based search
 - play podcast episodes directly from source
 - pause podcasts during play
+- maintain a podcast subscription list
 - *more (still in development)*
 
 ## Installation and Running
@@ -54,6 +55,9 @@ Also provided are batch (`tuipod.bat`) and shell (`tuipod.sh`) files to simplify
 - `TAB` and `SHIFT`+`TAB` will move the cursor focus between sections (e.g. search, podcast list, and episode list)
 - `CTRL`+`Q` will quit the application
 - `CTRL`+`P` will show the textual command palette
+- when a podcast is selected:
+  - `S` will subscribe to the podcast
+  - `U` will unsubscribe from the podcast
 - when an episode is selected:
   - `I` will show the episode information
   - `SPACE` will toggle between episode playing/paused state

diff --git a/subscriptions.opml b/subscriptions.opml
@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="utf-8" standalone="no"?>
+<opml version="1.0">
+<body>
+<outline text="feeds">
+<outline text="Adventures in Lollygagging" xmlUrl="https://feeds.transistor.fm/adventures-in-lollygagging-4b1c91f1-9282-46d3-84ab-10ac9187ea0a" type="rss" />
+</outline>
+</body>
+</opml>
diff --git a/tuipod/models/subscription_list.py b/tuipod/models/subscription_list.py
@@ -0,0 +1,57 @@
+from os.path import exists
+import xml.etree.ElementTree as ET
+import xml.sax.saxutils as saxu
+
+from tuipod.models.podcast import Podcast
+
+class SubscriptionList:
+
+    SUBSCRIPTION_FILE = "subscriptions.opml"
+
+    def __init__(self) -> None:
+        self.podcasts = []
+
+    def add_podcast(self, p: Podcast) -> None:
+        self.podcasts.append(p)
+
+    def remove_podcast(self, url: str) -> None:
+        for p in self.podcasts:
+            if p.url == url:
+                self.podcasts.remove(p)
+                break
+
+    def retrieve(self) -> []:
+        self.podcasts = []
+
+        if exists(self.SUBSCRIPTION_FILE):
+            with open(self.SUBSCRIPTION_FILE, "rt", encoding="utf-8") as subscription_file:
+                contents = subscription_file.readlines()
+                subscription_file.close()
+
+            doc = ET.fromstringlist(contents)
+
+            for item in doc.iter("outline"):
+                title = item.get("text")
+                url = item.get("xmlUrl")
+                if not title is None and not url is None:
+                    self.add_podcast(Podcast(title, url, ""))
+
+    def persist(self) -> None:
+        with open(self.SUBSCRIPTION_FILE, "wt", encoding="utf-8") as subscription_file:
+            lines = []
+            lines.append('<?xml version="1.0" encoding="utf-8" standalone="no"?>\n')
+            lines.append('<opml version="1.0">\n')
+            lines.append('<body>\n')
+            lines.append('<outline text="feeds">\n')
+
+            for p in self.podcasts:
+                escaped_title = saxu.escape(p.title)
+                lines.append('<outline text="{0}" xmlUrl="{1}" type="rss" />\n'.format(escaped_title, p.url))
+
+            lines.append('</outline>\n')
+            lines.append('</body>\n')
+            lines.append('</opml>\n')
+
+            subscription_file.writelines(lines)
+            subscription_file.flush()
+            subscription_file.close()
diff --git a/tuipod/ui/about_info.py b/tuipod/ui/about_info.py
@@ -69,6 +69,8 @@ class AboutInfoScreen(ModalScreen):
 - `CTRL` + `Q` - quit the application
 - `D` - toggle dark mode
 - `I` - show episode information
+- `S` - subscribe to highlighted podcast
+- `U` - unsubscribe from highlighted podcast
 - `SPACE` - play/pause an episode after selection
 - `TAB` / `SHIFT` + `TAB` - move cursor from section to section
 

diff --git a/tuipod/ui/podcast_app.py b/tuipod/ui/podcast_app.py
@@ -1,11 +1,13 @@
 import json
+import uuid
 
 from textual import on
 from textual.app import App, ComposeResult
 from textual.binding import Binding
 from textual.widgets import Button, DataTable, Header, Input, Static
 
 from tuipod.models.search import Search
+from tuipod.models.subscription_list import SubscriptionList
 from tuipod.ui.about_info import AboutInfoScreen
 from tuipod.ui.episode_info import EpisodeInfoScreen
 from tuipod.ui.episode_list import EpisodeList
@@ -14,6 +16,10 @@
 from tuipod.ui.podcast_player import PodcastPlayer
 from tuipod.ui.search_input import SearchInput
 
+# DEBUG:
+# import logging
+# logging.basicConfig(filename="debug.log", filemode="w", level="NOTSET")
+
 APPLICATION_NAME = "tuipod"
 APPLICATION_VERSION = "2024-11-22.5c24b1e90d6c4ae28faceec6bbcdff7a"
 
@@ -24,19 +30,21 @@ class PodcastApp(App):
         # don't let search swallow input (but don't prioritize standard text entry characters to hamper search)
         Binding("ctrl+q", "quit_application", "Quit application", priority=True),
 
-        # dupes, but lets us avoid the need to CTRL chord keys for the most part (these will be 'undocumented' to avoid confusion re: the conditions necessary for these to work)
+        # may dupe, but lets us avoid the need to CTRL chord keys for some functions (these will be 'undocumented' to avoid confusion re: the conditions necessary for these to work)
         Binding("space", "toggle_play", "Play/Pause"),
         Binding("d", "toggle_dark", "Toggle dark mode"),
         Binding("i", "display_info", "Display information"),
         Binding("q", "quit_application", "Quit application"),
-        #TODO: Binding("s", "subscribe_to_podcast", "Subscribe to Podcast")
+        Binding("s", "subscribe_to_podcast", "Subscribe to Podcast"),
+        Binding("u", "unsubscribe_from_podcast", "Unsubscribe from Podcast")
     ]
     TITLE = APPLICATION_NAME
     SUB_TITLE = "version {0}".format(APPLICATION_VERSION)
 
     def __init__(self):
         super().__init__()
         self.searcher = Search("")
+        self.subscriptions = SubscriptionList()
         self.podcasts = []
         self.current_podcast = None
         self.current_episode = None
@@ -49,105 +57,157 @@ def compose(self) -> ComposeResult:
         yield EpisodeList()
         yield PodcastPlayer()
 
-    @on(Input.Submitted)
-    async def action_submit(self, event: Input.Submitted) -> None:
+    async def on_mount(self) -> None:
+        self.subscriptions.retrieve()
+        if len(self.subscriptions.podcasts) > 0:
+            await self._refresh_podcast_list("")
+
+    async def _refresh_podcast_list(self, search_term: str) -> None:
         podcast_list = self.query_one(PodcastList)
+        episode_list = self.query_one(EpisodeList)
+
         table = podcast_list.query_one(DataTable)
         table.loading = True
 
-        search_input: Input = event.input
-        search_term = search_input.value
+        episodes_table = episode_list.query_one(DataTable)
 
-        try:
+        table.clear()
+        episodes_table.clear()
 
-            self.podcasts = await self.searcher.search(search_term)
+        if len(self.subscriptions.podcasts) > 0:
+            self.podcasts = self.subscriptions.podcasts
+            for p in self.subscriptions.podcasts:
+                row_key = json.dumps((p.id, p.url))
+                table.add_row(p.title, key=row_key)
 
-            table.clear()
+        if search_term.strip() != "":
+            try:
 
-            if len(self.podcasts) > 0:
-                for podcast in self.podcasts:
-                    row_key = json.dumps((podcast.id, podcast.url))
-                    table.add_row(podcast.title, key=row_key)
+                self.podcasts = await self.searcher.search(search_term)
 
-                table.focus()
-            else:
-                table.add_row("no results")
+                if len(self.podcasts) > 0:
+                    for podcast in self.podcasts:
+                        row_key = json.dumps((podcast.id, podcast.url))
+                        if not row_key in table.rows.keys():
+                            table.add_row(podcast.title, key=row_key)
 
-        except Exception as err:
-            table.add_row("error occurred")
-            self.app.push_screen(ErrorInfoScreen(str(err)))
+                    table.focus()
+                else:
+                    table.add_row("no results")
+
+            except Exception as err:
+                table.add_row("error occurred")
+                self.app.push_screen(ErrorInfoScreen(str(err)))
 
         table.loading = False
 
-    @on(DataTable.RowSelected)
-    def action_row_selected(self, event: DataTable.RowSelected) -> None:
-        triggering_table = event.data_table
-        row_keys = json.loads(event.row_key.value)
+    @on(Input.Submitted)
+    async def action_submit(self, event: Input.Submitted) -> None:
+        search_input: Input = event.input
+        search_term = search_input.value
+        await self._refresh_podcast_list(search_term)
 
-        try:
+    def _set_player_button_status(self, mode: str):
+        player: PodcastPlayer = self.query_one(PodcastPlayer)
+        play_button: Button = player.query_one("#playButton")
 
-            if triggering_table.id == "PodcastList":
+        if mode == "playing":
+            play_button.label = "playing"
+            play_button.styles.background = "green"
+            play_button.styles.color = "white"
+        elif mode == "paused":
+            play_button.label = "paused"
+            play_button.styles.background = "red"
+            play_button.styles.color = "white"
+        else:
+            play_button.label = "..."
+            play_button.styles.background = "blue"
+            play_button.styles.color = "white"
+
+    def _action_podcast_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
+        k = event.row_key
+        if not k is None:
+            v = k.value
+            if not v is None:
+                row_keys = json.loads(v)
                 podcast_id = row_keys[0]
                 for p in self.podcasts:
                     if p.id == podcast_id:
                         self.current_podcast = p
                         break
 
-                self.current_podcast.get_episode_list()
-
-                episode_list = self.query_one(EpisodeList)
-                table = episode_list.query_one(DataTable)
-                table.clear()
-
-                for e in self.current_podcast.episodes:
-                    row_key = json.dumps((e.id, e.url))
-                    table.add_row(e.title, e.duration, e.pubdate, key=row_key)
-
-                table.focus()
-            elif triggering_table.id == "EpisodeList":
-                playing_episode = self.current_episode
-
+    def _action_episode_row_highlighted(self, event: DataTable.RowSelected) -> None:
+        k = event.row_key
+        if not k is None:
+            v = k.value
+            if not v is None:
+                row_keys = json.loads(v)
                 episode_id = row_keys[0]
                 for e in self.current_podcast.episodes:
                     if e.id == episode_id:
                         self.current_episode = e
                         break
 
-                player: PodcastPlayer = self.query_one(PodcastPlayer)
-                player_title: Static = player.query_one("#playerTitleText")
-                player_title.update(self.current_episode.title)
+    def _action_podcast_row_selected(self, event: DataTable.RowSelected) -> None:
+        try:
+            self.current_podcast.get_episode_list()
 
-                if playing_episode and playing_episode.is_playing:
-                    playing_episode.stop_episode()
+            episode_list = self.query_one(EpisodeList)
+            table = episode_list.query_one(DataTable)
+            table.loading = True
+            table.clear()
 
-                self.current_episode.play_episode()
+            for e in self.current_podcast.episodes:
+                row_key = json.dumps((e.id, e.url))
+                table.add_row(e.title, e.duration, e.pubdate, key=row_key)
+
+            table.focus()
+        except Exception as err:
+            self.app.push_screen(ErrorInfoScreen(str(err)))
+
+        table.loading = False
+
+    def _action_episode_row_selected(self, event: DataTable.RowSelected) -> None:
+        try:
+            playing_episode = self.current_episode
 
-                play_button: Button = player.query_one("#playButton")
-                play_button.label = "pause"
-                play_button.styles.background = "green"
-                play_button.styles.color = "white"
+            player: PodcastPlayer = self.query_one(PodcastPlayer)
+            player_title: Static = player.query_one("#playerTitleText")
+            player_title.update(self.current_episode.title)
 
+            if playing_episode and playing_episode.is_playing:
+                playing_episode.stop_episode()
+
+            self.current_episode.play_episode()
+            self._set_player_button_status("playing")
         except Exception as err:
             self.app.push_screen(ErrorInfoScreen(str(err)))
 
+    @on(DataTable.RowHighlighted)
+    def action_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
+        if event.data_table.id == "PodcastList":
+            self._action_podcast_row_highlighted(event)
+        elif event.data_table.id == "EpisodeList":
+            self._action_episode_row_highlighted(event)
+
+    @on(DataTable.RowSelected)
+    def action_row_selected(self, event: DataTable.RowSelected) -> None:
+        if event.data_table.id == "PodcastList":
+            self._action_podcast_row_selected(event)
+        elif event.data_table.id == "EpisodeList":
+            self._action_episode_row_selected(event)
+
     def action_toggle_dark(self) -> None:
         self.dark = not self.dark
 
     def action_toggle_play(self) -> None:
-        player: PodcastPlayer = self.query_one(PodcastPlayer)
-        play_button: Button = player.query_one("#playButton")
-
         if not self.current_episode is None:
             if self.current_episode.is_playing:
                 self.current_episode.stop_episode()
-                play_button.label = "play"
-                play_button.styles.background = "red"
-                play_button.styles.color = "white"
+                self._set_player_button_status("paused")
             else:
                 self.current_episode.play_episode()
-                play_button.label = "pause"
-                play_button.styles.background = "green"
-                play_button.styles.color = "white"
+                self._set_player_button_status("playing")
 
     def action_display_about(self) -> None:
         self.app.push_screen(AboutInfoScreen())
@@ -159,5 +219,14 @@ def action_display_info(self) -> None:
     def action_quit_application(self) -> None:
         self.exit()
 
-    # def action_subscribe_to_podcast(self) -> None:
-    #     pass
+    async def action_subscribe_to_podcast(self) -> None:
+        if not self.current_podcast is None:
+            self.subscriptions.add_podcast(self.current_podcast)
+            self.subscriptions.persist()
+            await self._refresh_podcast_list(self.searcher.search_text)
+
+    async def action_unsubscribe_from_podcast(self) -> None:
+        if not self.current_podcast is None:
+            self.subscriptions.remove_podcast(self.current_podcast.url)
+            self.subscriptions.persist()
+            await self._refresh_podcast_list(self.searcher.search_text)