diff --git a/.gitignore b/.gitignore index 9595be5..f77d6bd 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,7 @@ dist .vscode .theia .venv/ +venv/ # Created by unit tests .pytest_cache/ diff --git a/README.md b/README.md index a058e6c..910f764 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,21 @@ # TuneIn -unnoficial python api for TuneIn \ No newline at end of file +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` \ No newline at end of file diff --git a/setup.py b/setup.py index 68bc721..0a54ee1 100644 --- a/setup.py +++ b/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', + ], +}, ) diff --git a/test/unittests/test_tunein.py b/test/unittests/test_tunein.py index 2dab0ab..ef848b3 100644 --- a/test/unittests/test_tunein.py +++ b/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__': diff --git a/tunein/__init__.py b/tunein/__init__.py index 7fa36d9..27776fd 100644 --- a/tunein/__init__.py +++ b/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 diff --git a/tunein/cli.py b/tunein/cli.py new file mode 100644 index 0000000..c9c4c3a --- /dev/null +++ b/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() + + + diff --git a/tunein/subcommands/__init__.py b/tunein/subcommands/__init__.py new file mode 100644 index 0000000..e57ecec --- /dev/null +++ b/tunein/subcommands/__init__.py @@ -0,0 +1,3 @@ +"""The subcommands for tunein.""" + +from .search import Search diff --git a/tunein/subcommands/search.py b/tunein/subcommands/search.py new file mode 100644 index 0000000..f09c89b --- /dev/null +++ b/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 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) + +