Add simple caching of radio stations

requirements.txt

@@ -1 +1,2 @@
-requests
+requests~=2.31.0
+dead-simple-cache>=1.2.2

tunein/__init__.py

@@ -1,8 +1,12 @@
-from urllib.parse import urlparse, urlunparse
-
+import os
 import requests
+from dead_simple_cache import SimpleCache
+from urllib.parse import urlparse, urlunparse
 from tunein.parse import fuzzy_match
 
+BASE_DIR = os.getenv("HOME") or os.path.dirname(os.path.abspath(__file__))
+DEFAULT_CACHE_PATH = os.path.join(BASE_DIR, ".cache", "radios")
+
 
 class TuneInStation:
     def __init__(self, raw):
@@ -67,6 +71,7 @@ class TuneIn:
     search_url = "https://opml.radiotime.com/Search.ashx"
     featured_url = "http://opml.radiotime.com/Browse.ashx"  # local stations
     stnd_query = {"formats": "mp3,aac,ogg,html,hls", "render": "json"}
+    cache = SimpleCache(file_path=DEFAULT_CACHE_PATH)
 
     @staticmethod
     def get_stream_urls(url):
@@ -86,14 +91,20 @@ def get_stream_urls(url):
 
         stations = res.json().get("body", {})
 
+        working_stations = []
         for station in stations:
             if station.get("url", "").endswith(".pls"):
-                res = requests.get(station["url"])
+                # TODO: come up with a better fix 
+                # Catch and avoid invalid certificate errors
+                try:
+                    res = requests.get(station["url"])
+                except Exception:
+                    continue
                 file1 = [line for line in res.text.split("\n") if line.startswith("File1=")]
                 if file1:
                     station["url"] = file1[0].split("File1=")[1]
-
-        return stations
+                working_stations.append(station)
+        return working_stations
 
     @staticmethod
     def featured():
@@ -104,14 +115,56 @@ def featured():
         stations = res.json().get("body", [{}])[0].get("children", [])
         return list(TuneIn._get_stations(stations))
 
+    @staticmethod
+    def search_cache(query):
+        """Search for cached stations."""
+        items = TuneIn.cache.get(query, fuzzy=True)
+        server_alive = {}
+        for key, stations in items.items():
+            for station in stations:
+                url = station["url"]
+                # Check whether each server is alive
+                if url not in server_alive:
+                    # TODO: find a faster way to check whether each server is up
+                    # response = requests.head(url, timeout=1)
+                    # code = response.status_code
+                    # server_alive[url] = str(code).startswith('2') or str(code).startswith('3')
+                    server_alive[url] = True
+                if not server_alive[url]:
+                    stations.remove(station)
+        for key, stations in items.items():
+            if stations:
+                TuneIn.cache.replace(key=key, data=stations)
+            else:
+                TuneIn.cache.delete(key=key)
+        return sum(list(items.values()), [])
+
     @staticmethod
     def search(query):
-        res = requests.post(
-            TuneIn.search_url,
-            data={**TuneIn.stnd_query, **{"query": query}}
-        )
-        stations = res.json().get("body", [])
-        return list(TuneIn._get_stations(stations, query))
+        # NOTE: to make the cache persistent on disk, it is necessary to sync it,
+        # but since the cache is a static attribute, one must open/close it explicitly.
+        # To make the cache persistent on disk, uncomment the following line.
+        # TuneIn.cache.open()
+        cached_items = TuneIn.search_cache(query)
+        if cached_items:
+            stations = [TuneInStation(item) for item in cached_items]
+        else:
+            # Search again
+            res = requests.post(
+                TuneIn.search_url,
+                data={**TuneIn.stnd_query, **{"query": query}}
+            )
+            stations = list(
+                TuneIn._get_stations(res.json().get("body", []), query)
+            )
+            # Update cache
+            for station in filter(lambda s: s.title != '', stations):
+                TuneIn.cache.add(key=query, data=station.raw)
+        # NOTE: to make the cache persistent on disk, it is necessary to sync it,
+        # but since the cache is a static attribute, one must open/close it explicitly.
+        # To make the cache persistent on disk, uncomment the following line.
+        # TuneIn.cache.close()
+        return stations
 
     @staticmethod
     def _get_stations(stations: requests.Response, query: str = ""):