(request: WSGIRequest)
return JsonResponse(results, safe=False)
Returns online suggestions for a given query.
Returns online suggestions for a given query.
Returns online suggestions for a given query .
def online_suggestions(request: WSGIRequest) -> JsonResponse:
    """Returns online suggestions for a given query."""
    query = request.GET["term"]
    suggest_playlist = request.GET["playlist"] == "true"

    if storage.get("new_music_only") and not suggest_playlist:
        return JsonResponse([], safe=False)

    results: List[Dict[str, Union[str, int]]] = []
    if storage.get("online_suggestions") and redis.get("has_internet"):
        threads = []
        results_lock = threading.Lock()

        def fetch_youtube() -> None:
            from core.musiq.youtube import Youtube

            youtube_suggestions = Youtube().get_search_suggestions(query)
            youtube_suggestions = youtube_suggestions[
                : storage.get("youtube_suggestions")
            ]
            with results_lock:
                for suggestion in youtube_suggestions:
                    results.append(
                        {"key": -1, "value": suggestion, "type": "youtube-online"}
                    )

        def fetch_spotify() -> None:
            from core.musiq.spotify import Spotify

            spotify_suggestions = Spotify().get_search_suggestions(
                query, suggest_playlist
            )
            spotify_suggestions = spotify_suggestions[
                : storage.get("spotify_suggestions")
            ]
            with results_lock:
                for suggestion, external_url in spotify_suggestions:
                    results.append(
                        {
                            "key": external_url,
                            "value": suggestion,
                            "type": "spotify-online",
                        }
                    )

        def fetch_soundcloud() -> None:
            from core.musiq.soundcloud import Soundcloud

            soundcloud_suggestions = Soundcloud().get_search_suggestions(query)
            soundcloud_suggestions = soundcloud_suggestions[
                : storage.get("soundcloud_suggestions")
            ]
            with results_lock:
                for suggestion in soundcloud_suggestions:
                    results.append(
                        {"key": -1, "value": suggestion, "type": "soundcloud-online"}
                    )

        def fetch_jamendo() -> None:
            from core.musiq.jamendo import Jamendo

            jamendo_suggestions = Jamendo().get_search_suggestions(query)
            jamendo_suggestions = jamendo_suggestions[
                : storage.get("jamendo_suggestions")
            ]
            with results_lock:
                for suggestion in jamendo_suggestions:
                    results.append(
                        {"key": -1, "value": suggestion, "type": "jamendo-online"}
                    )

        suggestion_fetchers = {
            "youtube": fetch_youtube,
            "spotify": fetch_spotify,
            "soundcloud": fetch_soundcloud,
            "jamendo": fetch_jamendo,
        }

        for platform in ["youtube", "spotify", "soundcloud", "jamendo"]:
            if (
                storage.get(f"{platform}_enabled")
                and storage.get(f"{platform}_suggestions") > 0
            ):
                thread = threading.Thread(target=suggestion_fetchers[platform])
                threads.append(thread)
                thread.start()

        for thread in threads:
            thread.join()

    return JsonResponse(results, safe=False)
def online_suggestions ( request : WSGIRequest ) -> JsonResponse : query = request . GET [ "term" ] suggest_playlist = request . GET [ "playlist" ] == "true" if storage . get ( "new_music_only" ) and not suggest_playlist : return JsonResponse ( [ ] , safe = False ) results : List [ Dict [ str , Union [ str , int ] ] ] = [ ] if storage . get ( "online_suggestions" ) and redis . get ( "has_internet" ) : threads = [ ] results_lock = threading . Lock ( ) def fetch_youtube ( ) -> None : from core . musiq . youtube import Youtube youtube_suggestions = Youtube ( ) . get_search_suggestions ( query ) youtube_suggestions = youtube_suggestions [ : storage . get ( "youtube_suggestions" ) ] with results_lock : for suggestion in youtube_suggestions : results . append ( { "key" : - 1 , "value" : suggestion , "type" : "youtube-online" } ) def fetch_spotify ( ) -> None : from core . musiq . spotify import Spotify spotify_suggestions = Spotify ( ) . get_search_suggestions ( query , suggest_playlist ) spotify_suggestions = spotify_suggestions [ : storage . get ( "spotify_suggestions" ) ] with results_lock : for suggestion , external_url in spotify_suggestions : results . append ( { "key" : external_url , "value" : suggestion , "type" : "spotify-online" , } ) def fetch_soundcloud ( ) -> None : from core . musiq . soundcloud import Soundcloud soundcloud_suggestions = Soundcloud ( ) . get_search_suggestions ( query ) soundcloud_suggestions = soundcloud_suggestions [ : storage . get ( "soundcloud_suggestions" ) ] with results_lock : for suggestion in soundcloud_suggestions : results . append ( { "key" : - 1 , "value" : suggestion , "type" : "soundcloud-online" } ) def fetch_jamendo ( ) -> None : from core . musiq . jamendo import Jamendo jamendo_suggestions = Jamendo ( ) . get_search_suggestions ( query ) jamendo_suggestions = jamendo_suggestions [ : storage . get ( "jamendo_suggestions" ) ] with results_lock : for suggestion in jamendo_suggestions : results . append ( { "key" : - 1 , "value" : suggestion , "type" : "jamendo-online" } ) suggestion_fetchers = { "youtube" : fetch_youtube , "spotify" : fetch_spotify , "soundcloud" : fetch_soundcloud , "jamendo" : fetch_jamendo , } for platform in [ "youtube" , "spotify" , "soundcloud" , "jamendo" ] : if ( storage . get ( f"{platform}_enabled" ) and storage . get ( f"{platform}_suggestions" ) > 0 ) : thread = threading . Thread ( target = suggestion_fetchers [ platform ] ) threads . append ( thread ) thread . start ( ) for thread in threads : thread . join ( ) return JsonResponse ( results , safe = False )