path: root/tidal_dl_ng/download.py

import os
import pathlib
import random
import shutil
import tempfile
import time
from collections.abc import Callable
from uuid import uuid4

import ffmpeg
import m3u8
import requests
from requests.exceptions import HTTPError
from rich.progress import Progress, TaskID
from tidalapi import Album, Mix, Playlist, Session, Track, UserPlaylist, Video
from tidalapi.media import AudioExtensions, Codec, Quality, StreamManifest, VideoExtensions

from tidal_dl_ng.config import Settings, Tidal
from tidal_dl_ng.constants import EXTENSION_LYRICS, REQUESTS_TIMEOUT_SEC, MediaType, QualityVideo, SkipExisting
from tidal_dl_ng.helper.decryption import decrypt_file, decrypt_security_token
from tidal_dl_ng.helper.exceptions import MediaMissing
from tidal_dl_ng.helper.path import check_file_exists, format_path_media, path_file_sanitize
from tidal_dl_ng.helper.tidal import (
    instantiate_media,
    items_results_all,
    name_builder_album_artist,
    name_builder_artist,
    name_builder_item,
    name_builder_title,
)
from tidal_dl_ng.metadata import Metadata
from tidal_dl_ng.model.gui_data import ProgressBars


# TODO: Set appropriate client string and use it for video download.
# https://github.com/globocom/m3u8#using-different-http-clients
class RequestsClient:
    def download(
        self, uri: str, timeout: int = REQUESTS_TIMEOUT_SEC, headers: dict | None = None, verify_ssl: bool = True
    ):
        if not headers:
            headers = {}

        o = requests.get(uri, timeout=timeout, headers=headers)

        return o.text, o.url


