Skip to content
Back to top

# Source code for searx.engines.radio_browser

    # SPDX-License-Identifier: AGPL-3.0-or-later
    """Search radio stations from RadioBrowser by `Advanced station search API`_.

    .. _Advanced station search API:
       https://de1.api.radio-browser.info/#Advanced_station_search

    """

    import os
    import random
    import socket
    from urllib.parse import urlencode

    import babel
    from flask_babel import gettext

    from searx.enginelib import EngineCache
    from searx.enginelib.traits import EngineTraits
    from searx.locales import language_tag

    about = {
        "website": "https://www.radio-browser.info/",
        "wikidata_id": "Q111664849",
        "official_api_documentation": "https://de1.api.radio-browser.info/",
        "use_official_api": True,
        "require_api_key": False,
        "results": "JSON",
    }
    paging = True
    categories = ["music", "radio"]

    page_size = 10

    station_filters = []  # ['countrycode', 'language']
    """A list of filters to be applied to the search of radio stations.  By default
    none filters are applied. Valid filters are:

    ``language``
      Filter stations by selected language.  For instance the ``de`` from ``:de-AU``
      will be translated to `german` and used in the argument ``language=``.

    ``countrycode``
      Filter stations by selected country.  The 2-digit countrycode of the station
      comes from the region the user selected.  For instance ``:de-AU`` will filter
      out all stations not in ``AU``.

    .. note::

       RadioBrowser has registered a lot of languages and countrycodes unknown to
       :py:obj:`babel` and note that when searching for radio stations, users are
       more likely to search by name than by region or language.

    """

    CACHE: EngineCache
    """Persistent (SQLite) key/value cache that deletes its values after ``expire``
    seconds."""

    def init(_):
        global CACHE  # pylint: disable=global-statement
        CACHE = EngineCache("radio_browser")

        # In an environment with competing processes, the initial loading of the
        # cache is required only once.
        eng_state: str | None = CACHE.get("eng_state")
        if not eng_state or not eng_state.startswith("STATE:"):
            CACHE.set("eng_state", f"STATE: being initialized by PID {os.getpid()}")
            try:
                server_list()
            except Exception:
                CACHE.set("eng_state", f"ERROR: initialization by PID {os.getpid()} failed.")
                raise
        else:
            logger.debug(eng_state)

    def server_list() -> list[str]:

        servers = CACHE.get("servers", [])
        if servers:
            return servers

        # hint: can take up to 40sec!
        ips = socket.getaddrinfo("all.api.radio-browser.info", 80, 0, 0, socket.IPPROTO_TCP)
        for ip_tuple in ips:
            _ip: str = ip_tuple[4][0]  # type: ignore
            try:
                url = socket.gethostbyaddr(_ip)[0]
            except socket.herror:
                # https://github.com/searxng/searxng/issues/5439
                continue
            srv = "https://" + url
            if srv not in servers:
                servers.append(srv)

        # update server list once in 24h
        CACHE.set(key="servers", value=servers, expire=60 * 60 * 24)

        return servers

    def request(query, params):

        servers = server_list()
        if not servers:
            logger.error("Fetched server list is empty!")
            params["url"] = None
            return

        server = random.choice(servers)

        args = {
            "name": query,
            "order": "votes",
            "offset": (params["pageno"] - 1) * page_size,
            "limit": page_size,
            "hidebroken": "true",
            "reverse": "true",
        }

        if "language" in station_filters:
            lang = traits.get_language(params["searxng_locale"])  # type: ignore
            if lang:
                args["language"] = lang

        if "countrycode" in station_filters:
            if len(params["searxng_locale"].split("-")) > 1:
                countrycode = params["searxng_locale"].split("-")[-1].upper()
                if countrycode in traits.custom["countrycodes"]:  # type: ignore
                    args["countrycode"] = countrycode

        params["url"] = f"{server}/json/stations/search?{urlencode(args)}"

    def response(resp):
        results = []

        json_resp = resp.json()

        for result in json_resp:
            url = result["homepage"]
            if not url:
                url = result["url_resolved"]

            content = []
            tags = ", ".join(result.get("tags", "").split(","))
            if tags:
                content.append(tags)
            for x in ["state", "country"]:
                v = result.get(x)
                if v:
                    v = str(v).strip()
                    content.append(v)

            metadata = []
            codec = result.get("codec")
            if codec and codec.lower() != "unknown":
                metadata.append(f"{codec} " + gettext("radio"))
            for x, y in [
                (gettext("bitrate"), "bitrate"),
                (gettext("votes"), "votes"),
                (gettext("clicks"), "clickcount"),
            ]:
                v = result.get(y)
                if v:
                    v = str(v).strip()
                    metadata.append(f"{x} {v}")
            results.append(
                {
                    "url": url,
                    "title": result["name"],
                    "thumbnail": result.get("favicon", "").replace("http://", "https://"),
                    "content": " | ".join(content),
                    "metadata": " | ".join(metadata),
                    "iframe_src": result["url_resolved"].replace("http://", "https://"),
                }
            )

        return results

    def fetch_traits(engine_traits: EngineTraits):
        """Fetch languages and countrycodes from RadioBrowser

        - ``traits.languages``: `list of languages API`_
        - ``traits.custom['countrycodes']``: `list of countries API`_

        .. _list of countries API: https://de1.api.radio-browser.info/#List_of_countries
        .. _list of languages API: https://de1.api.radio-browser.info/#List_of_languages
        """
        # pylint: disable=import-outside-toplevel

        init(None)
        from babel.core import get_global

        from searx.network import get  # see https://github.com/searxng/searxng/issues/762

        babel_reg_list = get_global("territory_languages").keys()

        server = server_list()[0]

        resp = get(
            f"{server}/json/languages",
            timeout=5,
        )
        if not resp.ok:
            raise RuntimeError("Response from radio-browser languages is not OK.")

        language_list = resp.json()

        resp = get(
            f"{server}/json/countries",
            timeout=5,
        )
        if not resp.ok:
            raise RuntimeError("Response from radio-browser countries is not OK.")

        country_list = resp.json()

        for lang in language_list:
            babel_lang = lang.get("iso_639")
            if not babel_lang:
                # the language doesn't have any iso code, and hence can't be parsed
                # print(f"ERROR: lang - no iso code in {lang}")
                continue
            try:
                sxng_tag = language_tag(babel.Locale.parse(babel_lang, sep="-"))
            except babel.UnknownLocaleError:
                # print(f"ERROR: language tag {babel_lang} is unknown by babel")
                continue

            eng_tag = lang["name"]
            conflict = engine_traits.languages.get(sxng_tag)
            if conflict:
                if conflict != eng_tag:
                    print("CONFLICT: babel %s --> %s, %s" % (sxng_tag, conflict, eng_tag))
                continue
            engine_traits.languages[sxng_tag] = eng_tag

        countrycodes = set()
        for region in country_list:
            # country_list contains duplicates that differ only in upper/lower case
            _reg = region["iso_3166_1"].upper()
            if _reg not in babel_reg_list:
                print(f"ERROR: region tag {region['iso_3166_1']} is unknown by babel")
                continue
            countrycodes.add(_reg)

        countrycodes = list(countrycodes)
        countrycodes.sort()
        engine_traits.custom["countrycodes"] = countrycodes