piped

Piped module for downloading and searching songs.

Piped(output_format='mp3', cookie_file=None, search_query=None, filter_results=True, yt_dlp_args=None)

Bases: AudioProvider

YouTube Music audio provider class

Arguments

• output_directory: The directory to save the downloaded songs to.
• output_format: The format to save the downloaded songs in.
• cookie_file: The path to a file containing cookies to be used by YTDL.
• search_query: The query to use when searching for songs.
• filter_results: Whether to filter results.

Source code in spotdl/providers/audio/piped.py

def __init__(  # pylint: disable=super-init-not-called
    self,
    output_format: str = "mp3",
    cookie_file: Optional[str] = None,
    search_query: Optional[str] = None,
    filter_results: bool = True,
    yt_dlp_args: Optional[str] = None,
) -> None:
    """
    Pipe audio provider class

    ### Arguments
    - output_directory: The directory to save the downloaded songs to.
    - output_format: The format to save the downloaded songs in.
    - cookie_file: The path to a file containing cookies to be used by YTDL.
    - search_query: The query to use when searching for songs.
    - filter_results: Whether to filter results.
    """

    self.output_format = output_format
    self.cookie_file = cookie_file
    self.search_query = search_query
    self.filter_results = filter_results

    if self.output_format == "m4a":
        ytdl_format = "best[ext=m4a]/best"
    elif self.output_format == "opus":
        ytdl_format = "best[ext=webm]/best"
    else:
        ytdl_format = "best"

    yt_dlp_options = {
        "format": ytdl_format,
        "quiet": True,
        "no_warnings": True,
        "encoding": "UTF-8",
        "logger": YTDLLogger(),
        "cookiefile": self.cookie_file,
        "outtmpl": f"{get_temp_path()}/%(id)s.%(ext)s",
        "retries": 5,
    }

    if yt_dlp_args:
        user_options = args_to_ytdlp_options(shlex.split(yt_dlp_args))
        yt_dlp_options.update(user_options)

    self.audio_handler = YoutubeDL(yt_dlp_options)
    self.session = requests.Session()

get_download_metadata(url, download=False)

Get metadata for a download using yt-dlp.

Arguments

• url: The url to get metadata for.

Returns

• A dictionary containing the metadata.

Source code in spotdl/providers/audio/piped.py

def get_download_metadata(self, url: str, download: bool = False) -> Dict:
    """
    Get metadata for a download using yt-dlp.

    ### Arguments
    - url: The url to get metadata for.

    ### Returns
    - A dictionary containing the metadata.
    """

    url_id = url.split("?v=")[1]
    piped_response = requests.get(
        f"https://piped.video/streams/{url_id}",
        timeout=10,
        proxies=GlobalConfig.get_parameter("proxies"),
    )

    if piped_response.status_code != 200:
        raise AudioProviderError(
            f"Failed to get metadata for {url} from Piped: {piped_response.text}"
        )

    piped_data = piped_response.json()

    yt_dlp_json = {
        "title": piped_data["title"],
        "id": url_id,
        "view_count": piped_data["views"],
        "extractor": "Generic",
        "formats": [],
    }

    for audio_stream in piped_data["audioStreams"]:
        yt_dlp_json["formats"].append(
            {
                "url": audio_stream["url"],
                "ext": "webm" if audio_stream["codec"] == "opus" else "m4a",
                "abr": audio_stream["quality"].split(" ")[0],
                "filesize": audio_stream["contentLength"],
            }
        )

    return self.audio_handler.process_video_result(yt_dlp_json, download=download)

get_results(search_term, **kwargs)

Get results from YouTube Music API and simplify them

Arguments

• search_term: The search term to search for.
• kwargs: other keyword arguments passed to the YTMusic.search method.

Returns

• A list of simplified results (dicts)

Source code in spotdl/providers/audio/piped.py

def get_results(self, search_term: str, **kwargs) -> List[Result]:
    """
    Get results from YouTube Music API and simplify them

    ### Arguments
    - search_term: The search term to search for.
    - kwargs: other keyword arguments passed to the `YTMusic.search` method.

    ### Returns
    - A list of simplified results (dicts)
    """

    if kwargs is None:
        kwargs = {}

    params = {"q": search_term, **kwargs}
    if params.get("filter") is None:
        params["filter"] = "music_videos"

    response = self.session.get(
        "https://piped.video/search",
        params=params,
        headers=HEADERS,
        timeout=20,
    )

    if response.status_code != 200:
        raise AudioProviderError(
            f"Failed to get results for {search_term} from Piped: {response.text}"
        )

    search_results = response.json()

    # Simplify results
    results = []
    for result in search_results["items"]:
        if result["type"] != "stream":
            continue

        isrc_result = ISRC_REGEX.search(search_term)

        results.append(
            Result(
                source="piped",
                url=f"https://piped.video{result['url']}",
                verified=kwargs.get("filter") == "music_songs",
                name=result["title"],
                duration=result["duration"],
                author=result["uploaderName"],
                result_id=result["url"].split("?v=")[1],
                artists=(
                    (result["uploaderName"],)
                    if kwargs.get("filter") == "music_songs"
                    else None
                ),
                isrc_search=isrc_result is not None,
                search_query=search_term,
            )
        )

    return results