mirror of
https://github.com/glomatico/votify.git
synced 2026-09-30 21:12:14 +02:00
Rafael Moraes
@@ -1,10 +1,12 @@
|
||||
# Votify
|
||||
A Python CLI app for downloading songs/podcasts from Spotify.
|
||||
A Python CLI app for downloading songs/podcasts/videos from Spotify.
|
||||
|
||||
**Discord Server:** https://discord.gg/aBjMEZ9tnq
|
||||
|
||||
## Features
|
||||
* Download songs and podcasts in Vorbis 96/128kbps or in 320kbps with a premium account
|
||||
* Download songs and podcasts in Vorbis or AAC
|
||||
* Download podcast videos
|
||||
* Download music videos with a premium account
|
||||
* Support for artist links to download all of their albums
|
||||
* Download synced lyrics
|
||||
* Highly configurable
|
||||
@@ -12,21 +14,35 @@ A Python CLI app for downloading songs/podcasts from Spotify.
|
||||
## Prerequisites
|
||||
* Python 3.8 or higher
|
||||
* The cookies file of your Spotify browser session in Netscape format (free or premium)
|
||||
* You can get your cookies by using one of the following extensions on your browser of choice at the Spotify website with your account signed in:
|
||||
* To export your cookies, use one of the following browser extensions while signed in to Spotify:
|
||||
* Firefox: https://addons.mozilla.org/addon/export-cookies-txt
|
||||
* Chromium based browsers: https://chrome.google.com/webstore/detail/gdocmgbfkjnnpapoeobnolbbkoibbcif
|
||||
* Unplayplay
|
||||
* Build it from source: https://git.gay/glomatico/unplayplay.
|
||||
|
||||
### Optional dependencies
|
||||
The following tools are optional but required for specific features. Add them to your system’s PATH or specify their paths using command-line arguments or the config file.
|
||||
* [FFmpeg](https://ffmpeg.org/download.html)
|
||||
* Used when setting `ffmpeg` as remux mode.
|
||||
* Used when setting `mp4` or `webm` as video format.
|
||||
* [MP4Box](https://gpac.io/downloads/gpac-nightly-builds/)
|
||||
* Used when setting `mp4box` as remux mode.
|
||||
* [Shaka Packager](https://github.com/shaka-project/shaka-packager/releases/latest)
|
||||
* Used when setting `webm` as video format and when downloading music videos.
|
||||
* [mp4decrypt](https://www.bento4.com/downloads/)
|
||||
* Used when setting `mp4box` or `mp4decrypt` as remux mode.
|
||||
* [aria2c](https://github.com/aria2/aria2/releases)
|
||||
* Used when setting `aria2c` as download mode.
|
||||
* .wvd file
|
||||
* Used when setting `aac` as audio quality or when downloading music videos.
|
||||
* A .wvd file contains the Widevine keys from a device and is required to decrypt music videos and songs in AAC. The easiest method of obtaining one is using KeyDive, which extracts it from an Android device. Detailed instructions can be found here: https://github.com/hyugogirubato/KeyDive.
|
||||
* Note: .wvd files extracted from emulated devices may not work.
|
||||
|
||||
## Installation
|
||||
1. Install the package `votify` using pip
|
||||
```bash
|
||||
pip install votify
|
||||
```
|
||||
2. Set up the `cookies.txt`.
|
||||
* You can either move to the current directory from which you will be running Votify or specify its path using the command line arguments/config file.
|
||||
3. Set up Unplayplay
|
||||
* You can either add it to your PATH or specify its path using the command line arguments/config file.
|
||||
2. Set up the cookies file.
|
||||
* You can either move to the current directory from which you will be running Votify as `cookies.txt` or specify its path using the command-line arguments/config file.
|
||||
|
||||
## Usage
|
||||
```bash
|
||||
@@ -42,6 +58,25 @@ votify [OPTIONS] URLS...
|
||||
```bash
|
||||
votify "https://open.spotify.com/album/0r8D5N674HbTXlR3zNxeU1"
|
||||
```
|
||||
* Download a podcast episode
|
||||
```bash
|
||||
votify "https://open.spotify.com/episode/3kwxWnzGH8T6UY2Nq582zx"
|
||||
```
|
||||
* Download a podcast series
|
||||
```bash
|
||||
votify "https://open.spotify.com/show/4rOoJ6Egrf8K2IrywzwOMk"
|
||||
* Download a music video
|
||||
```bash
|
||||
votify "https://open.spotify.com/track/31k4hgHmrbzorLZMvMWuzq" --enable-videos
|
||||
```
|
||||
* List and select a related music video to download from a song
|
||||
```bash
|
||||
votify "https://open.spotify.com/track/0a0n6u6j3t6m0p4k0t0k0u0" --enable-videos --download-music-videos
|
||||
```
|
||||
* Download a podcast video
|
||||
```bash
|
||||
votify "https://open.spotify.com/episode/3kwxWnzGH8T6UY2Nq582zx" --enable-videos --download-podcast-videos
|
||||
```
|
||||
* Choose which albums to download from an artist
|
||||
```bash
|
||||
votify "https://open.spotify.com/artist/0gxyHStUsqpMadRV0Di1Qt"
|
||||
@@ -54,46 +89,56 @@ votify [OPTIONS] URLS...
|
||||
* Enter - Confirm selection
|
||||
|
||||
## Configuration
|
||||
Votify can be configured using the command line arguments or the config file.
|
||||
Votify can be configured using the command-line arguments or the config file.
|
||||
|
||||
The config file is created automatically when you run Votify for the first time at `~/.votify/config.json` on Linux and `%USERPROFILE%\.votify\config.json` on Windows.
|
||||
|
||||
Config file values can be overridden using command line arguments.
|
||||
Config file values can be overridden using command-line arguments.
|
||||
|
||||
| Command line argument / Config file key | Description | Default value |
|
||||
| Command-line argument / Config file key | Description | Default value |
|
||||
| --------------------------------------------------------------- | ------------------------------------------------------------------ | ---------------------------------------------- |
|
||||
| `--wait-interval`, `-w` / `wait_interval` | Wait interval between downloads in seconds. | `10` |
|
||||
| `--wait-interval`, `-w` / `wait_interval` | Wait interval between downloads in seconds. | `5` |
|
||||
| `--enable-videos` / `enable_videos` | Enable video downloads when available. | `false` |
|
||||
| `--download-music-videos` / `download_music_videos` | List and select a related music video to download from songs. | `false` |
|
||||
| `--download-podcast-videos` / `download_podcast_videos` | Attempt to download the video version of podcasts. | `false` |
|
||||
| `--force-premium`, `-f` / `force_premium` | Force to detect the account as premium. | `false` |
|
||||
| `--read-urls-as-txt`, `-r` / - | Interpret URLs as paths to text files containing URLs. | `false` |
|
||||
| `--config-path` / - | Path to config file. | `<home>/.votify/config.json` |
|
||||
| `--config-path` / - | Path to config file. | `<home>/.spotify-web-downloader/config.json` |
|
||||
| `--log-level` / `log_level` | Log level. | `INFO` |
|
||||
| `--print-exceptions` / `print_exceptions` | Print exceptions. | `false` |
|
||||
| `--cookies-path` / `cookies_path` | Path to cookies file. | `./cookies.txt` |
|
||||
| `--quality`, `-q` / `quality` | Audio quality. | `medium` |
|
||||
| `--output-path`, `-o` / `output_path` | Path to output directory. | `./Spotify` |
|
||||
| `--temp-path` / `temp_path` | Path to temporary directory. | `./temp` |
|
||||
| `--download-mode`, `-d` / `download_mode` | Download mode. | `ytdlp` |
|
||||
| `--no-exceptions` / `no_exceptions` | Don't print exceptions. | `false` |
|
||||
| `--cookies-path` / `cookies_path` | Path to cookies file. | `cookies.txt` |
|
||||
| `--output-path`, `-o` / `output_path` | Path to output directory. | `Spotify` |
|
||||
| `--temp-path` / `temp_path` | Path to temporary directory. | `temp` |
|
||||
| `--wvd-path` / `wvd_path` | Path to .wvd file. | `device.wvd` |
|
||||
| `--aria2c-path` / `aria2c_path` | Path to aria2c binary. | `aria2c` |
|
||||
| `--unplayplay-path` / `unplayplay_path` | Path to unplayplay binary. | `unplayplay` |
|
||||
| `--ffmpeg-path` / `ffmpeg_path` | Path to ffmpeg binary. | `ffmpeg` |
|
||||
| `--mp4box-path` / `mp4box_path` | Path to MP4Box binary. | `mp4box` |
|
||||
| `--mp4decrypt-path` / `mp4decrypt_path` | Path to mp4decrypt binary. | `mp4decrypt` |
|
||||
| `--packager-path` / `packager_path` | Path to Shaka Packager binary. | `packager` |
|
||||
| `--template-folder-album` / `template_folder_album` | Template folder for tracks that are part of an album. | `{album_artist}/{album}` |
|
||||
| `--template-folder-compilation` / `template_folder_compilation` | Template folder for tracks that are part of a compilation album. | `Compilations/{album}` |
|
||||
| `--template-file-single-disc` / `template_file_single_disc` | Template file for the tracks that are part of a single-disc album. | `{track:02d} {title}` |
|
||||
| `--template-file-multi-disc` / `template_file_multi_disc` | Template file for the tracks that are part of a multi-disc album. | `{disc}-{track:02d} {title}` |
|
||||
| `--template-folder-episode` / `template_folder_episode` | Template folder for episodes (podcasts). | `Podcasts/{album}` |
|
||||
| `--template-file-episode` / `template_file_episode` | Template file for episodes (podcasts). | `{track:02d} {title}` |
|
||||
| `--template-file-episode` / `template_file_episode` | Template file for music videos. | `{track:02d} {title}` |
|
||||
| `--template-folder-music-video` / `template_folder_music_video` | Template folder for music videos | `{artist}/Unknown Album` |
|
||||
| `--template-file-music-video` / `template_file_music_video` | Template file for the tracks that are not part of an album. | `{title}` |
|
||||
| `--template-file-playlist` / `template_file_playlist` | Template file for the M3U8 playlist. | `Playlists/{playlist_artist}/{playlist_title}` |
|
||||
| `--date-tag-template` / `date_tag_template` | Date tag template. | `%Y-%m-%dT%H:%M:%SZ` |
|
||||
| `--save-cover`, `-s` / `save_cover` | Save cover as a separate file. | `false` |
|
||||
| `--save-cover` / `save_cover` | Save cover as a separate file. | `false` |
|
||||
| `--save-playlist` / `save_playlist` | Save a M3U8 playlist file when downloading a playlist. | `false` |
|
||||
| `--overwrite` / `overwrite` | Overwrite existing files. | `false` |
|
||||
| `--exclude-tags` / `exclude_tags` | Comma-separated tags to exclude. | `null` |
|
||||
| `--truncate` / `truncate` | Maximum length of the file/folder names. | `null` |
|
||||
| `--audio-quality`, `-a` / `audio_quality` | Audio quality for songs and podcasts. | `vorbis-medium` |
|
||||
| `--download-mode`, `-d` / `download_mode` | Download mode for songs and podcasts. | `ytdlp` |
|
||||
| `--remux-mode-audio` / `remux_mode_audio` | Remux mode for songs and podcasts. | `ffmpeg` |
|
||||
| `--lrc-only`, `-l` / `lrc_only` | Download only the synced lyrics. | `false` |
|
||||
| `--no-lrc` / `no_lrc` | Don't download the synced lyrics. | `false` |
|
||||
| `--video-format` / `video_format` | Video format. | `mp4` |
|
||||
| `--remux-mode-video` / `remux_mode_video` | Remux mode for videos. | `ffmpeg` |
|
||||
| `--no-config-file`, `-n` / - | Do not use a config file. | `false` |
|
||||
|
||||
|
||||
|
||||
### Tag variables
|
||||
The following variables can be used in the template folder/file and/or in the `exclude_tags` list:
|
||||
- `album`
|
||||
@@ -108,9 +153,11 @@ The following variables can be used in the template folder/file and/or in the `e
|
||||
- `isrc`
|
||||
- `label`
|
||||
- `lyrics`
|
||||
- `media_type`
|
||||
- `playlist_artist`
|
||||
- `playlist_title`
|
||||
- `playlist_track`
|
||||
- `publisher`
|
||||
- `producer`
|
||||
- `rating`
|
||||
- `release_date`
|
||||
@@ -119,22 +166,41 @@ The following variables can be used in the template folder/file and/or in the `e
|
||||
- `track`
|
||||
- `track_total`
|
||||
- `url`
|
||||
|
||||
### Download qualities
|
||||
|
||||
### Audio qualities
|
||||
The following qualities are available:
|
||||
* `high` (320kbps, requires premium account)
|
||||
* `medium` (160kbps)
|
||||
* `low` (96kbps)
|
||||
* `vorbis-high` (320kbps, requires an active premium subscription)
|
||||
* `vorbis-medium` (160kbps)
|
||||
* `vorbis-low` (96kbps)
|
||||
* `aac-medium` (128kbps)
|
||||
* `aac-high` (256kbps, requires an active premium subscription)
|
||||
|
||||
### Video formats
|
||||
The following video formats are available:
|
||||
* `mp4`
|
||||
* `webm`
|
||||
* `ask`
|
||||
* When using this option, Votify will ask you which audio and video codec to use that is available for the video.
|
||||
|
||||
### Download modes
|
||||
The following modes are available:
|
||||
* `ytdlp`
|
||||
* `aria2c`
|
||||
* Will not be used for downloading videos
|
||||
* Faster than `ytdlp`
|
||||
* Can be obtained from here: https://github.com/aria2/aria2/releases
|
||||
|
||||
### Video remux modes
|
||||
The following remux modes for videos are available:
|
||||
* `ffmpeg`
|
||||
* `mp4box`
|
||||
|
||||
## Credits
|
||||
### Audio remux modes
|
||||
The following remux modes for songs and podcasts are available when downloading in AAC quality:
|
||||
* `ffmpeg`
|
||||
* `mp4box`
|
||||
* `mp4decrypt`
|
||||
|
||||
### Credits
|
||||
* [spotify-oggmp4-dl](https://github.com/DevLARLEY/spotify-oggmp4-dl)
|
||||
* [spsync](https://github.com/baltitenger/spsync)
|
||||
* [unplayplay](https://git.gay/uhwot/unplayplay)
|
||||
* [unplayplay](https://git.gay/uhwot/unplayplay)
|
||||
|
||||
@@ -10,6 +10,7 @@ dependencies = [
|
||||
"protobuf",
|
||||
"pybase62",
|
||||
"pycryptodome",
|
||||
"re-unplayplay",
|
||||
"yt-dlp",
|
||||
]
|
||||
readme = "README.md"
|
||||
|
||||
@@ -4,4 +4,5 @@ pillow
|
||||
protobuf
|
||||
pybase62
|
||||
pycryptodome
|
||||
re-unplayplay
|
||||
yt-dlp
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
__version__ = "1.3.3"
|
||||
__version__ = "1.4"
|
||||
|
||||
+268
-47
@@ -10,18 +10,35 @@ from pathlib import Path
|
||||
import click
|
||||
|
||||
from . import __version__
|
||||
from .constants import EXCLUDED_CONFIG_FILE_PARAMS, X_NOT_FOUND_STRING
|
||||
from .constants import (
|
||||
AAC_AUDIO_QUALITIES,
|
||||
EXCLUDED_CONFIG_FILE_PARAMS,
|
||||
PREMIUM_AUDIO_QUALITIES,
|
||||
X_NOT_FOUND_STRING,
|
||||
)
|
||||
from .downloader import Downloader
|
||||
from .downloader_audio import DownloaderAudio
|
||||
from .downloader_episode import DownloaderEpisode
|
||||
from .downloader_episode_video import DownloaderEpisodeVideo
|
||||
from .downloader_music_video import DownloaderMusicVideo
|
||||
from .downloader_song import DownloaderSong
|
||||
from .enums import DownloadMode, Quality
|
||||
from .downloader_video import DownloaderVideo
|
||||
from .enums import (
|
||||
AudioQuality,
|
||||
DownloadMode,
|
||||
RemuxModeAudio,
|
||||
RemuxModeVideo,
|
||||
VideoFormat,
|
||||
)
|
||||
from .spotify_api import SpotifyApi
|
||||
|
||||
logger = logging.getLogger("votify")
|
||||
|
||||
spotify_api_sig = inspect.signature(SpotifyApi.__init__)
|
||||
downloader_sig = inspect.signature(Downloader.__init__)
|
||||
downloader_audio_sig = inspect.signature(DownloaderAudio.__init__)
|
||||
downloader_song_sig = inspect.signature(DownloaderSong.__init__)
|
||||
downloader_video_sig = inspect.signature(DownloaderVideo.__init__)
|
||||
|
||||
|
||||
def get_param_string(param: click.Parameter) -> str:
|
||||
@@ -77,9 +94,24 @@ def load_config_file(
|
||||
"--wait-interval",
|
||||
"-w",
|
||||
type=float,
|
||||
default=10,
|
||||
default=5,
|
||||
help="Wait interval between downloads in seconds.",
|
||||
)
|
||||
@click.option(
|
||||
"--enable-videos",
|
||||
is_flag=True,
|
||||
help="Enable video downloads when available.",
|
||||
)
|
||||
@click.option(
|
||||
"--download-music-videos",
|
||||
is_flag=True,
|
||||
help="List and select a related music video to download from songs.",
|
||||
)
|
||||
@click.option(
|
||||
"--download-podcast-videos",
|
||||
is_flag=True,
|
||||
help="Attempt to download the video version of podcasts.",
|
||||
)
|
||||
@click.option(
|
||||
"--force-premium",
|
||||
"-f",
|
||||
@@ -105,9 +137,9 @@ def load_config_file(
|
||||
help="Log level.",
|
||||
)
|
||||
@click.option(
|
||||
"--print-exceptions",
|
||||
"--no-exceptions",
|
||||
is_flag=True,
|
||||
help="Print exceptions.",
|
||||
help="Don't print exceptions.",
|
||||
)
|
||||
# SpotifyApi specific options
|
||||
@click.option(
|
||||
@@ -117,13 +149,6 @@ def load_config_file(
|
||||
help="Path to cookies file.",
|
||||
)
|
||||
# Downloader specific options
|
||||
@click.option(
|
||||
"--quality",
|
||||
"-q",
|
||||
type=Quality,
|
||||
default=downloader_sig.parameters["quality"].default,
|
||||
help="Audio quality.",
|
||||
)
|
||||
@click.option(
|
||||
"--output-path",
|
||||
"-o",
|
||||
@@ -138,11 +163,10 @@ def load_config_file(
|
||||
help="Path to temporary directory.",
|
||||
)
|
||||
@click.option(
|
||||
"--download-mode",
|
||||
"-d",
|
||||
type=DownloadMode,
|
||||
default=downloader_sig.parameters["download_mode"].default,
|
||||
help="Download mode.",
|
||||
"--wvd-path",
|
||||
type=Path,
|
||||
default=downloader_sig.parameters["wvd_path"].default,
|
||||
help="Path to .wvd file.",
|
||||
)
|
||||
@click.option(
|
||||
"--aria2c-path",
|
||||
@@ -151,10 +175,28 @@ def load_config_file(
|
||||
help="Path to aria2c binary.",
|
||||
)
|
||||
@click.option(
|
||||
"--unplayplay-path",
|
||||
"--ffmpeg-path",
|
||||
type=str,
|
||||
default=downloader_sig.parameters["unplayplay_path"].default,
|
||||
help="Path to unplayplay binary.",
|
||||
default=downloader_sig.parameters["ffmpeg_path"].default,
|
||||
help="Path to ffmpeg binary.",
|
||||
)
|
||||
@click.option(
|
||||
"--mp4box-path",
|
||||
type=str,
|
||||
default=downloader_sig.parameters["mp4box_path"].default,
|
||||
help="Path to MP4Box binary.",
|
||||
)
|
||||
@click.option(
|
||||
"--mp4decrypt-path",
|
||||
type=str,
|
||||
default=downloader_sig.parameters["mp4decrypt_path"].default,
|
||||
help="Path to mp4decrypt binary.",
|
||||
)
|
||||
@click.option(
|
||||
"--packager-path",
|
||||
type=str,
|
||||
default=downloader_sig.parameters["packager_path"].default,
|
||||
help="Path to Shaka Packager binary.",
|
||||
)
|
||||
@click.option(
|
||||
"--template-folder-album",
|
||||
@@ -190,7 +232,19 @@ def load_config_file(
|
||||
"--template-file-episode",
|
||||
type=str,
|
||||
default=downloader_sig.parameters["template_file_episode"].default,
|
||||
help="Template file for episodes (podcasts).",
|
||||
help="Template file for music videos.",
|
||||
)
|
||||
@click.option(
|
||||
"--template-folder-music-video",
|
||||
type=str,
|
||||
default=downloader_sig.parameters["template_folder_music_video"].default,
|
||||
help="Template folder for music videos",
|
||||
)
|
||||
@click.option(
|
||||
"--template-file-music-video",
|
||||
type=str,
|
||||
default=downloader_sig.parameters["template_file_music_video"].default,
|
||||
help="Template file for the tracks that are not part of an album.",
|
||||
)
|
||||
@click.option(
|
||||
"--template-file-playlist",
|
||||
@@ -206,7 +260,6 @@ def load_config_file(
|
||||
)
|
||||
@click.option(
|
||||
"--save-cover",
|
||||
"-s",
|
||||
is_flag=True,
|
||||
help="Save cover as a separate file.",
|
||||
)
|
||||
@@ -232,6 +285,27 @@ def load_config_file(
|
||||
default=downloader_sig.parameters["truncate"].default,
|
||||
help="Maximum length of the file/folder names.",
|
||||
)
|
||||
# DownloaderAudio specific options
|
||||
@click.option(
|
||||
"--audio-quality",
|
||||
"-a",
|
||||
type=AudioQuality,
|
||||
default=downloader_audio_sig.parameters["audio_quality"].default,
|
||||
help="Audio quality for songs and podcasts.",
|
||||
)
|
||||
@click.option(
|
||||
"--download-mode",
|
||||
"-d",
|
||||
type=DownloadMode,
|
||||
default=downloader_audio_sig.parameters["download_mode"].default,
|
||||
help="Download mode for songs and podcasts.",
|
||||
)
|
||||
@click.option(
|
||||
"--remux-mode-audio",
|
||||
type=RemuxModeAudio,
|
||||
default=downloader_audio_sig.parameters["remux_mode"].default,
|
||||
help="Remux mode for songs and podcasts.",
|
||||
)
|
||||
# DownloaderSong specific options
|
||||
@click.option(
|
||||
"--lrc-only",
|
||||
@@ -244,6 +318,19 @@ def load_config_file(
|
||||
is_flag=True,
|
||||
help="Don't download the synced lyrics.",
|
||||
)
|
||||
# DownloaderVideo specific options
|
||||
@click.option(
|
||||
"--video-format",
|
||||
type=VideoFormat,
|
||||
default=downloader_video_sig.parameters["video_format"].default,
|
||||
help="Video format.",
|
||||
)
|
||||
@click.option(
|
||||
"--remux-mode-video",
|
||||
type=RemuxModeVideo,
|
||||
default=downloader_video_sig.parameters["remux_mode"].default,
|
||||
help="Remux mode for videos.",
|
||||
)
|
||||
# This option should always be last
|
||||
@click.option(
|
||||
"--no-config-file",
|
||||
@@ -255,24 +342,32 @@ def load_config_file(
|
||||
def main(
|
||||
urls: list[str],
|
||||
wait_interval: float,
|
||||
enable_videos: bool,
|
||||
download_music_videos: bool,
|
||||
download_podcast_videos: bool,
|
||||
force_premium: bool,
|
||||
read_urls_as_txt: bool,
|
||||
config_path: Path,
|
||||
log_level: str,
|
||||
print_exceptions: bool,
|
||||
no_exceptions: bool,
|
||||
cookies_path: Path,
|
||||
quality: Quality,
|
||||
output_path: Path,
|
||||
temp_path: Path,
|
||||
wvd_path: Path,
|
||||
download_mode: DownloadMode,
|
||||
aria2c_path: str,
|
||||
unplayplay_path: str,
|
||||
ffmpeg_path: str,
|
||||
mp4box_path: str,
|
||||
mp4decrypt_path: str,
|
||||
packager_path: str,
|
||||
template_folder_album: str,
|
||||
template_folder_compilation: str,
|
||||
template_file_single_disc: str,
|
||||
template_file_multi_disc: str,
|
||||
template_folder_episode: str,
|
||||
template_file_episode: str,
|
||||
template_folder_music_video: str,
|
||||
template_file_music_video: str,
|
||||
template_file_playlist: str,
|
||||
date_tag_template: str,
|
||||
save_cover: bool,
|
||||
@@ -280,8 +375,12 @@ def main(
|
||||
overwrite: bool,
|
||||
exclude_tags: str,
|
||||
truncate: int,
|
||||
audio_quality: AudioQuality,
|
||||
remux_mode_audio: RemuxModeAudio,
|
||||
lrc_only: bool,
|
||||
no_lrc: bool,
|
||||
video_format: VideoFormat,
|
||||
remux_mode_video: RemuxModeVideo,
|
||||
no_config_file: bool,
|
||||
) -> None:
|
||||
logging.basicConfig(
|
||||
@@ -293,18 +392,22 @@ def main(
|
||||
spotify_api = SpotifyApi(cookies_path)
|
||||
downloader = Downloader(
|
||||
spotify_api,
|
||||
quality,
|
||||
output_path,
|
||||
temp_path,
|
||||
download_mode,
|
||||
wvd_path,
|
||||
aria2c_path,
|
||||
unplayplay_path,
|
||||
ffmpeg_path,
|
||||
mp4box_path,
|
||||
mp4decrypt_path,
|
||||
packager_path,
|
||||
template_folder_album,
|
||||
template_folder_compilation,
|
||||
template_file_single_disc,
|
||||
template_file_multi_disc,
|
||||
template_folder_episode,
|
||||
template_file_episode,
|
||||
template_folder_music_video,
|
||||
template_file_music_video,
|
||||
template_file_playlist,
|
||||
date_tag_template,
|
||||
save_cover,
|
||||
@@ -313,27 +416,115 @@ def main(
|
||||
exclude_tags,
|
||||
truncate,
|
||||
)
|
||||
downloader_song = DownloaderSong(
|
||||
downloader_audio = DownloaderAudio(
|
||||
downloader,
|
||||
audio_quality,
|
||||
download_mode,
|
||||
remux_mode_audio,
|
||||
)
|
||||
downloader_song = DownloaderSong(
|
||||
downloader_audio,
|
||||
lrc_only,
|
||||
no_lrc,
|
||||
)
|
||||
downloader_episode = DownloaderEpisode(
|
||||
downloader_audio,
|
||||
)
|
||||
downloader_video = DownloaderVideo(
|
||||
downloader,
|
||||
video_format,
|
||||
remux_mode_video,
|
||||
)
|
||||
downloader_episode_video = DownloaderEpisodeVideo(
|
||||
downloader_video,
|
||||
downloader_episode,
|
||||
)
|
||||
downloader_music_video = DownloaderMusicVideo(
|
||||
downloader_video,
|
||||
)
|
||||
spotify_api.config_info["isPremium"] = (
|
||||
True if force_premium else spotify_api.config_info["isPremium"]
|
||||
)
|
||||
if not lrc_only:
|
||||
if not downloader.unplayplay_path_full:
|
||||
logger.critical(X_NOT_FOUND_STRING.format("Unplayplay", unplayplay_path))
|
||||
return
|
||||
if audio_quality in AAC_AUDIO_QUALITIES:
|
||||
if (
|
||||
remux_mode_audio == RemuxModeAudio.FFMPEG
|
||||
and not downloader.ffmpeg_path_full
|
||||
):
|
||||
logger.critical(X_NOT_FOUND_STRING.format("FFmpeg", ffmpeg_path))
|
||||
if (
|
||||
remux_mode_audio == RemuxModeAudio.MP4BOX
|
||||
and not downloader.mp4box_path_full
|
||||
):
|
||||
logger.critical(X_NOT_FOUND_STRING.format("MP4Box", mp4box_path))
|
||||
return
|
||||
if (
|
||||
remux_mode_audio
|
||||
in (
|
||||
RemuxModeAudio.MP4DECRYPT,
|
||||
RemuxModeAudio.MP4BOX,
|
||||
)
|
||||
and not downloader.mp4decrypt_path_full
|
||||
):
|
||||
logger.critical(
|
||||
X_NOT_FOUND_STRING.format("mp4decrypt", mp4decrypt_path)
|
||||
)
|
||||
return
|
||||
if not wvd_path.exists() and audio_quality in AAC_AUDIO_QUALITIES:
|
||||
logger.critical(
|
||||
X_NOT_FOUND_STRING.format(".wvd", wvd_path)
|
||||
+ ", a .wvd file is required for downloading in AAC quality"
|
||||
)
|
||||
return
|
||||
downloader.set_cdm()
|
||||
if download_mode == DownloadMode.ARIA2C and not downloader.aria2c_path_full:
|
||||
logger.critical(X_NOT_FOUND_STRING.format("aria2c", aria2c_path))
|
||||
return
|
||||
spotify_api.config_info["isPremium"] = (
|
||||
True if force_premium else spotify_api.config_info["isPremium"]
|
||||
)
|
||||
if not spotify_api.config_info["isPremium"] and quality == Quality.HIGH:
|
||||
if (
|
||||
not spotify_api.config_info["isPremium"]
|
||||
and audio_quality in PREMIUM_AUDIO_QUALITIES
|
||||
):
|
||||
logger.critical("Cannot download at chosen quality with a free account")
|
||||
return
|
||||
can_download_music_videos = True
|
||||
if enable_videos:
|
||||
if (
|
||||
downloader_music_video.remux_mode == RemuxModeVideo.FFMPEG
|
||||
and not downloader.ffmpeg_path_full
|
||||
):
|
||||
logger.critical(X_NOT_FOUND_STRING.format("FFmpeg", ffmpeg_path))
|
||||
return
|
||||
if (
|
||||
downloader_music_video.remux_mode == RemuxModeVideo.MP4BOX
|
||||
and not downloader.mp4box_path_full
|
||||
):
|
||||
logger.critical(X_NOT_FOUND_STRING.format("MP4Box", mp4box_path))
|
||||
return
|
||||
music_video_warning_message = []
|
||||
if not downloader.mp4decrypt_path_full and video_format != VideoFormat.WEBM:
|
||||
music_video_warning_message.append(
|
||||
X_NOT_FOUND_STRING.format("mp4decrypt", mp4decrypt_path)
|
||||
)
|
||||
elif not downloader.packager_path_full:
|
||||
music_video_warning_message.append(
|
||||
X_NOT_FOUND_STRING.format("Shaka Packager", packager_path)
|
||||
)
|
||||
if not wvd_path.exists():
|
||||
music_video_warning_message.append(
|
||||
X_NOT_FOUND_STRING.format(".wvd file", wvd_path)
|
||||
)
|
||||
else:
|
||||
downloader.set_cdm()
|
||||
if not spotify_api.config_info["isPremium"]:
|
||||
music_video_warning_message.append(
|
||||
"Cannot download music videos with a non-premium account"
|
||||
)
|
||||
if music_video_warning_message:
|
||||
logger.warning(
|
||||
"Music videos will not be downloaded due to the following reasons:\n"
|
||||
+ "\n".join(music_video_warning_message)
|
||||
)
|
||||
can_download_music_videos = False
|
||||
error_count = 0
|
||||
if read_urls_as_txt:
|
||||
_urls = []
|
||||
@@ -357,7 +548,7 @@ def main(
|
||||
error_count += 1
|
||||
logger.error(
|
||||
f'({url_progress}) Failed to check "{url}"',
|
||||
exc_info=print_exceptions,
|
||||
exc_info=no_exceptions,
|
||||
)
|
||||
continue
|
||||
for index, download_queue_item in enumerate(download_queue, start=1):
|
||||
@@ -371,27 +562,57 @@ def main(
|
||||
)
|
||||
media_id = downloader.get_media_id(media_metadata)
|
||||
media_type = media_metadata["type"]
|
||||
if media_type == "track":
|
||||
gid_metadata = downloader.get_gid_metadata(media_id, media_type)
|
||||
if gid_metadata.get("original_video") or (
|
||||
media_type == "track" and download_music_videos
|
||||
):
|
||||
if not enable_videos or not can_download_music_videos:
|
||||
logger.warning(
|
||||
"Music videos are not downloadable with current "
|
||||
"configuration, skipping"
|
||||
)
|
||||
continue
|
||||
downloader_music_video.download(
|
||||
music_video_id=media_id,
|
||||
music_video_metadata=media_metadata,
|
||||
album_metadata=download_queue_item.album_metadata,
|
||||
gid_metadata=gid_metadata,
|
||||
playlist_metadata=download_queue_item.playlist_metadata,
|
||||
playlist_track=index,
|
||||
)
|
||||
elif media_type == "track":
|
||||
downloader_song.download(
|
||||
track_id=media_id,
|
||||
track_metadata=media_metadata,
|
||||
album_metadata=download_queue_item.album_metadata,
|
||||
gid_metadata=gid_metadata,
|
||||
playlist_metadata=download_queue_item.playlist_metadata,
|
||||
playlist_track=index,
|
||||
)
|
||||
elif media_type == "episode" and not lrc_only:
|
||||
downloader_episode.download(
|
||||
episode_id=media_id,
|
||||
episode_metadata=media_metadata,
|
||||
show_metadata=download_queue_item.show_metadata,
|
||||
playlist_metadata=download_queue_item.playlist_metadata,
|
||||
playlist_track=index,
|
||||
)
|
||||
elif media_type == "episode":
|
||||
if enable_videos and download_podcast_videos:
|
||||
downloader_episode_video.download(
|
||||
episode_id=media_id,
|
||||
episode_metadata=media_metadata,
|
||||
show_metadata=download_queue_item.show_metadata,
|
||||
gid_metadata=gid_metadata,
|
||||
playlist_metadata=download_queue_item.playlist_metadata,
|
||||
playlist_track=index,
|
||||
)
|
||||
else:
|
||||
downloader_episode.download(
|
||||
episode_id=media_id,
|
||||
episode_metadata=media_metadata,
|
||||
show_metadata=download_queue_item.show_metadata,
|
||||
gid_metadata=gid_metadata,
|
||||
playlist_metadata=download_queue_item.playlist_metadata,
|
||||
playlist_track=index,
|
||||
)
|
||||
except Exception as e:
|
||||
error_count += 1
|
||||
logger.error(
|
||||
f'({queue_progress}) Failed to download "{media_metadata["name"]}"',
|
||||
exc_info=print_exceptions,
|
||||
exc_info=not no_exceptions,
|
||||
)
|
||||
finally:
|
||||
if wait_interval > 0 and index != len(download_queue):
|
||||
|
||||
+37
-5
@@ -1,6 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from .enums import Quality
|
||||
from .enums import AudioQuality
|
||||
|
||||
EXCLUDED_CONFIG_FILE_PARAMS = (
|
||||
"urls",
|
||||
@@ -32,10 +32,42 @@ VORBIS_TAGS_MAPPING = {
|
||||
"url": "URL",
|
||||
}
|
||||
|
||||
QUALITY_X_FORMAT_ID_MAPPING = {
|
||||
Quality.HIGH: "OGG_VORBIS_320",
|
||||
Quality.MEDIUM: "OGG_VORBIS_160",
|
||||
Quality.LOW: "OGG_VORBIS_96",
|
||||
MP4_TAGS_MAP = {
|
||||
"album": "\xa9alb",
|
||||
"album_artist": "aART",
|
||||
"artist": "\xa9ART",
|
||||
"composer": "\xa9wrt",
|
||||
"copyright": "cprt",
|
||||
"lyrics": "\xa9lyr",
|
||||
"publisher": "\xa9pub",
|
||||
"producer": "\xa9prd",
|
||||
"rating": "rtng",
|
||||
"release_date": "\xa9day",
|
||||
"title": "\xa9nam",
|
||||
"url": "\xa9url",
|
||||
}
|
||||
|
||||
AUDIO_QUALITY_X_FORMAT_ID_MAPPING = {
|
||||
AudioQuality.VORBIS_HIGH: "OGG_VORBIS_320",
|
||||
AudioQuality.VORBIS_MEDIUM: "OGG_VORBIS_160",
|
||||
AudioQuality.VORBIS_LOW: "OGG_VORBIS_96",
|
||||
AudioQuality.AAC_HIGH: "MP4_256",
|
||||
AudioQuality.AAC_MEDIUM: "MP4_128",
|
||||
}
|
||||
|
||||
VORBIS_AUDIO_QUALITIES = (
|
||||
AudioQuality.VORBIS_HIGH,
|
||||
AudioQuality.VORBIS_MEDIUM,
|
||||
AudioQuality.VORBIS_LOW,
|
||||
)
|
||||
AAC_AUDIO_QUALITIES = (AudioQuality.AAC_HIGH, AudioQuality.AAC_MEDIUM)
|
||||
|
||||
X_NOT_FOUND_STRING = "{} not found at {}"
|
||||
|
||||
PREMIUM_AUDIO_QUALITIES = (AudioQuality.VORBIS_HIGH, AudioQuality.AAC_HIGH)
|
||||
|
||||
MEDIA_TYPE_MP4_MAPPING = {
|
||||
"Song": 1,
|
||||
"Podcast": 21,
|
||||
"Music video": 6,
|
||||
}
|
||||
|
||||
+149
-130
@@ -3,24 +3,25 @@ from __future__ import annotations
|
||||
import base64
|
||||
import datetime
|
||||
import functools
|
||||
import logging
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
from re_unplayplay import decrypt_and_bind_key
|
||||
import requests
|
||||
from Crypto.Cipher import AES
|
||||
from InquirerPy import inquirer
|
||||
from InquirerPy.base.control import Choice
|
||||
from mutagen.flac import Picture
|
||||
from mutagen.mp4 import MP4, MP4Cover, MP4FreeForm
|
||||
from mutagen.oggvorbis import OggVorbis, OggVorbisHeaderError
|
||||
from PIL import Image
|
||||
from yt_dlp import YoutubeDL
|
||||
from pywidevine import PSSH, Cdm, Device
|
||||
|
||||
from .constants import QUALITY_X_FORMAT_ID_MAPPING, VORBIS_TAGS_MAPPING
|
||||
from .enums import DownloadMode, Quality
|
||||
from .models import DownloadQueueItem, StreamInfo, UrlInfo
|
||||
from .constants import MEDIA_TYPE_MP4_MAPPING, MP4_TAGS_MAP, VORBIS_TAGS_MAPPING
|
||||
from .models import DownloadQueueItem, UrlInfo
|
||||
from .playplay_pb2 import (
|
||||
AUDIO_TRACK,
|
||||
Interactivity,
|
||||
@@ -30,6 +31,8 @@ from .playplay_pb2 import (
|
||||
from .spotify_api import SpotifyApi
|
||||
from .utils import check_response
|
||||
|
||||
logger = logging.getLogger("votify")
|
||||
|
||||
|
||||
class Downloader:
|
||||
ILLEGAL_CHARACTERS_REGEX = r'[\\/:*?"<>|;]'
|
||||
@@ -44,18 +47,22 @@ class Downloader:
|
||||
def __init__(
|
||||
self,
|
||||
spotify_api: SpotifyApi,
|
||||
quality: Quality = Quality.MEDIUM,
|
||||
output_path: Path = Path("./Spotify"),
|
||||
temp_path: Path = Path("./temp"),
|
||||
download_mode: DownloadMode = DownloadMode.YTDLP,
|
||||
aria2c_path: Path = "aria2c",
|
||||
unplayplay_path: Path = "unplayplay",
|
||||
wvd_path: Path = Path("./device.wvd"),
|
||||
aria2c_path: str = "aria2c",
|
||||
ffmpeg_path: str = "ffmpeg",
|
||||
mp4box_path: str = "mp4box",
|
||||
mp4decrypt_path: str = "mp4decrypt",
|
||||
packager_path: str = "packager",
|
||||
template_folder_album: str = "{album_artist}/{album}",
|
||||
template_folder_compilation: str = "Compilations/{album}",
|
||||
template_file_single_disc: str = "{track:02d} {title}",
|
||||
template_file_multi_disc: str = "{disc}-{track:02d} {title}",
|
||||
template_folder_episode: str = "Podcasts/{album}",
|
||||
template_file_episode: str = "{track:02d} {title}",
|
||||
template_folder_music_video: str = "{artist}/Unknown Album",
|
||||
template_file_music_video: str = "{title}",
|
||||
template_file_playlist: str = "Playlists/{playlist_artist}/{playlist_title}",
|
||||
date_tag_template: str = "%Y-%m-%dT%H:%M:%SZ",
|
||||
save_cover: bool = False,
|
||||
@@ -64,20 +71,25 @@ class Downloader:
|
||||
exclude_tags: str = None,
|
||||
truncate: int = None,
|
||||
silence: bool = False,
|
||||
skip_cleanup: bool = False,
|
||||
):
|
||||
self.spotify_api = spotify_api
|
||||
self.quality = quality
|
||||
self.output_path = output_path
|
||||
self.temp_path = temp_path
|
||||
self.download_mode = download_mode
|
||||
self.wvd_path = wvd_path
|
||||
self.aria2c_path = aria2c_path
|
||||
self.unplayplay_path = unplayplay_path
|
||||
self.ffmpeg_path = ffmpeg_path
|
||||
self.mp4box_path = mp4box_path
|
||||
self.mp4decrypt_path = mp4decrypt_path
|
||||
self.packager_path = packager_path
|
||||
self.template_folder_album = template_folder_album
|
||||
self.template_folder_compilation = template_folder_compilation
|
||||
self.template_file_single_disc = template_file_single_disc
|
||||
self.template_file_multi_disc = template_file_multi_disc
|
||||
self.template_folder_episode = template_folder_episode
|
||||
self.template_file_episode = template_file_episode
|
||||
self.template_folder_music_video = template_folder_music_video
|
||||
self.template_file_music_video = template_file_music_video
|
||||
self.template_file_playlist = template_file_playlist
|
||||
self.date_tag_template = date_tag_template
|
||||
self.save_cover = save_cover
|
||||
@@ -86,6 +98,7 @@ class Downloader:
|
||||
self.exclude_tags = exclude_tags
|
||||
self.truncate = truncate
|
||||
self.silence = silence
|
||||
self.skip_cleanup = skip_cleanup
|
||||
self._set_binaries_full_path()
|
||||
self._set_exclude_tags_list()
|
||||
self._set_truncate()
|
||||
@@ -93,7 +106,10 @@ class Downloader:
|
||||
|
||||
def _set_binaries_full_path(self):
|
||||
self.aria2c_path_full = shutil.which(self.aria2c_path)
|
||||
self.unplayplay_path_full = shutil.which(self.unplayplay_path)
|
||||
self.ffmpeg_path_full = shutil.which(self.ffmpeg_path)
|
||||
self.mp4box_path_full = shutil.which(self.mp4box_path)
|
||||
self.mp4decrypt_path_full = shutil.which(self.mp4decrypt_path)
|
||||
self.packager_path_full = shutil.which(self.packager_path)
|
||||
|
||||
def _set_exclude_tags_list(self):
|
||||
self.exclude_tags_list = (
|
||||
@@ -115,6 +131,9 @@ class Downloader:
|
||||
else:
|
||||
self.subprocess_additional_args = {}
|
||||
|
||||
def set_cdm(self) -> None:
|
||||
self.cdm = Cdm.from_device(Device.load(self.wvd_path))
|
||||
|
||||
def get_url_info(self, url: str) -> UrlInfo:
|
||||
url_regex_result = re.search(self.URL_RE, url)
|
||||
if url_regex_result is None:
|
||||
@@ -289,6 +308,9 @@ class Downloader:
|
||||
elif media_type == "episode":
|
||||
template_folder = self.template_folder_episode.split("/")
|
||||
template_file = self.template_file_episode.split("/")
|
||||
elif media_type == "music-video":
|
||||
template_folder = self.template_folder_music_video.split("/")
|
||||
template_file = self.template_file_music_video.split("/")
|
||||
else:
|
||||
raise RuntimeError()
|
||||
template_final = template_folder + template_file
|
||||
@@ -330,18 +352,6 @@ class Downloader:
|
||||
with playlist_file_path.open("w", encoding="utf8") as playlist_file:
|
||||
playlist_file.writelines(playlist_file_lines)
|
||||
|
||||
def get_audio_file(
|
||||
self,
|
||||
audio_files: list[dict],
|
||||
) -> tuple[Quality, dict] | tuple[None, None]:
|
||||
qualities = list(Quality)
|
||||
start_index = qualities.index(self.quality)
|
||||
for quality in qualities[start_index:]:
|
||||
for audio_file in audio_files:
|
||||
if audio_file["format"] == QUALITY_X_FORMAT_ID_MAPPING[quality]:
|
||||
return quality, audio_file
|
||||
return None, None
|
||||
|
||||
def get_gid_metadata(
|
||||
self,
|
||||
media_id: str,
|
||||
@@ -350,34 +360,7 @@ class Downloader:
|
||||
gid = self.spotify_api.media_id_to_gid(media_id)
|
||||
return self.spotify_api.get_gid_metadata(gid, media_type)
|
||||
|
||||
def get_stream_info(
|
||||
self,
|
||||
gid_metadata: dict,
|
||||
media_type: str,
|
||||
) -> StreamInfo:
|
||||
stream_info = StreamInfo()
|
||||
if media_type == "track":
|
||||
audio_files = gid_metadata.get("file")
|
||||
elif media_type == "episode":
|
||||
audio_files = gid_metadata.get("audio")
|
||||
else:
|
||||
raise RuntimeError()
|
||||
audio_files = audio_files or gid_metadata.get("alternative")
|
||||
if not audio_files:
|
||||
return stream_info
|
||||
if audio_files[0].get("gid"):
|
||||
audio_files = audio_files[0]["file"]
|
||||
quality, audio_file = self.get_audio_file(audio_files)
|
||||
if not audio_file:
|
||||
return stream_info
|
||||
file_id = audio_file["file_id"]
|
||||
stream_url = self.spotify_api.get_stream_urls(file_id)["cdnurl"][0]
|
||||
stream_info.stream_url = stream_url
|
||||
stream_info.file_id = file_id
|
||||
stream_info.quality = quality
|
||||
return stream_info
|
||||
|
||||
def get_decryption_key(self, file_id: str) -> bytes:
|
||||
def get_playplay_decryption_key(self, file_id: str) -> bytes:
|
||||
playplay_license_request = PlayPlayLicenseRequest(
|
||||
version=2,
|
||||
token=bytes.fromhex("01e132cae527bd21620e822f58514932"),
|
||||
@@ -390,74 +373,29 @@ class Downloader:
|
||||
)
|
||||
playplay_license_response = PlayPlayLicenseResponse()
|
||||
playplay_license_response.ParseFromString(playplay_license_response_bytes)
|
||||
obfuscated = playplay_license_response.obfuscated_key.hex()
|
||||
output = subprocess.check_output(
|
||||
[
|
||||
self.unplayplay_path_full,
|
||||
file_id,
|
||||
obfuscated,
|
||||
],
|
||||
shell=False,
|
||||
)
|
||||
key = bytes.fromhex(output.strip().decode("utf-8"))
|
||||
assert key
|
||||
obfuscated_key = playplay_license_response.obfuscated_key
|
||||
key = decrypt_and_bind_key(obfuscated_key, file_id)
|
||||
return key
|
||||
|
||||
def download_stream_url(self, input_path: Path, stream_url: str):
|
||||
if self.download_mode == DownloadMode.YTDLP:
|
||||
self.download_stream_url_ytdlp(input_path, stream_url)
|
||||
elif self.download_mode == DownloadMode.ARIA2C:
|
||||
self.download_stream_url_aria2c(input_path, stream_url)
|
||||
|
||||
def download_stream_url_ytdlp(self, input_path: Path, stream_url: str) -> None:
|
||||
with YoutubeDL(
|
||||
{
|
||||
"quiet": True,
|
||||
"no_warnings": True,
|
||||
"outtmpl": str(input_path),
|
||||
"allow_unplayable_formats": True,
|
||||
"fixup": "never",
|
||||
"allowed_extractors": ["generic"],
|
||||
"noprogress": self.silence,
|
||||
}
|
||||
) as ydl:
|
||||
ydl.download(stream_url)
|
||||
|
||||
def download_stream_url_aria2c(self, input_path: Path, stream_url: str) -> None:
|
||||
input_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
subprocess.run(
|
||||
[
|
||||
self.aria2c_path_full,
|
||||
"--no-conf",
|
||||
"--download-result=hide",
|
||||
"--console-log-level=error",
|
||||
"--summary-interval=0",
|
||||
"--file-allocation=none",
|
||||
stream_url,
|
||||
"--out",
|
||||
input_path,
|
||||
],
|
||||
check=True,
|
||||
**self.subprocess_additional_args,
|
||||
)
|
||||
print("\r", end="")
|
||||
|
||||
def decrypt(
|
||||
def get_widevine_decryption_key(
|
||||
self,
|
||||
decryption_key: bytes,
|
||||
encrypted_path: Path,
|
||||
decrypted_path: Path,
|
||||
):
|
||||
cipher = AES.new(
|
||||
decryption_key,
|
||||
AES.MODE_CTR,
|
||||
nonce=bytes.fromhex("72e067fbddcbcf77"),
|
||||
initial_value=bytes.fromhex("ebe8bc643f630d93"),
|
||||
)
|
||||
skip = 167
|
||||
with decrypted_path.open("wb") as decrypted_file:
|
||||
with encrypted_path.open("rb") as encrypted_file:
|
||||
decrypted_file.write(cipher.decrypt(encrypted_file.read())[skip:])
|
||||
pssh: bytes,
|
||||
media_type: str,
|
||||
) -> tuple[str, str]:
|
||||
try:
|
||||
pssh = PSSH(pssh)
|
||||
cdm_session = self.cdm.open()
|
||||
challenge = self.cdm.get_license_challenge(cdm_session, pssh)
|
||||
license = self.spotify_api.get_widevine_license(challenge, media_type)
|
||||
self.cdm.parse_license(cdm_session, license)
|
||||
keys = next(
|
||||
i for i in self.cdm.get_keys(cdm_session) if i.type == "CONTENT"
|
||||
)
|
||||
decryption_key = keys.key.hex()
|
||||
key_id = keys.kid.hex
|
||||
finally:
|
||||
self.cdm.close(cdm_session)
|
||||
return key_id, decryption_key
|
||||
|
||||
def get_sanitized_string(self, dirty_string: str, is_folder: bool) -> str:
|
||||
dirty_string = re.sub(
|
||||
@@ -503,19 +441,15 @@ class Downloader:
|
||||
def _get_cover_url(self, images_dict: list[dict]) -> str:
|
||||
return max(images_dict, key=lambda img: img["height"])["url"]
|
||||
|
||||
def get_encrypted_path(
|
||||
def get_file_temp_path(
|
||||
self,
|
||||
track_id: str,
|
||||
extra_string: str,
|
||||
extension: str,
|
||||
) -> Path:
|
||||
return self.temp_path / (f"{track_id}_encrypted.ogg")
|
||||
return self.temp_path / (track_id + extra_string + extension)
|
||||
|
||||
def get_decrypted_path(
|
||||
self,
|
||||
track_id: str,
|
||||
) -> Path:
|
||||
return self.temp_path / (f"{track_id}_decrypted.ogg")
|
||||
|
||||
def apply_tags(
|
||||
def apply_tags_ogg(
|
||||
self,
|
||||
input_path: Path,
|
||||
tags: dict,
|
||||
@@ -544,6 +478,94 @@ class Downloader:
|
||||
except OggVorbisHeaderError:
|
||||
pass
|
||||
|
||||
def apply_tags_mp4(self, fixed_location: Path, tags: dict, cover_url: str):
|
||||
to_apply_tags = [
|
||||
tag_name
|
||||
for tag_name in tags.keys()
|
||||
if tag_name not in self.exclude_tags_list
|
||||
]
|
||||
mp4_tags = {}
|
||||
for tag_name in to_apply_tags:
|
||||
if tags.get(tag_name) is None:
|
||||
continue
|
||||
if tag_name in ("disc", "disc_total"):
|
||||
if mp4_tags.get("disk") is None:
|
||||
mp4_tags["disk"] = [[0, 0]]
|
||||
if tag_name == "disc":
|
||||
mp4_tags["disk"][0][0] = tags[tag_name]
|
||||
elif tag_name == "disc_total":
|
||||
mp4_tags["disk"][0][1] = tags[tag_name]
|
||||
elif tag_name in ("track", "track_total"):
|
||||
if mp4_tags.get("trkn") is None:
|
||||
mp4_tags["trkn"] = [[0, 0]]
|
||||
if tag_name == "track":
|
||||
mp4_tags["trkn"][0][0] = tags[tag_name]
|
||||
elif tag_name == "track_total":
|
||||
mp4_tags["trkn"][0][1] = tags[tag_name]
|
||||
elif tag_name == "compilation":
|
||||
mp4_tags["cpil"] = tags["compilation"]
|
||||
elif tag_name == "isrc":
|
||||
mp4_tags["----:com.apple.iTunes:ISRC"] = [
|
||||
MP4FreeForm(tags["isrc"].encode("utf-8"))
|
||||
]
|
||||
elif tag_name == "label":
|
||||
mp4_tags["----:com.apple.iTunes:LABEL"] = [
|
||||
MP4FreeForm(tags["label"].encode("utf-8"))
|
||||
]
|
||||
elif tag_name == "rating":
|
||||
mp4_tags["rtng"] = [1] if tags[tag_name] == "Explicit" else [0]
|
||||
elif tag_name == "media_type":
|
||||
mp4_tags["stik"] = [MEDIA_TYPE_MP4_MAPPING[tags[tag_name]]]
|
||||
elif MP4_TAGS_MAP.get(tag_name) is not None:
|
||||
mp4_tags[MP4_TAGS_MAP[tag_name]] = [tags[tag_name]]
|
||||
if "cover" not in self.exclude_tags_list and cover_url is not None:
|
||||
mp4_tags["covr"] = [
|
||||
MP4Cover(
|
||||
self.get_response_bytes(cover_url), imageformat=MP4Cover.FORMAT_JPEG
|
||||
)
|
||||
]
|
||||
mp4 = MP4(fixed_location)
|
||||
mp4.clear()
|
||||
mp4.update(mp4_tags)
|
||||
mp4.save()
|
||||
|
||||
def _final_processing(
|
||||
self,
|
||||
cover_path: Path,
|
||||
cover_url: str,
|
||||
media_temp_path: Path,
|
||||
final_path: Path,
|
||||
tags: dict,
|
||||
playlist_metadata: dict,
|
||||
playlist_track: int,
|
||||
):
|
||||
if self.save_cover and cover_path.exists() and not self.overwrite:
|
||||
logger.debug(f'Cover already exists at "{cover_path}", skipping')
|
||||
elif self.save_cover and cover_url is not None:
|
||||
logger.debug(f'Saving cover to "{cover_path}"')
|
||||
self.save_cover_file(cover_path, cover_url)
|
||||
if media_temp_path:
|
||||
logger.debug("Applying tags")
|
||||
if media_temp_path.suffix in (".mp4", ".m4a"):
|
||||
self.apply_tags_mp4(media_temp_path, tags, cover_url)
|
||||
elif media_temp_path.suffix == ".ogg":
|
||||
self.apply_tags_ogg(media_temp_path, tags, cover_url)
|
||||
logger.debug(f'Moving to "{final_path}"')
|
||||
self.move_to_final_path(media_temp_path, final_path)
|
||||
if self.save_playlist and playlist_metadata:
|
||||
playlist_file_path = self.get_playlist_file_path(tags)
|
||||
logger.debug(f'Updating M3U8 playlist from "{playlist_file_path}"')
|
||||
self.update_playlist_file(
|
||||
playlist_file_path,
|
||||
final_path,
|
||||
playlist_track,
|
||||
)
|
||||
|
||||
def cleanup_temp_path(self):
|
||||
if self.temp_path.exists() and not self.skip_cleanup:
|
||||
logger.debug(f'Cleaning up "{self.temp_path}"')
|
||||
shutil.rmtree(self.temp_path)
|
||||
|
||||
@staticmethod
|
||||
@functools.lru_cache()
|
||||
def get_response_bytes(url: str) -> bytes:
|
||||
@@ -560,6 +582,3 @@ class Downloader:
|
||||
if cover_url is not None:
|
||||
cover_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
cover_path.write_bytes(self.get_response_bytes(cover_url))
|
||||
|
||||
def cleanup_temp_path(self):
|
||||
shutil.rmtree(self.temp_path)
|
||||
|
||||
@@ -0,0 +1,248 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from Crypto.Cipher import AES
|
||||
from yt_dlp.downloader.http import HttpFD
|
||||
from yt_dlp.YoutubeDL import YoutubeDL
|
||||
|
||||
from .constants import (
|
||||
AAC_AUDIO_QUALITIES,
|
||||
AUDIO_QUALITY_X_FORMAT_ID_MAPPING,
|
||||
VORBIS_AUDIO_QUALITIES,
|
||||
)
|
||||
from .downloader import Downloader
|
||||
from .enums import AudioQuality, DownloadMode, RemuxModeAudio
|
||||
from .models import StreamInfoAudio
|
||||
|
||||
|
||||
class DownloaderAudio:
|
||||
def __init__(
|
||||
self,
|
||||
downloader: Downloader,
|
||||
audio_quality: AudioQuality = AudioQuality.VORBIS_MEDIUM,
|
||||
download_mode: DownloadMode = DownloadMode.YTDLP,
|
||||
remux_mode: RemuxModeAudio = RemuxModeAudio.FFMPEG,
|
||||
):
|
||||
self.downloader = downloader
|
||||
self.audio_quality = audio_quality
|
||||
self.download_mode = download_mode
|
||||
self.remux_mode = remux_mode
|
||||
|
||||
def get_file_extension(
|
||||
self,
|
||||
) -> str:
|
||||
if self.audio_quality in AAC_AUDIO_QUALITIES:
|
||||
return ".m4a"
|
||||
else:
|
||||
return ".ogg"
|
||||
|
||||
def get_audio_file(
|
||||
self,
|
||||
audio_files: list[dict],
|
||||
) -> tuple[AudioQuality, dict] | tuple[None, None]:
|
||||
if self.audio_quality in AAC_AUDIO_QUALITIES:
|
||||
qualities = AAC_AUDIO_QUALITIES
|
||||
else:
|
||||
qualities = VORBIS_AUDIO_QUALITIES
|
||||
start_index = qualities.index(self.audio_quality)
|
||||
for quality in qualities[start_index:]:
|
||||
for audio_file in audio_files:
|
||||
if audio_file["format"] == AUDIO_QUALITY_X_FORMAT_ID_MAPPING[quality]:
|
||||
return quality, audio_file
|
||||
return None, None
|
||||
|
||||
def get_stream_info(
|
||||
self,
|
||||
gid_metadata: dict,
|
||||
media_type: str,
|
||||
) -> StreamInfoAudio:
|
||||
stream_info = StreamInfoAudio()
|
||||
if media_type == "track":
|
||||
audio_files = gid_metadata.get("file")
|
||||
elif media_type == "episode":
|
||||
audio_files = gid_metadata.get("audio")
|
||||
else:
|
||||
raise RuntimeError()
|
||||
audio_files = audio_files or gid_metadata.get("alternative")
|
||||
if not audio_files:
|
||||
return stream_info
|
||||
if audio_files[0].get("gid"):
|
||||
audio_files = audio_files[0]["file"]
|
||||
quality, audio_file = self.get_audio_file(audio_files)
|
||||
if not audio_file:
|
||||
return stream_info
|
||||
file_id = audio_file["file_id"]
|
||||
stream_url = self.downloader.spotify_api.get_stream_urls(file_id)["cdnurl"][0]
|
||||
stream_info.stream_url = stream_url
|
||||
stream_info.file_id = file_id
|
||||
stream_info.quality = quality
|
||||
if self.audio_quality in AAC_AUDIO_QUALITIES:
|
||||
seek_table = self.downloader.spotify_api.get_seek_table(file_id)
|
||||
pssh = seek_table["pssh"]
|
||||
stream_info.widevine_pssh = pssh
|
||||
return stream_info
|
||||
|
||||
def get_decryption_key(self, stream_info: StreamInfoAudio) -> str:
|
||||
if self.audio_quality in AAC_AUDIO_QUALITIES:
|
||||
_, decryption_key = self.downloader.get_widevine_decryption_key(
|
||||
stream_info.widevine_pssh,
|
||||
"audio",
|
||||
)
|
||||
else:
|
||||
decryption_key = self.downloader.get_playplay_decryption_key(
|
||||
stream_info.file_id
|
||||
)
|
||||
return decryption_key
|
||||
|
||||
def download_stream_url(self, input_path: Path, stream_url: str):
|
||||
if self.download_mode == DownloadMode.YTDLP:
|
||||
self.download_stream_url_ytdlp(input_path, stream_url)
|
||||
elif self.download_mode == DownloadMode.ARIA2C:
|
||||
self.download_stream_url_aria2c(input_path, stream_url)
|
||||
|
||||
def download_stream_url_ytdlp(self, input_path: Path, stream_url: str) -> None:
|
||||
input_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with YoutubeDL(
|
||||
{
|
||||
"quiet": True,
|
||||
"no_warnings": True,
|
||||
"noprogress": self.downloader.silence,
|
||||
}
|
||||
) as ydl:
|
||||
http_downloader = HttpFD(ydl, ydl.params)
|
||||
http_downloader.download(
|
||||
str(input_path),
|
||||
{
|
||||
"url": stream_url,
|
||||
},
|
||||
)
|
||||
|
||||
def download_stream_url_aria2c(self, input_path: Path, stream_url: str) -> None:
|
||||
input_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.aria2c_path_full,
|
||||
"--no-conf",
|
||||
"--download-result=hide",
|
||||
"--console-log-level=error",
|
||||
"--summary-interval=0",
|
||||
"--file-allocation=none",
|
||||
stream_url,
|
||||
"--out",
|
||||
input_path,
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
print("\r", end="")
|
||||
|
||||
def decrypt(
|
||||
self,
|
||||
decryption_key: bytes | str,
|
||||
encrypted_path: Path,
|
||||
decrypted_path: Path,
|
||||
remuxed_path: Path,
|
||||
):
|
||||
if self.audio_quality in AAC_AUDIO_QUALITIES:
|
||||
if self.remux_mode == RemuxModeAudio.FFMPEG:
|
||||
self.decrypt_widevine_ffmpeg(
|
||||
decryption_key,
|
||||
encrypted_path,
|
||||
decrypted_path,
|
||||
)
|
||||
else:
|
||||
self.decrypt_widevine_mp4decrypt(
|
||||
decryption_key,
|
||||
encrypted_path,
|
||||
decrypted_path,
|
||||
)
|
||||
if self.remux_mode == RemuxModeAudio.MP4BOX:
|
||||
self.remux_mp4box(decrypted_path, remuxed_path)
|
||||
else:
|
||||
self.decrypt_playplay(
|
||||
decryption_key,
|
||||
encrypted_path,
|
||||
decrypted_path,
|
||||
)
|
||||
|
||||
def decrypt_playplay(
|
||||
self,
|
||||
decryption_key: bytes,
|
||||
encrypted_path: Path,
|
||||
decrypted_path: Path,
|
||||
):
|
||||
cipher = AES.new(
|
||||
decryption_key,
|
||||
AES.MODE_CTR,
|
||||
nonce=bytes.fromhex("72e067fbddcbcf77"),
|
||||
initial_value=bytes.fromhex("ebe8bc643f630d93"),
|
||||
)
|
||||
skip = 167
|
||||
with decrypted_path.open("wb") as decrypted_file:
|
||||
with encrypted_path.open("rb") as encrypted_file:
|
||||
decrypted_file.write(cipher.decrypt(encrypted_file.read())[skip:])
|
||||
|
||||
def decrypt_widevine_ffmpeg(
|
||||
self,
|
||||
decryption_key: str,
|
||||
encrypted_path: Path,
|
||||
decrypted_path: Path,
|
||||
):
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.ffmpeg_path_full,
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-y",
|
||||
"-decryption_key",
|
||||
decryption_key,
|
||||
"-i",
|
||||
encrypted_path,
|
||||
"-c",
|
||||
"copy",
|
||||
decrypted_path,
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
|
||||
def decrypt_widevine_mp4decrypt(
|
||||
self,
|
||||
decryption_key: str,
|
||||
encrypted_path: Path,
|
||||
decrypted_path: Path,
|
||||
):
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.mp4decrypt_path_full,
|
||||
"--key",
|
||||
f"1:{decryption_key}",
|
||||
encrypted_path,
|
||||
decrypted_path,
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
|
||||
def remux_mp4box(
|
||||
self,
|
||||
decrypted_path: Path,
|
||||
remuxed_path: Path,
|
||||
):
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.mp4box_path_full,
|
||||
"-quiet",
|
||||
"-itags",
|
||||
"artist=placeholder",
|
||||
"-keep-utc",
|
||||
"-add",
|
||||
decrypted_path,
|
||||
"-new",
|
||||
remuxed_path,
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
@@ -3,18 +3,18 @@ from __future__ import annotations
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from .downloader import Downloader
|
||||
from .models import StreamInfo
|
||||
from .downloader_audio import DownloaderAudio
|
||||
from .models import StreamInfoAudio
|
||||
|
||||
logger = logging.getLogger("votify")
|
||||
|
||||
|
||||
class DownloaderEpisode:
|
||||
class DownloaderEpisode(DownloaderAudio):
|
||||
def __init__(
|
||||
self,
|
||||
downloader: Downloader,
|
||||
downloader_audio: DownloaderAudio,
|
||||
):
|
||||
self.downloader = downloader
|
||||
self.__dict__.update(downloader_audio.__dict__)
|
||||
|
||||
def get_tags(
|
||||
self,
|
||||
@@ -28,6 +28,7 @@ class DownloaderEpisode:
|
||||
tags = {
|
||||
"album": show_metadata["name"],
|
||||
"description": episode_metadata["description"],
|
||||
"media_type": "Podcast",
|
||||
"publisher": show_metadata.get("publisher"),
|
||||
"rating": "Explicit" if episode_metadata.get("explicit") else "Unknown",
|
||||
"release_date": self.downloader.get_release_date_tag(
|
||||
@@ -58,9 +59,7 @@ class DownloaderEpisode:
|
||||
try:
|
||||
self._download(*args, **kwargs)
|
||||
finally:
|
||||
if self.downloader.temp_path.exists():
|
||||
logger.debug(f'Cleaning up "{self.downloader.temp_path}"')
|
||||
self.downloader.cleanup_temp_path()
|
||||
self.downloader.cleanup_temp_path()
|
||||
|
||||
def _download(
|
||||
self,
|
||||
@@ -68,7 +67,7 @@ class DownloaderEpisode:
|
||||
episode_metadata: dict = None,
|
||||
show_metadata: dict = None,
|
||||
gid_metadata: dict = None,
|
||||
stream_info: StreamInfo = None,
|
||||
stream_info: StreamInfoAudio = None,
|
||||
playlist_metadata: dict = None,
|
||||
playlist_track: int = None,
|
||||
decryption_key: bytes = None,
|
||||
@@ -86,14 +85,14 @@ class DownloaderEpisode:
|
||||
gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode")
|
||||
if not stream_info:
|
||||
logger.debug("Getting stream info")
|
||||
stream_info = self.downloader.get_stream_info(gid_metadata, "episode")
|
||||
stream_info = self.get_stream_info(gid_metadata, "episode")
|
||||
if not stream_info.file_id:
|
||||
logger.warning(
|
||||
"Episode is not available on Spotify's "
|
||||
"servers and no alternative found, skipping"
|
||||
)
|
||||
return
|
||||
if stream_info.quality != self.downloader.quality:
|
||||
if stream_info.quality != self.audio_quality:
|
||||
logger.warning(f"Quality has been changed to {stream_info.quality.value}")
|
||||
tags = self.get_tags(
|
||||
episode_metadata,
|
||||
@@ -107,49 +106,59 @@ class DownloaderEpisode:
|
||||
playlist_track,
|
||||
),
|
||||
}
|
||||
file_extension = self.get_file_extension()
|
||||
final_path = self.downloader.get_final_path(
|
||||
"episode",
|
||||
tags,
|
||||
".ogg",
|
||||
file_extension,
|
||||
)
|
||||
cover_path = self.get_cover_path(final_path)
|
||||
cover_url = self.downloader.get_cover_url(episode_metadata)
|
||||
decrypted_path = None
|
||||
remuxed_path = None
|
||||
if final_path.exists() and not self.downloader.overwrite:
|
||||
logger.warning(f'Track already exists at "{final_path}", skipping')
|
||||
else:
|
||||
if not decryption_key:
|
||||
logger.debug("Getting decryption key")
|
||||
decryption_key = self.downloader.get_decryption_key(stream_info.file_id)
|
||||
encrypted_path = self.downloader.get_encrypted_path(episode_id)
|
||||
decrypted_path = self.downloader.get_decrypted_path(episode_id)
|
||||
decryption_key = self.get_decryption_key(stream_info)
|
||||
encrypted_path = self.downloader.get_file_temp_path(
|
||||
episode_id,
|
||||
"_encrypted",
|
||||
file_extension,
|
||||
)
|
||||
decrypted_path = self.downloader.get_file_temp_path(
|
||||
episode_id,
|
||||
"_decrypted",
|
||||
file_extension,
|
||||
)
|
||||
remuxed_path = self.downloader.get_file_temp_path(
|
||||
episode_id,
|
||||
"_remuxed",
|
||||
file_extension,
|
||||
)
|
||||
logger.debug(f'Downloading to "{encrypted_path}"')
|
||||
self.downloader.download_stream_url(encrypted_path, stream_info.stream_url)
|
||||
logger.debug(f'Decrypting to "{decrypted_path}"')
|
||||
self.downloader.decrypt(
|
||||
self.download_stream_url(encrypted_path, stream_info.stream_url)
|
||||
logger.debug(
|
||||
f'Decrypting to "{decrypted_path}" and remuxing to "{remuxed_path}"'
|
||||
)
|
||||
self.decrypt(
|
||||
decryption_key,
|
||||
encrypted_path,
|
||||
decrypted_path,
|
||||
remuxed_path,
|
||||
)
|
||||
if (
|
||||
self.downloader.save_cover
|
||||
and cover_path.exists()
|
||||
and not self.downloader.overwrite
|
||||
):
|
||||
logger.debug(f'Cover already exists at "{cover_path}", skipping')
|
||||
elif self.downloader.save_cover and cover_url is not None:
|
||||
logger.debug(f'Saving cover to "{cover_path}"')
|
||||
self.downloader.save_cover_file(cover_path, cover_url)
|
||||
if decrypted_path:
|
||||
logger.debug("Applying tags")
|
||||
self.downloader.apply_tags(decrypted_path, tags, cover_url)
|
||||
logger.debug(f'Moving to "{final_path}"')
|
||||
self.downloader.move_to_final_path(decrypted_path, final_path)
|
||||
if self.downloader.save_playlist and playlist_metadata:
|
||||
playlist_file_path = self.downloader.get_playlist_file_path(tags)
|
||||
logger.debug(f'Updating M3U8 playlist from "{playlist_file_path}"')
|
||||
self.downloader.update_playlist_file(
|
||||
playlist_file_path,
|
||||
final_path,
|
||||
playlist_track,
|
||||
)
|
||||
media_temp_path = (
|
||||
remuxed_path
|
||||
if remuxed_path is not None and remuxed_path.exists()
|
||||
else decrypted_path
|
||||
)
|
||||
self.downloader._final_processing(
|
||||
cover_path,
|
||||
cover_url,
|
||||
media_temp_path,
|
||||
final_path,
|
||||
tags,
|
||||
playlist_metadata,
|
||||
playlist_track,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from .downloader_episode import DownloaderEpisode
|
||||
from .downloader_video import DownloaderVideo
|
||||
|
||||
logger = logging.getLogger("votify")
|
||||
|
||||
|
||||
class DownloaderEpisodeVideo(DownloaderVideo):
|
||||
def __init__(
|
||||
self,
|
||||
downloader_video: DownloaderVideo,
|
||||
downloader_episode: DownloaderEpisode,
|
||||
):
|
||||
self.__dict__.update(downloader_video.__dict__)
|
||||
self.downloader_episode = downloader_episode
|
||||
|
||||
def get_video_gid(self, gid_metadata: dict) -> str | None:
|
||||
if not gid_metadata.get("video"):
|
||||
return None
|
||||
return gid_metadata["video"][0]["file_id"]
|
||||
|
||||
def download(
|
||||
self,
|
||||
*args,
|
||||
**kwargs,
|
||||
):
|
||||
try:
|
||||
self._download(*args, **kwargs)
|
||||
finally:
|
||||
self.downloader.cleanup_temp_path()
|
||||
|
||||
def _download(
|
||||
self,
|
||||
episode_id: str,
|
||||
episode_metadata: dict = None,
|
||||
show_metadata: dict = None,
|
||||
gid_metadata: dict = None,
|
||||
playlist_metadata: dict = None,
|
||||
playlist_track: int = None,
|
||||
):
|
||||
if not episode_metadata:
|
||||
logger.debug("Getting episode metadata")
|
||||
episode_metadata = self.downloader.spotify_api.get_episode(episode_id)
|
||||
if not show_metadata:
|
||||
logger.debug("Getting show metadata")
|
||||
show_metadata = self.downloader.spotify_api.get_show(
|
||||
episode_metadata["show"]["id"]
|
||||
)
|
||||
if not gid_metadata:
|
||||
logger.debug("Getting GID metadata")
|
||||
gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode")
|
||||
video_gid = self.get_video_gid(gid_metadata)
|
||||
if not video_gid:
|
||||
logger.warning("Episode has no video, skipping")
|
||||
return
|
||||
stream_info = self.get_stream_info(video_gid)
|
||||
tags = self.downloader_episode.get_tags(
|
||||
episode_metadata,
|
||||
show_metadata,
|
||||
)
|
||||
file_extension = self.get_file_extension(
|
||||
stream_info.file_type_video,
|
||||
stream_info.file_type_audio,
|
||||
)
|
||||
final_path = self.downloader.get_final_path(
|
||||
"episode",
|
||||
tags,
|
||||
file_extension,
|
||||
)
|
||||
cover_path = self.get_cover_path(final_path)
|
||||
cover_url = self.downloader_episode.downloader.get_cover_url(show_metadata)
|
||||
remuxed_path = None
|
||||
if final_path.exists() and not self.downloader.overwrite:
|
||||
logger.warning(f'Episode already exists at "{final_path}", skipping')
|
||||
return
|
||||
else:
|
||||
temp_path_video = self.downloader.get_file_temp_path(
|
||||
episode_id,
|
||||
"_video",
|
||||
file_extension,
|
||||
)
|
||||
temp_path_audio = self.downloader.get_file_temp_path(
|
||||
episode_id,
|
||||
"_audio",
|
||||
file_extension,
|
||||
)
|
||||
remuxed_path = self.downloader.get_file_temp_path(
|
||||
episode_id,
|
||||
"_remuxed",
|
||||
file_extension,
|
||||
)
|
||||
logger.debug(f'Downloading video to "{temp_path_video}"')
|
||||
self.download_segments(stream_info.segment_urls_video, temp_path_video)
|
||||
logger.debug(f'Downloading audio to "{temp_path_audio}"')
|
||||
self.download_segments(stream_info.segment_urls_audio, temp_path_audio)
|
||||
logger.debug(f'Remuxing to "{remuxed_path}"')
|
||||
self.remux(
|
||||
temp_path_video,
|
||||
temp_path_audio,
|
||||
remuxed_path,
|
||||
)
|
||||
self.downloader._final_processing(
|
||||
cover_path,
|
||||
cover_url,
|
||||
remuxed_path,
|
||||
final_path,
|
||||
tags,
|
||||
playlist_metadata,
|
||||
playlist_track,
|
||||
)
|
||||
@@ -0,0 +1,251 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from InquirerPy import inquirer
|
||||
from InquirerPy.base.control import Choice
|
||||
|
||||
from .downloader_video import DownloaderVideo
|
||||
|
||||
logger = logging.getLogger("votify")
|
||||
|
||||
|
||||
class DownloaderMusicVideo(DownloaderVideo):
|
||||
def __init__(
|
||||
self,
|
||||
downloader_video: DownloaderVideo,
|
||||
):
|
||||
self.__dict__.update(downloader_video.__dict__)
|
||||
|
||||
def get_video_gid(self, gid_metadata: dict) -> str | None:
|
||||
if not gid_metadata.get("original_video"):
|
||||
return None
|
||||
return gid_metadata["original_video"][0]["gid"]
|
||||
|
||||
def get_tags(
|
||||
self,
|
||||
track_metadata: dict,
|
||||
album_metadata: dict,
|
||||
track_credits: dict,
|
||||
) -> dict:
|
||||
external_ids = track_metadata.get("external_ids")
|
||||
external_urls = (track_metadata.get("linked_from") or track_metadata)[
|
||||
"external_urls"
|
||||
]
|
||||
release_date_datetime_obj = self.downloader.get_release_date_datetime_obj(
|
||||
album_metadata["release_date"],
|
||||
album_metadata["release_date_precision"],
|
||||
)
|
||||
producers = next(
|
||||
role
|
||||
for role in track_credits["roleCredits"]
|
||||
if role["roleTitle"] == "Producers"
|
||||
)["artists"]
|
||||
composers = next(
|
||||
role
|
||||
for role in track_credits["roleCredits"]
|
||||
if role["roleTitle"] == "Writers"
|
||||
)["artists"]
|
||||
tags = {
|
||||
"artist": self.downloader.get_artist_string(track_metadata["artists"]),
|
||||
"composer": (
|
||||
self.downloader.get_artist_string(composers) if composers else None
|
||||
),
|
||||
"copyright": next(
|
||||
(i["text"] for i in album_metadata["copyrights"] if i["type"] == "P"),
|
||||
None,
|
||||
),
|
||||
"isrc": external_ids.get("isrc") if external_ids is not None else None,
|
||||
"label": album_metadata.get("label"),
|
||||
"media_type": "Music video",
|
||||
"producer": (
|
||||
self.downloader.get_artist_string(producers) if producers else None
|
||||
),
|
||||
"rating": "Explicit" if track_metadata.get("explicit") else "Unknown",
|
||||
"title": track_metadata["name"],
|
||||
"release_date": self.downloader.get_release_date_tag(
|
||||
release_date_datetime_obj
|
||||
),
|
||||
"release_year": str(release_date_datetime_obj.year),
|
||||
"url": external_urls["spotify"],
|
||||
}
|
||||
tags["release_year"] = str(release_date_datetime_obj.year)
|
||||
return tags
|
||||
|
||||
def get_music_video_id_from_song_id(
|
||||
self,
|
||||
track_id: str,
|
||||
artist_id: str,
|
||||
) -> dict | None:
|
||||
now_playing_view = self.downloader.spotify_api.get_now_playing_view(
|
||||
track_id, artist_id
|
||||
)
|
||||
related_music_videos = now_playing_view["data"]["trackUnion"]["relatedVideos"][
|
||||
"items"
|
||||
]
|
||||
if not related_music_videos:
|
||||
return
|
||||
choices = [
|
||||
Choice(
|
||||
name="None",
|
||||
value=None,
|
||||
)
|
||||
]
|
||||
choices.extend(
|
||||
[
|
||||
Choice(
|
||||
name=" | ".join(
|
||||
[
|
||||
self.downloader.get_artist_string(
|
||||
[
|
||||
{"name": artist["profile"]["name"]}
|
||||
for artist in related_music_video["trackOfVideo"][
|
||||
"data"
|
||||
]["artists"]["items"]
|
||||
]
|
||||
),
|
||||
related_music_video["trackOfVideo"]["data"]["name"],
|
||||
]
|
||||
),
|
||||
value=related_music_video["trackOfVideo"]["data"]["uri"].split(":")[
|
||||
-1
|
||||
],
|
||||
)
|
||||
for related_music_video in related_music_videos
|
||||
]
|
||||
)
|
||||
selected_music_video_id = inquirer.select(
|
||||
message="Select which music video to download: (Artist | Title)",
|
||||
choices=choices,
|
||||
).execute()
|
||||
return selected_music_video_id
|
||||
|
||||
def download(
|
||||
self,
|
||||
*args,
|
||||
**kwargs,
|
||||
):
|
||||
try:
|
||||
self._download(*args, **kwargs)
|
||||
finally:
|
||||
self.downloader.cleanup_temp_path()
|
||||
|
||||
def _download(
|
||||
self,
|
||||
music_video_id: str,
|
||||
music_video_metadata: dict = None,
|
||||
album_metadata: dict = None,
|
||||
gid_metadata: dict = None,
|
||||
playlist_metadata: dict = None,
|
||||
playlist_track: int = None,
|
||||
):
|
||||
if not music_video_metadata:
|
||||
logger.debug("Getting music video metadata")
|
||||
music_video_metadata = self.downloader.spotify_api.get_track(music_video_id)
|
||||
if not album_metadata:
|
||||
logger.debug("Getting album metadata")
|
||||
album_metadata = self.downloader.spotify_api.get_album(
|
||||
music_video_metadata["album"]["id"]
|
||||
)
|
||||
if not gid_metadata:
|
||||
logger.debug("Getting GID metadata")
|
||||
gid_metadata = self.downloader.get_gid_metadata(music_video_id, "track")
|
||||
video_gid = self.get_video_gid(gid_metadata)
|
||||
if not video_gid:
|
||||
logger.debug("Getting equivalent music video ID from song ID")
|
||||
music_video_id = self.get_music_video_id_from_song_id(
|
||||
music_video_id,
|
||||
album_metadata["artists"][0]["id"],
|
||||
)
|
||||
if not music_video_id:
|
||||
logger.warning(
|
||||
"No related music videos found or no music video selected, skipping"
|
||||
)
|
||||
return
|
||||
music_video_metadata = self.downloader.spotify_api.get_track(music_video_id)
|
||||
logger.warning(
|
||||
f'Switching to downloading music video "{music_video_metadata["name"]}"'
|
||||
)
|
||||
album_metadata = self.downloader.spotify_api.get_album(
|
||||
music_video_metadata["album"]["id"]
|
||||
)
|
||||
gid_metadata = self.downloader.get_gid_metadata(music_video_id, "track")
|
||||
video_gid = self.get_video_gid(gid_metadata)
|
||||
stream_info = self.get_stream_info(video_gid)
|
||||
logger.debug("Getting credits")
|
||||
track_credits = self.downloader.spotify_api.get_track_credits(music_video_id)
|
||||
tags = self.get_tags(
|
||||
music_video_metadata,
|
||||
album_metadata,
|
||||
track_credits,
|
||||
)
|
||||
file_extension = self.get_file_extension(
|
||||
stream_info.file_type_video,
|
||||
stream_info.file_type_audio,
|
||||
)
|
||||
final_path = self.downloader.get_final_path(
|
||||
"music-video",
|
||||
tags,
|
||||
file_extension,
|
||||
)
|
||||
cover_path = self.get_cover_path(final_path)
|
||||
cover_url = self.downloader.get_cover_url(album_metadata)
|
||||
remuxed_path = None
|
||||
if final_path.exists() and not self.downloader.overwrite:
|
||||
logger.warning(f'Music video already exists at "{final_path}", skipping')
|
||||
return
|
||||
else:
|
||||
key_id, decryption_key = self.downloader.get_widevine_decryption_key(
|
||||
stream_info.encryption_data_widevine,
|
||||
"video",
|
||||
)
|
||||
encrypted_path_video = self.downloader.get_file_temp_path(
|
||||
music_video_id,
|
||||
"_video_encrypted",
|
||||
file_extension,
|
||||
)
|
||||
encrypted_path_audio = self.downloader.get_file_temp_path(
|
||||
music_video_id,
|
||||
"_audio_encrypted",
|
||||
file_extension,
|
||||
)
|
||||
decrypted_path_video = self.downloader.get_file_temp_path(
|
||||
music_video_id,
|
||||
"_video_decrypted",
|
||||
file_extension,
|
||||
)
|
||||
decrypted_path_audio = self.downloader.get_file_temp_path(
|
||||
music_video_id,
|
||||
"_audio_decrypted",
|
||||
file_extension,
|
||||
)
|
||||
remuxed_path = self.downloader.get_file_temp_path(
|
||||
music_video_id,
|
||||
"_remuxed",
|
||||
file_extension,
|
||||
)
|
||||
logger.debug(f'Downloading video to "{encrypted_path_video}"')
|
||||
self.download_segments(stream_info.segment_urls_video, encrypted_path_video)
|
||||
logger.debug(f'Downloading audio to "{encrypted_path_audio}"')
|
||||
self.download_segments(stream_info.segment_urls_audio, encrypted_path_audio)
|
||||
logger.debug(
|
||||
f'Decryping video/audio to "{decrypted_path_video}/{decrypted_path_audio}" and remuxing to "{remuxed_path}"'
|
||||
)
|
||||
self.remux(
|
||||
decrypted_path_video,
|
||||
decrypted_path_audio,
|
||||
remuxed_path,
|
||||
key_id,
|
||||
decryption_key,
|
||||
encrypted_path_video,
|
||||
encrypted_path_audio,
|
||||
)
|
||||
self.downloader._final_processing(
|
||||
cover_path,
|
||||
cover_url,
|
||||
remuxed_path,
|
||||
final_path,
|
||||
tags,
|
||||
playlist_metadata,
|
||||
playlist_track,
|
||||
)
|
||||
+49
-41
@@ -4,20 +4,20 @@ import datetime
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from .downloader import Downloader
|
||||
from .models import Lyrics, StreamInfo
|
||||
from .downloader_audio import DownloaderAudio
|
||||
from .models import Lyrics, StreamInfoAudio
|
||||
|
||||
logger = logging.getLogger("votify")
|
||||
|
||||
|
||||
class DownloaderSong:
|
||||
class DownloaderSong(DownloaderAudio):
|
||||
def __init__(
|
||||
self,
|
||||
downloader: Downloader,
|
||||
downloader_audio: DownloaderAudio,
|
||||
lrc_only: bool = False,
|
||||
no_lrc: bool = False,
|
||||
):
|
||||
self.downloader = downloader
|
||||
self.__dict__.update(downloader_audio.__dict__)
|
||||
self.lrc_only = lrc_only
|
||||
self.no_lrc = no_lrc
|
||||
|
||||
@@ -75,6 +75,7 @@ class DownloaderSong:
|
||||
"isrc": external_ids.get("isrc") if external_ids is not None else None,
|
||||
"label": album_metadata.get("label"),
|
||||
"lyrics": lyrics_unsynced,
|
||||
"media_type": "Song",
|
||||
"producer": (
|
||||
self.downloader.get_artist_string(producers) if producers else None
|
||||
),
|
||||
@@ -135,9 +136,7 @@ class DownloaderSong:
|
||||
try:
|
||||
self._download(*args, **kwargs)
|
||||
finally:
|
||||
if self.downloader.temp_path.exists():
|
||||
logger.debug(f'Cleaning up "{self.downloader.temp_path}"')
|
||||
self.downloader.cleanup_temp_path()
|
||||
self.downloader.cleanup_temp_path()
|
||||
|
||||
def _download(
|
||||
self,
|
||||
@@ -145,7 +144,7 @@ class DownloaderSong:
|
||||
track_metadata: dict = None,
|
||||
album_metadata: dict = None,
|
||||
gid_metadata: dict = None,
|
||||
stream_info: StreamInfo = None,
|
||||
stream_info: StreamInfoAudio = None,
|
||||
playlist_metadata: dict = None,
|
||||
playlist_track: int = None,
|
||||
decryption_key: bytes = None,
|
||||
@@ -163,14 +162,14 @@ class DownloaderSong:
|
||||
gid_metadata = self.downloader.get_gid_metadata(track_id, "track")
|
||||
if not stream_info:
|
||||
logger.debug("Getting stream info")
|
||||
stream_info = self.downloader.get_stream_info(gid_metadata, "track")
|
||||
stream_info = self.get_stream_info(gid_metadata, "track")
|
||||
if not stream_info.file_id:
|
||||
logger.warning(
|
||||
"Track is not available on Spotify's "
|
||||
"servers and no alternative found, skipping"
|
||||
)
|
||||
return
|
||||
if stream_info.quality != self.downloader.quality:
|
||||
if stream_info.quality != self.audio_quality:
|
||||
logger.warning(f"Quality has been changed to {stream_info.quality.value}")
|
||||
if gid_metadata.get("has_lyrics"):
|
||||
logger.debug("Getting lyrics")
|
||||
@@ -193,15 +192,17 @@ class DownloaderSong:
|
||||
playlist_track,
|
||||
),
|
||||
}
|
||||
file_extension = self.get_file_extension()
|
||||
final_path = self.downloader.get_final_path(
|
||||
"track",
|
||||
tags,
|
||||
".ogg",
|
||||
file_extension,
|
||||
)
|
||||
lrc_path = self.downloader.get_lrc_path(final_path)
|
||||
cover_path = self.get_cover_path(final_path)
|
||||
cover_url = self.downloader.get_cover_url(album_metadata)
|
||||
decrypted_path = None
|
||||
remuxed_path = None
|
||||
if self.lrc_only:
|
||||
pass
|
||||
elif final_path.exists() and not self.downloader.overwrite:
|
||||
@@ -209,16 +210,32 @@ class DownloaderSong:
|
||||
else:
|
||||
if not decryption_key:
|
||||
logger.debug("Getting decryption key")
|
||||
decryption_key = self.downloader.get_decryption_key(stream_info.file_id)
|
||||
encrypted_path = self.downloader.get_encrypted_path(track_id)
|
||||
decrypted_path = self.downloader.get_decrypted_path(track_id)
|
||||
decryption_key = self.get_decryption_key(stream_info)
|
||||
encrypted_path = self.downloader.get_file_temp_path(
|
||||
track_id,
|
||||
"_encrypted",
|
||||
file_extension,
|
||||
)
|
||||
decrypted_path = self.downloader.get_file_temp_path(
|
||||
track_id,
|
||||
"_decrypted",
|
||||
file_extension,
|
||||
)
|
||||
remuxed_path = self.downloader.get_file_temp_path(
|
||||
track_id,
|
||||
"_remuxed",
|
||||
file_extension,
|
||||
)
|
||||
logger.debug(f'Downloading to "{encrypted_path}"')
|
||||
self.downloader.download_stream_url(encrypted_path, stream_info.stream_url)
|
||||
logger.debug(f'Decrypting to "{decrypted_path}"')
|
||||
self.downloader.decrypt(
|
||||
self.download_stream_url(encrypted_path, stream_info.stream_url)
|
||||
logger.debug(
|
||||
f'Decrypting to "{decrypted_path}" and remuxing to "{remuxed_path}"'
|
||||
)
|
||||
self.decrypt(
|
||||
decryption_key,
|
||||
encrypted_path,
|
||||
decrypted_path,
|
||||
remuxed_path,
|
||||
)
|
||||
if self.no_lrc or not lyrics.synced:
|
||||
pass
|
||||
@@ -227,26 +244,17 @@ class DownloaderSong:
|
||||
else:
|
||||
logger.debug(f'Saving synced lyrics to "{lrc_path}"')
|
||||
self.downloader.save_lrc(lrc_path, lyrics.synced)
|
||||
if (
|
||||
self.downloader.save_cover
|
||||
and cover_path.exists()
|
||||
and not self.downloader.overwrite
|
||||
):
|
||||
logger.debug(f'Cover already exists at "{cover_path}", skipping')
|
||||
return
|
||||
elif self.downloader.save_cover and cover_url is not None:
|
||||
logger.debug(f'Saving cover to "{cover_path}"')
|
||||
self.downloader.save_cover_file(cover_path, cover_url)
|
||||
if decrypted_path:
|
||||
logger.debug("Applying tags")
|
||||
self.downloader.apply_tags(decrypted_path, tags, cover_url)
|
||||
logger.debug(f'Moving to "{final_path}"')
|
||||
self.downloader.move_to_final_path(decrypted_path, final_path)
|
||||
if not self.lrc_only and self.downloader.save_playlist and playlist_metadata:
|
||||
playlist_file_path = self.downloader.get_playlist_file_path(tags)
|
||||
logger.debug(f'Updating M3U8 playlist from "{playlist_file_path}"')
|
||||
self.downloader.update_playlist_file(
|
||||
playlist_file_path,
|
||||
final_path,
|
||||
playlist_track,
|
||||
)
|
||||
media_temp_path = (
|
||||
remuxed_path
|
||||
if remuxed_path is not None and remuxed_path.exists()
|
||||
else decrypted_path
|
||||
)
|
||||
self.downloader._final_processing(
|
||||
cover_path,
|
||||
cover_url,
|
||||
media_temp_path,
|
||||
final_path,
|
||||
tags,
|
||||
playlist_metadata,
|
||||
playlist_track,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,400 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from InquirerPy import inquirer
|
||||
from InquirerPy.base.control import Choice
|
||||
from yt_dlp.downloader.fragment import FragmentFD
|
||||
from yt_dlp.YoutubeDL import YoutubeDL
|
||||
|
||||
from .downloader import Downloader
|
||||
from .enums import RemuxModeAudio, RemuxModeVideo, VideoFormat
|
||||
from .models import StreamInfoVideo
|
||||
|
||||
|
||||
class DownloaderVideo:
|
||||
def __init__(
|
||||
self,
|
||||
downloader: Downloader,
|
||||
video_format: VideoFormat = VideoFormat.MP4,
|
||||
remux_mode: RemuxModeVideo = RemuxModeVideo.FFMPEG,
|
||||
):
|
||||
self.downloader = downloader
|
||||
self.video_format = video_format
|
||||
self.remux_mode = remux_mode
|
||||
self._adjust_remux_mode()
|
||||
|
||||
def _adjust_remux_mode(self):
|
||||
if self.video_format == VideoFormat.WEBM:
|
||||
self.remux_mode = RemuxModeVideo.FFMPEG
|
||||
|
||||
def get_cover_path(self, final_path: Path) -> Path:
|
||||
return final_path.with_suffix(".jpg")
|
||||
|
||||
def get_file_extension(
|
||||
self,
|
||||
file_type_video: str,
|
||||
file_type_audio: str,
|
||||
) -> str:
|
||||
if file_type_video == file_type_audio:
|
||||
return "." + file_type_video
|
||||
else:
|
||||
return ".mp4"
|
||||
|
||||
def get_encryption_info(
|
||||
self,
|
||||
encryption_infos: list[dict],
|
||||
key_system: str,
|
||||
) -> tuple[int, str]:
|
||||
for index, encryption_info in enumerate(encryption_infos):
|
||||
if encryption_info["key_system"] == key_system:
|
||||
return index, encryption_info
|
||||
|
||||
def get_stream_info(
|
||||
self,
|
||||
gid: str,
|
||||
) -> StreamInfoVideo:
|
||||
stream_info = StreamInfoVideo()
|
||||
manifest = self.downloader.spotify_api.get_video_manifest(gid)
|
||||
base_url = manifest["base_urls"][0]
|
||||
initialization_template = manifest["initialization_template"]
|
||||
segment_template = manifest["segment_template"]
|
||||
end_time_millis = manifest["end_time_millis"]
|
||||
segment_length = manifest["contents"][0]["segment_length"]
|
||||
encryption_index = None
|
||||
if manifest["contents"][0].get("encryption_infos"):
|
||||
encryption_index, encryption_info = self.get_encryption_info(
|
||||
manifest["contents"][0]["encryption_infos"],
|
||||
"widevine",
|
||||
)
|
||||
stream_info.encryption_data_widevine = encryption_info["encryption_data"]
|
||||
video_profiles = list(
|
||||
profile
|
||||
for profile in manifest["contents"][0]["profiles"]
|
||||
if profile["mime_type"].startswith("video")
|
||||
and encryption_index
|
||||
in (profile.get("encryption_indices", [encryption_index]))
|
||||
)
|
||||
video_profiles = list(
|
||||
filter(
|
||||
lambda x: x["mime_type"].startswith("video")
|
||||
and encryption_index
|
||||
in (x.get("encryption_indices", [encryption_index])),
|
||||
manifest["contents"][0]["profiles"],
|
||||
)
|
||||
)
|
||||
audio_profiles = list(
|
||||
filter(
|
||||
lambda x: x["mime_type"].startswith("audio")
|
||||
and encryption_index
|
||||
in (x.get("encryption_indices", [encryption_index])),
|
||||
manifest["contents"][0]["profiles"],
|
||||
)
|
||||
)
|
||||
if not video_profiles or not audio_profiles:
|
||||
return stream_info
|
||||
if self.video_format == VideoFormat.ASK:
|
||||
profile_video, profile_audio = (
|
||||
self.get_video_profile_from_user(
|
||||
video_profiles,
|
||||
),
|
||||
self.get_audio_profile_from_user(
|
||||
audio_profiles,
|
||||
),
|
||||
)
|
||||
else:
|
||||
profile_video, profile_audio = (
|
||||
self.get_best_profile_by_bitrate(
|
||||
video_profiles,
|
||||
f"video/{self.video_format.value}",
|
||||
),
|
||||
self.get_best_profile_by_bitrate(
|
||||
audio_profiles,
|
||||
f"audio/{self.video_format.value}",
|
||||
),
|
||||
)
|
||||
profile_id_video = profile_video["id"]
|
||||
file_type_video = profile_video["file_type"]
|
||||
profile_id_audio = profile_audio["id"]
|
||||
file_type_audio = profile_audio["file_type"]
|
||||
stream_info.segment_urls_video = self.get_segment_urls(
|
||||
base_url,
|
||||
initialization_template,
|
||||
segment_template,
|
||||
end_time_millis,
|
||||
segment_length,
|
||||
profile_id_video,
|
||||
file_type_video,
|
||||
)
|
||||
stream_info.segment_urls_audio = self.get_segment_urls(
|
||||
base_url,
|
||||
initialization_template,
|
||||
segment_template,
|
||||
end_time_millis,
|
||||
segment_length,
|
||||
profile_id_audio,
|
||||
file_type_audio,
|
||||
)
|
||||
stream_info.file_type_video = file_type_video
|
||||
stream_info.file_type_audio = file_type_audio
|
||||
return stream_info
|
||||
|
||||
def get_best_profile_by_bitrate(
|
||||
self,
|
||||
profiles: list[dict],
|
||||
mime_type: str,
|
||||
) -> str:
|
||||
profiles_filtered = list(
|
||||
filter(
|
||||
lambda x: x["mime_type"] == mime_type,
|
||||
profiles,
|
||||
)
|
||||
)
|
||||
bitrate_key = f"{mime_type.split('/')[0]}_bitrate"
|
||||
best_profile = max(profiles_filtered, key=lambda x: x[bitrate_key])
|
||||
return best_profile
|
||||
|
||||
def get_video_profile_from_user(
|
||||
self,
|
||||
video_profiles: list[dict],
|
||||
) -> str:
|
||||
choices = [
|
||||
Choice(
|
||||
name=" | ".join(
|
||||
[
|
||||
profile["video_codec"],
|
||||
f"{profile['video_width']}x{profile['video_height']}",
|
||||
str(profile["video_bitrate"]),
|
||||
]
|
||||
),
|
||||
value=profile,
|
||||
)
|
||||
for profile in video_profiles
|
||||
]
|
||||
selected = inquirer.select(
|
||||
message="Select which video codec to download: (Codec | Resolution | Bitrate)",
|
||||
choices=choices,
|
||||
).execute()
|
||||
return selected
|
||||
|
||||
def get_audio_profile_from_user(
|
||||
self,
|
||||
audio_profiles: list[dict],
|
||||
) -> str:
|
||||
choices = [
|
||||
Choice(
|
||||
name=" | ".join(
|
||||
[
|
||||
profile["audio_codec"],
|
||||
str(profile["audio_bitrate"]),
|
||||
]
|
||||
),
|
||||
value=profile,
|
||||
)
|
||||
for profile in audio_profiles
|
||||
]
|
||||
selected = inquirer.select(
|
||||
message="Select which audio codec to download: (Codec | Bitrate)",
|
||||
choices=choices,
|
||||
).execute()
|
||||
return selected
|
||||
|
||||
def get_segment_urls(
|
||||
self,
|
||||
base_url: str,
|
||||
initialization_template: str,
|
||||
segment_template: str,
|
||||
end_time_millis: int,
|
||||
segment_length: int,
|
||||
profile_id: int,
|
||||
file_type: str,
|
||||
) -> list[str]:
|
||||
initialization_template_formatted = initialization_template.replace(
|
||||
"{{profile_id}}", str(profile_id)
|
||||
).replace("{{file_type}}", file_type)
|
||||
segment_urls = []
|
||||
first_segment = base_url + initialization_template_formatted
|
||||
segment_urls.append(first_segment)
|
||||
for i in range(0, int(end_time_millis / 1000) + 5, segment_length):
|
||||
segment_template_url_formatted = (
|
||||
segment_template.replace("{{profile_id}}", str(profile_id))
|
||||
.replace("{{segment_timestamp}}", str(i))
|
||||
.replace("{{file_type}}", file_type)
|
||||
)
|
||||
segment_urls.append(base_url + segment_template_url_formatted)
|
||||
return segment_urls
|
||||
|
||||
def download_segments(
|
||||
self,
|
||||
segment_urls: list[str],
|
||||
input_path: Path,
|
||||
):
|
||||
input_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
segments_dict = [
|
||||
{"url": url, "frag_index": idx + 1} for idx, url in enumerate(segment_urls)
|
||||
]
|
||||
ctx = {
|
||||
"filename": str(input_path),
|
||||
"total_frags": len(segments_dict),
|
||||
}
|
||||
info_dict = {}
|
||||
with YoutubeDL(
|
||||
{
|
||||
"quiet": True,
|
||||
"no_warnings": True,
|
||||
"noprogress": self.downloader.silence,
|
||||
},
|
||||
) as ydl:
|
||||
try:
|
||||
fragment_downloader = FragmentFD(ydl, ydl.params)
|
||||
fragment_downloader._prepare_and_start_frag_download(
|
||||
ctx,
|
||||
info_dict,
|
||||
)
|
||||
fragment_downloader.download_and_append_fragments(
|
||||
ctx,
|
||||
segments_dict,
|
||||
info_dict,
|
||||
)
|
||||
finally:
|
||||
fragment_downloader._finish_multiline_status()
|
||||
|
||||
def remux(
|
||||
self,
|
||||
decrypted_path_video: Path,
|
||||
decrypted_path_audio: Path,
|
||||
remuxed_path: Path,
|
||||
key_id: str | None = None,
|
||||
decryption_key: str | None = None,
|
||||
encrypted_path_video: Path | None = None,
|
||||
encrypted_path_audio: Path | None = None,
|
||||
):
|
||||
if decryption_key:
|
||||
if remuxed_path.suffix == ".webm":
|
||||
self.decrypt_packager(
|
||||
key_id,
|
||||
decryption_key,
|
||||
encrypted_path_video,
|
||||
decrypted_path_video,
|
||||
)
|
||||
self.decrypt_packager(
|
||||
key_id,
|
||||
decryption_key,
|
||||
encrypted_path_audio,
|
||||
decrypted_path_audio,
|
||||
)
|
||||
else:
|
||||
self.decrypt_mp4decrypt(
|
||||
decryption_key,
|
||||
encrypted_path_video,
|
||||
decrypted_path_video,
|
||||
)
|
||||
self.decrypt_mp4decrypt(
|
||||
decryption_key,
|
||||
encrypted_path_audio,
|
||||
decrypted_path_audio,
|
||||
)
|
||||
if self.remux_mode == RemuxModeAudio.MP4BOX:
|
||||
self.remux_mp4box(
|
||||
decrypted_path_video,
|
||||
decrypted_path_audio,
|
||||
remuxed_path,
|
||||
)
|
||||
else:
|
||||
self.remux_ffmpeg(
|
||||
decrypted_path_video,
|
||||
decrypted_path_audio,
|
||||
remuxed_path,
|
||||
)
|
||||
|
||||
def decrypt_packager(
|
||||
self,
|
||||
key_id: str,
|
||||
decryption_key: str,
|
||||
encrypted_path: Path,
|
||||
decrypted_path: Path,
|
||||
):
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.packager_path_full,
|
||||
"--quiet",
|
||||
f"stream=0,in={encrypted_path},output={decrypted_path}",
|
||||
"-enable_raw_key_decryption",
|
||||
"-keys",
|
||||
f"key_id={key_id}:key={decryption_key}",
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
|
||||
def decrypt_mp4decrypt(
|
||||
self,
|
||||
decryption_key: str,
|
||||
encrypted_path: Path,
|
||||
decrypted_path: Path,
|
||||
):
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.mp4decrypt_path_full,
|
||||
"--key",
|
||||
f"1:{decryption_key}",
|
||||
encrypted_path,
|
||||
decrypted_path,
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
|
||||
def remux_ffmpeg(
|
||||
self,
|
||||
input_path_video: Path,
|
||||
input_path_audio: Path,
|
||||
output_path: Path,
|
||||
):
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.ffmpeg_path_full,
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-y",
|
||||
"-i",
|
||||
input_path_video,
|
||||
"-i",
|
||||
input_path_audio,
|
||||
"-c",
|
||||
"copy",
|
||||
"-map",
|
||||
"0:v:0",
|
||||
"-map",
|
||||
"1:a:0",
|
||||
output_path,
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
|
||||
def remux_mp4box(
|
||||
self,
|
||||
decrypted_path_video: Path,
|
||||
decrypted_path_audio: Path,
|
||||
remuxed_path: Path,
|
||||
):
|
||||
subprocess.run(
|
||||
[
|
||||
self.downloader.mp4box_path_full,
|
||||
"-quiet",
|
||||
"-itags",
|
||||
"artist=placeholder",
|
||||
"-keep-utc",
|
||||
"-add",
|
||||
decrypted_path_video,
|
||||
"-add",
|
||||
decrypted_path_audio,
|
||||
"-new",
|
||||
remuxed_path,
|
||||
],
|
||||
check=True,
|
||||
**self.downloader.subprocess_additional_args,
|
||||
)
|
||||
+23
-4
@@ -1,12 +1,31 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class Quality(Enum):
|
||||
HIGH = "high"
|
||||
MEDIUM = "medium"
|
||||
LOW = "low"
|
||||
class AudioQuality(Enum):
|
||||
VORBIS_HIGH = "vorbis-high"
|
||||
VORBIS_MEDIUM = "vorbis-medium"
|
||||
VORBIS_LOW = "vorbis-low"
|
||||
AAC_MEDIUM = "aac-medium"
|
||||
AAC_HIGH = "aac-high"
|
||||
|
||||
|
||||
class VideoFormat(Enum):
|
||||
MP4 = "mp4"
|
||||
WEBM = "webm"
|
||||
ASK = "ask"
|
||||
|
||||
|
||||
class DownloadMode(Enum):
|
||||
YTDLP = "ytdlp"
|
||||
ARIA2C = "aria2c"
|
||||
|
||||
|
||||
class RemuxModeAudio(Enum):
|
||||
FFMPEG = "ffmpeg"
|
||||
MP4BOX = "mp4box"
|
||||
MP4DECRYPT = "mp4decrypt"
|
||||
|
||||
|
||||
class RemuxModeVideo(Enum):
|
||||
FFMPEG = "ffmpeg"
|
||||
MP4BOX = "mp4box"
|
||||
|
||||
+13
-3
@@ -2,7 +2,7 @@ from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .enums import Quality
|
||||
from .enums import AudioQuality
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -26,7 +26,17 @@ class DownloadQueueItem:
|
||||
|
||||
|
||||
@dataclass
|
||||
class StreamInfo:
|
||||
class StreamInfoAudio:
|
||||
stream_url: str = None
|
||||
file_id: str = None
|
||||
quality: Quality = None
|
||||
widevine_pssh: str = None
|
||||
quality: AudioQuality = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class StreamInfoVideo:
|
||||
segment_urls_video: list[str] = None
|
||||
segment_urls_audio: list[str] = None
|
||||
file_type_video: str = None
|
||||
file_type_audio: str = None
|
||||
encryption_data_widevine: str = None
|
||||
|
||||
@@ -20,9 +20,15 @@ class SpotifyApi:
|
||||
LYRICS_API_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/{track_id}"
|
||||
METADATA_API_URL = "https://api.spotify.com/v1/{type}/{item_id}"
|
||||
GID_METADATA_API_URL = "https://spclient.wg.spotify.com/metadata/4/{media_type}/{gid}?market=from_token"
|
||||
PATHFINDER_API_URL = "https://api-partner.spotify.com/pathfinder/v1/query"
|
||||
VIDEO_MANIFEST_API_URL = "https://gue1-spclient.spotify.com/manifests/v7/json/sources/{gid}/options/supports_drm"
|
||||
PLAYPLAY_LICENSE_API_URL = (
|
||||
"https://gew4-spclient.spotify.com/playplay/v1/key/{file_id}"
|
||||
)
|
||||
WIDEVINE_LICENSE_API_URL = (
|
||||
"https://gue1-spclient.spotify.com/widevine-license/v1/{type}/license"
|
||||
)
|
||||
SEEK_TABLE_API_URL = "https://seektables.scdn.co/seektable/{file_id}.json"
|
||||
TRACK_CREDITS_API_URL = "https://spclient.wg.spotify.com/track-credits-view/v0/experimental/{track_id}/credits"
|
||||
STREAM_URLS_API_URL = (
|
||||
"https://gue1-spclient.spotify.com/storage-resolve/v2/files/audio/interactive/11/"
|
||||
@@ -253,6 +259,21 @@ class SpotifyApi:
|
||||
)
|
||||
return artist_albums
|
||||
|
||||
def get_video_manifest(
|
||||
self,
|
||||
gid: str,
|
||||
) -> dict:
|
||||
self._refresh_session_auth()
|
||||
response = self.session.get(self.VIDEO_MANIFEST_API_URL.format(gid=gid))
|
||||
check_response(response)
|
||||
return response.json()
|
||||
|
||||
def get_seek_table(self, file_id: str) -> dict:
|
||||
self._refresh_session_auth()
|
||||
response = self.session.get(self.SEEK_TABLE_API_URL.format(file_id=file_id))
|
||||
check_response(response)
|
||||
return response.json()
|
||||
|
||||
def get_playplay_license(self, file_id: str, challenge: bytes) -> bytes:
|
||||
self._refresh_session_auth()
|
||||
response = self.session.post(
|
||||
@@ -262,8 +283,44 @@ class SpotifyApi:
|
||||
check_response(response)
|
||||
return response.content
|
||||
|
||||
def get_widevine_license(self, challenge: bytes, media_type: str) -> bytes:
|
||||
self._refresh_session_auth()
|
||||
response = self.session.post(
|
||||
self.WIDEVINE_LICENSE_API_URL.format(type=media_type),
|
||||
challenge,
|
||||
)
|
||||
check_response(response)
|
||||
return response.content
|
||||
|
||||
def get_stream_urls(self, file_id: str) -> str:
|
||||
self._refresh_session_auth()
|
||||
response = self.session.get(self.STREAM_URLS_API_URL.format(file_id=file_id))
|
||||
check_response(response)
|
||||
return response.json()
|
||||
|
||||
def get_now_playing_view(self, track_id: str, artist_id: str) -> dict:
|
||||
self._refresh_session_auth()
|
||||
response = self.session.get(
|
||||
self.PATHFINDER_API_URL,
|
||||
params={
|
||||
"operationName": "queryNpvArtist",
|
||||
"variables": json.dumps(
|
||||
{
|
||||
"artistUri": f"spotify:artist:{artist_id}",
|
||||
"trackUri": f"spotify:track:{track_id}",
|
||||
"enableCredits": True,
|
||||
"enableRelatedVideos": True,
|
||||
}
|
||||
),
|
||||
"extensions": json.dumps(
|
||||
{
|
||||
"persistedQuery": {
|
||||
"version": 1,
|
||||
"sha256Hash": "4ec4ae302c609a517cab6b8868f601cd3457c751c570ab12e988723cc036284f",
|
||||
}
|
||||
}
|
||||
),
|
||||
},
|
||||
)
|
||||
check_response(response)
|
||||
return response.json()
|
||||
|
||||
Reference in New Issue
Block a user