Add a basic CLI (#5)

* Add CLI

* Remove debug

* Trim ws

* Setup and README

* typo fix

.gitignore

@@ -15,6 +15,7 @@ dist
 .vscode
 .theia
 .venv/
+venv/
 
 # Created by unit tests
 .pytest_cache/

README.md

@@ -1,3 +1,21 @@
 # TuneIn
 
-unnoficial python api for TuneIn
+unnoficial python api for TuneIn
+
+## Usage
+
+### From the command line
+
+tunein comes with a basic command line interface for searching. Output is availbe in both `json`` and table formats with the default being the later.
+
+Example:
+
+```
+tunein search "Radio paradise"
+```
+
+```
+tunein search "Radio paradise" --format json
+```
+
+Command line help is available with `tunein --help`

setup.py

@@ -50,8 +50,13 @@ def get_version():
     license='Apache-2.0',
     author="JarbasAi",
     url="https://github.com/OpenJarbas/tunein",
-    packages=["tunein"],
+    packages=["tunein", "tunein/subcommands"],
     include_package_data=True,
     install_requires=get_requirements("requirements.txt"),
     keywords='TuneIn internet radio',
+    entry_points={
+    'console_scripts': [
+        'tunein = tunein.cli:main',
+    ],
+},
 )

test/unittests/test_tunein.py

@@ -1,7 +1,13 @@
+import io
+import contextlib
+import json
+import sys
 import unittest
+from unittest.mock import patch
 
-from tunein import TuneIn, TuneInStation
 
+from tunein import TuneIn, TuneInStation
+from tunein.cli import Cli
 
 class TestTuneIn(unittest.TestCase):
 
@@ -10,6 +16,30 @@ def test_featured(self):
         print(stations)
         self.assertTrue(len(stations) > 0)
         self.assertTrue(isinstance(stations[0], TuneInStation))
+
+    def test_cli_table(self):
+        """Test the CLI output table format."""
+        testargs = ["tunein", "search", "kuow", "-f", "table"]
+        cli = Cli()
+        fhand = io.StringIO()
+        with patch.object(sys, 'argv', testargs):
+            with contextlib.redirect_stdout(fhand):
+                cli.parse_args()
+                cli.run()
+        self.assertTrue("KUOW" in fhand.getvalue())
+
+    def test_cli_json(self):
+        """Test the CLI output json format."""
+        testargs = ["tunein", "search", "kuow", "-f", "json"]
+        cli = Cli()
+        fhand = io.StringIO()
+        with patch.object(sys, 'argv', testargs):
+            with contextlib.redirect_stdout(fhand):
+                cli.parse_args()
+                cli.run()
+        json_loaded = json.loads(fhand.getvalue())
+        kuow = [i for i in json_loaded if i["title"] == "KUOW-FM"]
+        self.assertTrue(len(kuow) == 1)
 
 
 if __name__ == '__main__':

tunein/__init__.py

@@ -10,7 +10,7 @@ def __init__(self, raw):
     @property
     def title(self):
         return self.raw.get("title", "")
-    
+
     @property
     def artist(self):
         return self.raw.get("artist", "")
@@ -39,6 +39,18 @@ def __str__(self):
     def __repr__(self):
         return self.title
 
+    @property
+    def dict(self):
+        """Return a dict representation of the station."""
+        return {
+            "artist": self.artist,
+            "description": self.description,
+            "image": self.image,
+            "match": self.match(),
+            "stream": self.stream,
+            "title": self.title,
+        }
+
 
 class TuneIn:
     search_url = "http://opml.radiotime.com/Search.ashx"
@@ -67,7 +79,7 @@ def featured():
 
     @staticmethod
     def search(query):
-        res = requests.post(TuneIn.search_url, data={"query": query})
+        res = requests.post(TuneIn.search_url, data={"query": query, "formats": "mp3,aac,ogg,html,hls"})
         return list(TuneIn._get_stations(res, query))
 
     @staticmethod

tunein/cli.py

@@ -0,0 +1,63 @@
+"""The CLI entrypoint for tunein."""
+
+import argparse
+
+from tunein import subcommands
+
+class Cli:
+    """The CLI entrypoint for tunein."""
+
+    def __init__(self) -> None:
+        """Initialize the CLI entrypoint."""
+        self._args: argparse.Namespace
+
+    def parse_args(self) -> None:
+        """Parse the command line arguments."""
+        parser = argparse.ArgumentParser(
+            description="unnoficial python api for TuneIn.",
+        )
+
+        subparsers = parser.add_subparsers(
+            title="Commands",
+            dest="subcommand",
+            metavar="",
+            required=True,
+        )
+
+        search = subparsers.add_parser(
+            "search",
+            help="Search tunein for stations",
+        )
+
+        search.add_argument(
+            "station",
+            help="Station to search for",
+        )
+
+        search.add_argument(
+            "-f", "--format",
+            choices=["json", "table"],
+            default="table",
+            help="Output format",
+        )
+
+        self._args = parser.parse_args()
+
+    def run(self) -> None:
+        """Run the CLI."""
+        subcommand_cls = getattr(subcommands, self._args.subcommand.capitalize())
+        subcommand = subcommand_cls(args=self._args)
+        subcommand.run()
+
+def main() -> None:
+    """Run the CLI."""
+    cli = Cli()
+    cli.parse_args()
+    cli.run()
+
+
+if __name__ == "__main__":
+    main()
+
+
+

tunein/subcommands/__init__.py

@@ -0,0 +1,3 @@
+"""The subcommands for tunein."""
+
+from .search import Search

tunein/subcommands/search.py

@@ -0,0 +1,108 @@
+"""The search subcommand for tunein."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import sys
+
+from tunein import TuneIn
+
+class Ansi:
+    """ANSI escape codes."""
+
+    BLUE = "\x1B[34m"
+    BOLD = "\x1B[1m"
+    CYAN = "\x1B[36m"
+    GREEN = "\x1B[32m"
+    ITALIC = "\x1B[3m"
+    MAGENTA = "\x1B[35m"
+    RED = "\x1B[31m"
+    RESET = "\x1B[0m"
+    REVERSED = "\x1B[7m"
+    UNDERLINE = "\x1B[4m"
+    WHITE = "\x1B[37m"
+    YELLOW = "\x1B[33m"
+    GREY = "\x1B[90m"
+
+NOPRINT_TRANS_TABLE = {
+    i: None for i in range(0, sys.maxunicode + 1) if not chr(i).isprintable()
+}
+
+class Search:
+    """The search subcommand for tunein."""
+
+    def __init__(self: Search, args: argparse.Namespace) -> None:
+        """Initialize the search subcommand."""
+        self._args: argparse.Namespace = args
+
+    def run(self: Search) -> None:
+        """Run the search subcommand."""
+        tunein = TuneIn()
+        results = tunein.search(self._args.station)
+        stations = [station.dict for station in results]
+        if not stations:
+            print(f"No results for {self._args.station}")
+            sys.exit(1)
+        stations.sort(key=lambda x: x["match"], reverse=True)
+        for station in stations:
+            station["title"] = self._printable(station["title"])
+            station["artist"] = self._printable(station["artist"])
+            station["description"] = self._printable(station["description"])
+
+        if self._args.format == "json":
+            print(json.dumps(stations, indent=4))
+        elif self._args.format == "table":
+            max_widths = {}
+            columns = ["title", "artist", "description"]
+            for column in columns:
+                max_width = max(len(str(station[column])) for station in stations)
+                if column == "description":
+                    term_width = shutil.get_terminal_size().columns
+                    remaining = term_width - max_widths["title"] - max_widths["artist"] - 2
+                    max_width = min(max_width, remaining) 
+                max_widths[column] = max_width
+
+            print(" ".join(column.ljust(max_widths[column]).capitalize() for column in columns))
+            print(" ".join("-" * max_widths[column] for column in columns))
+            for station in stations:
+                line_parts = []
+                # title as link
+                link = self._term_link(station.get("stream"), station["title"])
+                line_parts.append(f"{link}{' '*(max_widths['title']-len(station['title']))}")
+                # artist
+                line_parts.append(str(station["artist"]).ljust(max_widths["artist"]))
+                # description clipped
+                line_parts.append(str(station["description"])[:max_widths["description"]])
+                print(" ".join(line_parts))
+
+    @staticmethod
+    def _term_link(uri: str, label: str) -> str:
+        """Return a link.
+
+        Args:
+            uri: The URI to link to
+            label: The label to use for the link
+        Returns:
+            The link
+        """
+        parameters = ""
+
+        # OSC 8 ; params ; URI ST <name> OSC 8 ;; ST
+        escape_mask = "\x1b]8;{};{}\x1b\\{}\x1b]8;;\x1b\\"
+        link_str = escape_mask.format(parameters, uri, label)
+        return f"{Ansi.BLUE}{link_str}{Ansi.RESET}"
+
+    @staticmethod
+    def _printable(string: str) -> str:
+        """Replace non-printable characters in a string.
+        
+        Args:
+            string: The string to replace non-printable characters in.
+        Returns:
+            The string with non-printable characters replaced.
+        """
+        return string.translate(NOPRINT_TRANS_TABLE)
+
+