feat: Make streamrip Termux compatible and fix bugs

pyproject.toml

@@ -23,9 +23,7 @@ python = ">=3.10 <4.0"
 mutagen = "^1.45.1"
 tomlkit = "^0.7.2"
 pathvalidate = "^2.4.1"
-simple-term-menu = { version = "^1.2.1", platform = 'darwin|linux' }
-pick = { version = "^2", platform = 'win32|cygwin' }
-windows-curses = { version = "^2.2.0", platform = 'win32|cygwin' }
+python-cli-menu = "*"
 Pillow = ">=9,<11"
 deezer-py = "1.3.6"
 pycryptodomex = "^3.10.1"
@@ -40,6 +38,7 @@ pytest-asyncio = "^0.21.1"
 rich = "^13.6.0"
 click-help-colors = "^0.9.2"
 certifi = { version = "^2025.1.31", optional = true }
+yt-dlp = "^2023.11.16"
 
 [tool.poetry.urls]
 "Bug Reports" = "https://github.com/nathom/streamrip/issues"

streamrip/client/downloadable.py

@@ -432,3 +432,53 @@ async def concat_audio_files(paths: list[str], out: str, ext: str, max_files_ope
 
     # Recurse on remaining batches
     await concat_audio_files(outpaths, out, ext)
+
+
+class YoutubeDLTrack(Downloadable):
+    def __init__(
+        self,
+        url: str,
+        config,
+        folder: str,
+        playlist_name: str,
+        track_number: int,
+        db,
+    ):
+        self.url = url
+        self.config = config
+        self.folder = folder
+        self.playlist_name = playlist_name
+        self.track_number = track_number
+        self.db = db
+        self.extension = "flac"
+
+    async def _download(self, path: str, callback: Callable[[int], None]):
+        import yt_dlp
+
+        ydl_opts = {
+            "format": "bestaudio/best",
+            "outtmpl": path,
+            "quiet": True,
+            "postprocessors": [
+                {
+                    "key": "FFmpegExtractAudio",
+                    "preferredcodec": "flac",
+                    "preferredquality": "5",
+                }
+            ],
+            "progress_hooks": [lambda d: callback(d["downloaded_bytes"])],
+        }
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([self.url])
+
+    async def size(self) -> int:
+        import yt_dlp
+
+        ydl_opts = {
+            "quiet": True,
+        }
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(self.url, download=False)
+            if info:
+                return info.get("filesize") or info.get("filesize_approx") or 0
+        return 0

streamrip/config.py

@@ -231,13 +231,10 @@ class MiscConfig:
 
 
 HOME = Path.home()
-DEFAULT_DOWNLOADS_FOLDER = os.path.join(HOME, "StreamripDownloads")
-DEFAULT_DOWNLOADS_DB_PATH = os.path.join(APP_DIR, "downloads.db")
-DEFAULT_FAILED_DOWNLOADS_DB_PATH = os.path.join(APP_DIR, "failed_downloads.db")
-DEFAULT_YOUTUBE_VIDEO_DOWNLOADS_FOLDER = os.path.join(
-    DEFAULT_DOWNLOADS_FOLDER,
-    "YouTubeVideos",
-)
+DEFAULT_DOWNLOADS_FOLDER = HOME / "StreamripDownloads"
+DEFAULT_DOWNLOADS_DB_PATH = Path(APP_DIR) / "downloads.db"
+DEFAULT_FAILED_DOWNLOADS_DB_PATH = Path(APP_DIR) / "failed_downloads.db"
+DEFAULT_YOUTUBE_VIDEO_DOWNLOADS_FOLDER = DEFAULT_DOWNLOADS_FOLDER / "YouTubeVideos"
 BLANK_CONFIG_PATH = os.path.join(os.path.dirname(__file__), "config.toml")
 assert os.path.isfile(BLANK_CONFIG_PATH), "Template config not found"
 
@@ -422,10 +419,12 @@ def set_user_defaults(path: str, /):
 
 
 def toml_set_user_defaults(toml: TOMLDocument):
-    toml["downloads"]["folder"] = DEFAULT_DOWNLOADS_FOLDER  # type: ignore
-    toml["database"]["downloads_path"] = DEFAULT_DOWNLOADS_DB_PATH  # type: ignore
-    toml["database"]["failed_downloads_path"] = DEFAULT_FAILED_DOWNLOADS_DB_PATH  # type: ignore
-    toml["youtube"]["video_downloads_folder"] = DEFAULT_YOUTUBE_VIDEO_DOWNLOADS_FOLDER  # type: ignore
+    toml["downloads"]["folder"] = str(DEFAULT_DOWNLOADS_FOLDER)  # type: ignore
+    toml["database"]["downloads_path"] = str(DEFAULT_DOWNLOADS_DB_PATH)  # type: ignore
+    toml["database"]["failed_downloads_path"] = str(DEFAULT_FAILED_DOWNLOADS_DB_PATH)  # type: ignore
+    toml["youtube"]["video_downloads_folder"] = str(
+        DEFAULT_YOUTUBE_VIDEO_DOWNLOADS_FOLDER
+    )  # type: ignore
 
 
 def _get_dict_keys_r(d: dict) -> set[tuple]:

streamrip/filepath_utils.py

@@ -1,15 +1,14 @@
+import textwrap
+from pathlib import Path
 from string import printable
 
-from pathvalidate import sanitize_filename, sanitize_filepath  # type: ignore
+from pathvalidate import sanitize_filename, sanitize_filepath
 
 ALLOWED_CHARS = set(printable)
 
 
-# TODO: remove this when new pathvalidate release arrives with https://github.com/thombashi/pathvalidate/pull/48
-def truncate_str(text: str) -> str:
-    str_bytes = text.encode()
-    str_bytes = str_bytes[:255]
-    return str_bytes.decode(errors="ignore")
+def truncate_str(text: str, max_len: int = 255) -> str:
+    return textwrap.shorten(text, width=max_len, placeholder="...")
 
 
 def clean_filename(fn: str, restrict: bool = False) -> str:
@@ -20,9 +19,9 @@ def clean_filename(fn: str, restrict: bool = False) -> str:
     return path
 
 
-def clean_filepath(fn: str, restrict: bool = False) -> str:
-    path = str(sanitize_filepath(fn))
+def clean_filepath(fn: str, restrict: bool = False) -> Path:
+    path = Path(sanitize_filepath(fn))
     if restrict:
-        path = "".join(c for c in path if c in ALLOWED_CHARS)
+        path = Path("".join(c for c in str(path) if c in ALLOWED_CHARS))
 
     return path

streamrip/media/album.py

@@ -2,6 +2,7 @@
 import logging
 import os
 from dataclasses import dataclass
+from pathlib import Path
 
 from .. import progress
 from ..client import Client
@@ -106,13 +107,14 @@ async def resolve(self) -> Album | None:
         logger.debug("Pending tracks: %s", pending_tracks)
         return Album(meta, pending_tracks, self.config, album_folder, self.db)
 
-    def _album_folder(self, parent: str, meta: AlbumMetadata) -> str:
+    def _album_folder(self, parent: str, meta: AlbumMetadata) -> Path:
         config = self.config.session
+        parent_path = Path(parent)
         if config.downloads.source_subdirectories:
-            parent = os.path.join(parent, self.client.source.capitalize())
+            parent_path = parent_path / self.client.source.capitalize()
         formatter = config.filepaths.folder_format
         folder = clean_filepath(
             meta.format_folder_path(formatter), config.filepaths.restrict_characters
         )
 
-        return os.path.join(parent, folder)
+        return parent_path / folder

streamrip/media/playlist.py

@@ -6,6 +6,7 @@
 import re
 from contextlib import ExitStack
 from dataclasses import dataclass
+from pathlib import Path
 
 import aiohttp
 from rich.text import Text
@@ -28,6 +29,7 @@
 from .artwork import download_artwork
 from .media import Media, Pending
 from .track import Track
+from ..client.downloadable import YoutubeDLTrack
 
 logger = logging.getLogger("streamrip")
 
@@ -170,8 +172,8 @@ async def resolve(self) -> Playlist | None:
             logger.error(f"Error creating playlist: {e}")
             return None
         name = meta.name
-        parent = self.config.session.downloads.folder
-        folder = os.path.join(parent, clean_filepath(name))
+        parent = Path(self.config.session.downloads.folder)
+        folder = parent / clean_filepath(name)
         tracks = [
             PendingPlaylistTrack(
                 id,
@@ -242,32 +244,53 @@ def callback():
                 requests.append(self._make_query(f"{title} {artist}", s, callback))
             results: list[tuple[str | None, bool]] = await asyncio.gather(*requests)
 
-        parent = self.config.session.downloads.folder
-        folder = os.path.join(parent, clean_filepath(playlist_title))
+        parent = Path(self.config.session.downloads.folder)
+        folder = parent / clean_filepath(playlist_title)
 
         pending_tracks = []
         for pos, (id, from_fallback) in enumerate(results, start=1):
             if id is None:
                 logger.warning(f"No results found for {titles_artists[pos-1]}")
                 continue
 
-            if from_fallback:
+            if from_fallback and "youtube.com" in id:
+                pending_tracks.append(
+                    YoutubeDLTrack(
+                        id,
+                        self.config,
+                        folder,
+                        playlist_name=playlist_title,
+                        track_number=pos,
+                        db=self.db,
+                    )
+                )
+            elif from_fallback:
                 assert self.fallback_client is not None
                 client = self.fallback_client
+                pending_tracks.append(
+                    PendingPlaylistTrack(
+                        id,
+                        client,
+                        self.config,
+                        folder,
+                        playlist_title,
+                        pos,
+                        self.db,
+                    ),
+                )
             else:
                 client = self.client
-
-            pending_tracks.append(
-                PendingPlaylistTrack(
-                    id,
-                    client,
-                    self.config,
-                    folder,
-                    playlist_title,
-                    pos,
-                    self.db,
-                ),
-            )
+                pending_tracks.append(
+                    PendingPlaylistTrack(
+                        id,
+                        client,
+                        self.config,
+                        folder,
+                        playlist_title,
+                        pos,
+                        self.db,
+                    ),
+                )
 
         return Playlist(playlist_title, self.config, self.client, pending_tracks)
 
@@ -322,8 +345,23 @@ async def _make_query(
                 ), True
 
             logger.debug(f"No result found for {query} on {self.client.source}")
-            search_status.failed += 1
-        return None, True
+
+        try:
+            import yt_dlp
+
+            with yt_dlp.YoutubeDL(
+                {"format": "bestaudio", "noplaylist": True, "quiet": True}
+            ) as ydl:
+                info = ydl.extract_info(f"ytsearch:{query}", download=False)
+                if info and "entries" in info and info["entries"]:
+                    video_url = info["entries"][0]["webpage_url"]
+                    search_status.found += 1
+                    return video_url, True
+        except Exception as e:
+            logger.error(f"Error searching on YouTube: {e}")
+
+        search_status.failed += 1
+        return None, False
 
     async def _parse_lastfm_playlist(
         self,

streamrip/metadata/album.py

@@ -3,6 +3,7 @@
 import logging
 import re
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Optional
 
 from ..filepath_utils import clean_filename, clean_filepath
@@ -62,7 +63,7 @@ def get_copyright(self) -> str | None:
         _copyright = re.sub(r"(?i)\(C\)", COPYRIGHT, _copyright)
         return _copyright
 
-    def format_folder_path(self, formatter: str) -> str:
+    def format_folder_path(self, formatter: str) -> Path:
         # Available keys: "albumartist", "title", "year", "bit_depth", "sampling_rate",
         # "id", and "albumcomposer",
 

streamrip/rip/main.py

@@ -191,47 +191,24 @@ async def search_interactive(self, source: str, media_type: str, query: str):
                 return
             search_results = SearchResults.from_pages(source, media_type, pages)
 
-        if platform.system() == "Windows":  # simple term menu not supported for windows
-            from pick import pick
-
-            choices = pick(
-                search_results.results,
-                title=(
-                    f"{source.capitalize()} {media_type} search.\n"
-                    "Press SPACE to select, RETURN to download, CTRL-C to exit."
-                ),
-                multiselect=True,
-                min_selection_count=1,
-            )
-            assert isinstance(choices, list)
-
-            await self.add_all_by_id(
-                [(source, media_type, item.id) for item, _ in choices],
-            )
-
+        from python_cli_menu import Menu
+
+        menu = Menu(
+            search_results.summaries(),
+            title=(
+                f"Results for {media_type} '{query}' from {source.capitalize()}\n"
+                "SPACE - select, ENTER - download, ESC - exit"
+            ),
+            multi_select=True,
+        )
+        chosen_ind = menu.show()
+        if chosen_ind is None:
+            console.print("[yellow]No items chosen. Exiting.")
         else:
-            from simple_term_menu import TerminalMenu
-
-            menu = TerminalMenu(
-                search_results.summaries(),
-                preview_command=search_results.preview,
-                preview_size=0.5,
-                title=(
-                    f"Results for {media_type} '{query}' from {source.capitalize()}\n"
-                    "SPACE - select, ENTER - download, ESC - exit"
-                ),
-                cycle_cursor=True,
-                clear_screen=True,
-                multi_select=True,
+            choices = search_results.get_choices(chosen_ind)
+            await self.add_all_by_id(
+                [(source, item.media_type(), item.id) for item in choices],
             )
-            chosen_ind = menu.show()
-            if chosen_ind is None:
-                console.print("[yellow]No items chosen. Exiting.")
-            else:
-                choices = search_results.get_choices(chosen_ind)
-                await self.add_all_by_id(
-                    [(source, item.media_type(), item.id) for item in choices],
-                )
 
     async def search_take_first(self, source: str, media_type: str, query: str):
         client = await self.get_logged_in_client(source)