# TODO: Use pathlib.Path everywhere
class Download:
    settings: Settings
    session: Session
    skip_existing: SkipExisting = SkipExisting.Disabled
    fn_logger: Callable
    progress_gui: ProgressBars
    progress: Progress

    def __init__(
        self,
        session: Session,
        path_base: str,
        fn_logger: Callable,
        skip_existing: SkipExisting = SkipExisting.Disabled,
        progress_gui: ProgressBars = None,
        progress: Progress = None,
    ):
        self.settings = Settings()
        self.session = session
        self.skip_existing = skip_existing
        self.fn_logger = fn_logger
        self.progress_gui = progress_gui
        self.progress = progress
        self.path_base = path_base

        if not self.settings.data.path_binary_ffmpeg and (
            self.settings.data.video_convert_mp4 or self.settings.data.extract_flac
        ):
            self.settings.data.video_convert_mp4 = False
            self.settings.data.extract_flac = False

            self.fn_logger.error(
                "FFmpeg path is not set. Videos can be downloaded but will not be processed. FLAC cannot be "
                "extracted from MP4 containers. Make sure FFmpeg is installed. The path to the FFmpeg binary must "
                "be set in (`path_binary_ffmpeg`)."
            )

    def _download(
        self,
        media: Track | Video,
        path_file: str,
    ) -> str:
        media_name: str = name_builder_item(media)
        urls: [str]

        # Get urls for media.
        if isinstance(media, Track):
            urls = media.get_stream().get_stream_manifest().urls
            stream_manifest: StreamManifest = media.get_stream().get_stream_manifest()
        elif isinstance(media, Video):
            m3u8_variant: m3u8.M3U8 = m3u8.load(media.get_url())
            # Find the desired video resolution or the next best one.
            m3u8_playlist, codecs = self._extract_video_stream(m3u8_variant, int(self.settings.data.quality_video))
            # Populate urls.
            urls = m3u8_playlist.files

        # Set the correct progress output channel.
        if self.progress_gui is None:
            progress_stdout: bool = True
        else:
            progress_stdout: bool = False
            # Send signal to GUI with media name
            self.progress_gui.item_name.emit(media_name[:30])

        try:
            # Compute total iterations for progress
            urls_count: int = len(urls)

            if urls_count > 1:
                progress_total: int = urls_count
                block_size: int | None = None
            elif urls_count == 1:
                # Compute progress iterations based on the file size.
                r = requests.get(urls[0], stream=True, timeout=REQUESTS_TIMEOUT_SEC)

                r.raise_for_status()

                # Get file size and compute progress steps
                total_size_in_bytes: int = int(r.headers.get("content-length", 0))
                block_size: int | None = 1048576
                progress_total: float = total_size_in_bytes / block_size
            else:
                raise ValueError

            # Create progress Task
            p_task: TaskID = self.progress.add_task(
                f"[blue]Item '{media_name[:30]}'",
                total=progress_total,
                visible=progress_stdout,
            )

            # Write content to file until progress is finished.
            while not self.progress.tasks[p_task].finished:
                with open(path_file, "wb") as f:
                    for url in urls:
                        # Create the request object with stream=True, so the content won't be loaded into memory at once.
                        r = requests.get(url, stream=True, timeout=REQUESTS_TIMEOUT_SEC)

                        r.raise_for_status()

                        # Write the content to disk. If `chunk_size` is set to `None` the whole file will be written at once.
                        for data in r.iter_content(chunk_size=block_size):
                            f.write(data)
                            # Advance progress bar.
                            self.progress.advance(p_task)

                            # To send the progress to the GUI, we need to emit the percentage.
                            if not progress_stdout:
                                self.progress_gui.item.emit(self.progress.tasks[p_task].percentage)
        except HTTPError as e:
            # TODO: Handle Exception...
            self.fn_logger(e)

        if isinstance(media, Track) and stream_manifest.is_encrypted:
            key, nonce = decrypt_security_token(stream_manifest.encryption_key)
            tmp_path_file_decrypted = path_file + "_decrypted"
            decrypt_file(path_file, tmp_path_file_decrypted, key, nonce)
        else:
            tmp_path_file_decrypted = path_file

        return tmp_path_file_decrypted

    def item(
        self,
        file_template: str,
        media: Track | Video = None,
        media_id: str = None,
        media_type: MediaType = None,
        video_download: bool = True,
        download_delay: bool = False,
        quality_audio: Quality | None = None,
        quality_video: QualityVideo | None = None,
    ) -> (bool, str):
        try:
            if media_id and media_type:
                # If no media instance is provided, we need to create the media instance.
                media = instantiate_media(self.session, media_type, media_id)
            elif isinstance(media, Track):  # Check if media is available not deactivated / removed from TIDAL.
                if not media.available:
                    self.fn_logger.info(
                        f"This track is not available for listening anymore on TIDAL. Skipping: {name_builder_item(media)}"
                    )

                    return False, ""
                else:
                    # Re-create media instance with full album information
                    media = self.session.track(media.id, with_album=True)
            elif not media:
                raise MediaMissing
        except:
            return False, ""

        # If video download is not allowed end here
        if not video_download and isinstance(media, Video):
            self.fn_logger.info(
                f"Video downloads are deactivated (see settings). Skipping video: {name_builder_item(media)}"
            )

            return False, ""

        # Get extension.
        file_extension: str

        if isinstance(media, Track):
            # If hi_res is forbidden in settings and highest track quality is hi_res, downgrade to high_lossless.
            if self.settings.data.downgrade_on_hi_res and media.get_stream().audio_quality == Quality.hi_res:
                quality_audio = Quality.high_lossless

            # If a quality is explicitly set, change it.
            if quality_audio:
                quality_audio_old: Quality = self.adjust_quality_audio(quality_audio)

            file_extension = media.get_stream().get_stream_manifest().file_extension
            # Use M4A extension for MP4 audio tracks, because it looks better and is completely interchangeable.
            file_extension = AudioExtensions.M4A if file_extension == AudioExtensions.MP4 else file_extension

            if self.settings.data.extract_flac:
                do_flac_extract = False

                if (
                    media.get_stream().get_stream_manifest().codecs.upper() == Codec.FLAC
                    and file_extension != AudioExtensions.FLAC
                ):
                    file_extension = AudioExtensions.FLAC
                    do_flac_extract = True
        elif isinstance(media, Video):
            if quality_video:
                quality_video_old: QualityVideo = self.adjust_quality_video(quality_video)

            file_extension = AudioExtensions.MP4 if self.settings.data.video_convert_mp4 else VideoExtensions.TS

        # Create file name and path
        file_name_relative = format_path_media(file_template, media)
        path_media_dst = os.path.abspath(
            os.path.normpath(os.path.join(os.path.expanduser(self.path_base), file_name_relative))
        )

        # Sanitize final path_file to fit into OS boundaries.
        uniquify: bool = self.skip_existing == SkipExisting.Append
        path_media_dst = path_file_sanitize(path_media_dst + file_extension, adapt=True, uniquify=uniquify)

        # Compute if and how downloads need to be skipped.
        if self.skip_existing in (SkipExisting.ExtensionIgnore, SkipExisting.Filename):
            extension_ignore: bool = self.skip_existing == SkipExisting.ExtensionIgnore
            file_exists: bool = check_file_exists(path_media_dst, extension_ignore=extension_ignore)
        else:
            file_exists: bool = False

        if not file_exists:
            # Create a temp directory and file.
            with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp_path_dir:
                tmp_path_file = os.path.join(tmp_path_dir, str(uuid4()))
                # Download media.
                tmp_path_file = self._download(media=media, path_file=tmp_path_file)

                # Convert video from TS to MP4
                if isinstance(media, Video) and self.settings.data.video_convert_mp4:
                    # Convert `*.ts` file to `*.mp4` using ffmpeg
                    tmp_path_file = self._video_convert(tmp_path_file)

                # Extract FLAC from MP4 container using ffmpeg
                if isinstance(media, Track) and self.settings.data.extract_flac and do_flac_extract:
                    tmp_path_file = self._extract_flac(tmp_path_file)

                tmp_path_lyrics: pathlib.Path | None = None

                # Write metadata to file.
                if not isinstance(media, Video):
                    result_metadata, tmp_path_lyrics = self.metadata_write(media, tmp_path_file)

                # Move final file to the configured destination directory.
                os.makedirs(os.path.dirname(path_media_dst), exist_ok=True)
                shutil.move(tmp_path_file, path_media_dst)

                # Move lyrics file
                if self.settings.data.lyrics_file and not isinstance(media, Video):
                    self._move_lyrics(path_media_dst, tmp_path_lyrics)
        else:
            self.fn_logger.debug(f"Download skipped, since file exists: '{path_media_dst}'")

        status_download: bool = not file_exists

        if quality_audio:
            # Set quality back to the global user value
            self.adjust_quality_audio(quality_audio_old)

        if quality_video:
            # Set quality back to the global user value
            self.adjust_quality_video(quality_video_old)

        # Whether a file was downloaded or skipped and the download delay is enabled, wait until the next download.
        # Only use this, if you have a list of several Track items.
        if download_delay:
            time_sleep: float = round(random.SystemRandom().uniform(2, 5), 1)

            self.fn_logger.debug(f"Next download will start in {time_sleep} seconds.")
            time.sleep(time_sleep)

        return status_download, path_media_dst

    def adjust_quality_audio(self, quality) -> Quality:
        # Save original quality settings
        quality_old: Quality = self.session.audio_quality
        self.session.audio_quality = quality
        tidal: Tidal = Tidal()

        # If track is not requested as hires_lossless do not use PKCE, because BTS downloads are faster than MPD.
        if quality == Quality.hi_res_lossless and not tidal.is_pkce:
            tidal.login_token(do_pkce=True)
        elif quality != Quality.hi_res_lossless and tidal.is_pkce:
            tidal.login_token(do_pkce=False)

        return quality_old

    def adjust_quality_video(self, quality) -> QualityVideo:
        quality_old: QualityVideo = self.settings.data.quality_video

        self.settings.data.quality_video = quality

        return quality_old

    def _move_lyrics(self, file_media_dst: str, path_lyrics: pathlib.Path) -> bool:
        result: bool
        # Build tmp lyrics filename
        # Check if the file was downloaded
        if path_lyrics and os.path.isfile(path_lyrics):
            # Move it.
            shutil.move(path_lyrics, os.path.splitext(file_media_dst)[0] + EXTENSION_LYRICS)

            result = True
        else:
            result = False

        return result

    def lyrics_write_file(self, dir_destination: pathlib.Path, lyrics: str) -> str:
        result: str = dir_destination / str(uuid4())

        try:
            with open(result, "x", encoding="utf-8") as f:
                f.write(lyrics)
        except:
            result = ""

        return result

    def metadata_write(self, track: Track, path_media: str) -> (bool, pathlib.Path | None):
        result: bool = False
        path_lyrics: pathlib.Path | None = None
        release_date: str = (
            track.album.available_release_date.strftime("%Y-%m-%d")
            if track.album.available_release_date
            else track.album.release_date.strftime("%Y-%m-%d") if track.album.release_date else ""
        )
        copy_right: str = track.copyright if hasattr(track, "copyright") and track.copyright else ""
        isrc: str = track.isrc if hasattr(track, "isrc") and track.isrc else ""
        lyrics: str = ""

        if self.settings.data.lyrics_embed or self.settings.data.lyrics_file:
            # Try to retrieve lyrics.
            try:
                lyrics_obj = track.lyrics()

                if lyrics_obj.subtitles:
                    lyrics = lyrics_obj.subtitles
                elif lyrics_obj.text:
                    lyrics = lyrics_obj.text
            except:
                lyrics = ""
                # TODO: Implement proper logging.
                print(f"Could not retrieve lyrics for `{name_builder_item(track)}`.")

        if lyrics and self.settings.data.lyrics_file:
            path_lyrics = self.lyrics_write_file(pathlib.Path(path_media).parent, lyrics)

        # `None` values are not allowed.
        m: Metadata = Metadata(
            path_file=path_media,
            lyrics=lyrics,
            copy_right=copy_right,
            title=name_builder_title(track),
            artists=name_builder_artist(track),
            album=track.album.name if track.album else "",
            tracknumber=track.track_num,
            date=release_date,
            isrc=isrc,
            albumartist=name_builder_album_artist(track),
            totaltrack=track.album.num_tracks if track.album and track.album.num_tracks else 1,
            totaldisc=track.album.num_volumes if track.album and track.album.num_volumes else 1,
            discnumber=track.volume_num if track.volume_num else 1,
            url_cover=track.album.image(int(self.settings.data.metadata_cover_dimension)),
        )

        m.save()

        result = True

        return result, path_lyrics

    def items(
        self,
        file_template: str,
        media: Album | Playlist | UserPlaylist | Mix = None,
        media_id: str = None,
        media_type: MediaType = None,
        video_download: bool = False,
        download_delay: bool = True,
        quality_audio: Quality | None = None,
        quality_video: QualityVideo | None = None,
    ):
        # If no media instance is provided, we need to create the media instance.
        if media_id and media_type:
            media = instantiate_media(self.session, media_type, media_id)
        elif not media:
            raise MediaMissing

        # Create file name and path
        file_name_relative = format_path_media(file_template, media)

        # Get the name of the list and check, if videos should be included.
        videos_include: bool = True

        if isinstance(media, Mix):
            list_media_name = media.title[:30]
        elif video_download:
            list_media_name = name_builder_title(media)[:30]
        else:
            videos_include = False
            list_media_name = name_builder_title(media)[:30]

        # Get all items of the list.
        items = items_results_all(media, videos_include=videos_include)

        # Determine where to redirect the progress information.
        if self.progress_gui is None:
            progress_stdout: bool = True
        else:
            progress_stdout: bool = False
            self.progress_gui.list_name.emit(list_media_name[:30])

        # Create the list progress task.
        p_task1: TaskID = self.progress.add_task(
            f"[green]List '{list_media_name}'", total=len(items), visible=progress_stdout
        )

        # Iterate through list items
        while not self.progress.finished:
            for media in items:
                # Download the item.
                status_download, result_path_file = self.item(
                    media=media,
                    file_template=file_name_relative,
                    quality_audio=quality_audio,
                    quality_video=quality_video,
                    download_delay=download_delay,
                )

                # Advance progress bar.
                self.progress.advance(p_task1)

                if not progress_stdout:
                    self.progress_gui.list_item.emit(self.progress.tasks[p_task1].percentage)

    def _video_convert(self, path_file: str) -> str:
        path_file_out = path_file + AudioExtensions.MP4
        result, _ = (
            ffmpeg.input(path_file)
            .output(path_file_out, map=0, c="copy", loglevel="quiet")
            .run(cmd=self.settings.data.path_binary_ffmpeg)
        )

        return path_file_out

    def _extract_flac(self, path_media_src: str) -> str:
        path_media_out = path_media_src + AudioExtensions.FLAC
        result, _ = (
            ffmpeg.input(path_media_src)
            .output(
                path_media_out, map=0, movflags="use_metadata_tags", acodec="copy", map_metadata="0:g", loglevel="quiet"
            )
            .run(cmd=self.settings.data.path_binary_ffmpeg)
        )

        return path_media_out

    def _extract_video_stream(self, m3u8_variant: m3u8.M3U8, quality: int) -> (m3u8.M3U8 | bool, str):
        m3u8_playlist: m3u8.M3U8 | bool = False
        resolution_best: int = 0
        mime_type: str = ""

        if m3u8_variant.is_variant:
            for playlist in m3u8_variant.playlists:
                if resolution_best < playlist.stream_info.resolution[1]:
                    resolution_best = playlist.stream_info.resolution[1]
                    m3u8_playlist = m3u8.load(playlist.uri)
                    mime_type = playlist.stream_info.codecs

                    if quality == playlist.stream_info.resolution[1]:
                        break

        return m3u8_playlist, mime_type