Skip to content

Commit

Permalink
feat: support audio_info format
Browse files Browse the repository at this point in the history
  • Loading branch information
WorldObservationLog committed May 29, 2024
1 parent 9081672 commit 0d98109
Show file tree
Hide file tree
Showing 6 changed files with 62 additions and 22 deletions.
16 changes: 14 additions & 2 deletions config.example.toml
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,22 @@ codecPriority = ["alac", "ec3", "ac3", "aac"]
# Encapsulate Atmos(ec-3/ac-3) as M4A and write the song metadata
atmosConventToM4a = true
# Follow the Python Format format (https://docs.python.org/3/library/string.html#formatstrings)
# Write the audio information to the songNameFormat and playlistSongNameFormat
# Only support alac codec
# Available values: bit_depth, sample_rate, sample_rate_kHz, codec
# This feature may slow down the speed of checking for existing songs
# For example:
# audioInfoFormat = " [{codec}][{bit_depth}bit][{sample_rate_kHz}kHz]"
# songNameFormat = "{disk}-{tracknum:02d} {title}{audio_info}"
# When transcribing audio with alac codec, the transcribed file name is:
# 1-01 名もなき何もかも [ALAC][16bit][44.1kHz]
# When transcribing audio with other codecs, the transcribed file name is:
# 1-01 名もなき何もかも
audioInfoFormat = ""
# Available values:
# title, artist, album, album_artist, composer,
# genre, created, track, tracknum, disk,
# record_company, upc, isrc, copyright, codec
# record_company, upc, isrc, copyright, codec, audio_info
songNameFormat = "{disk}-{tracknum:02d} {title}"
# Ditto
dirPathFormat = "downloads/{album_artist}/{album}"
Expand All @@ -60,7 +72,7 @@ playlistDirPathFormat = "downloads/playlists/{playlistName}"
# Available values:
# title, artist, album, album_artist, composer,
# genre, created, track, tracknum, disk,
# record_company, upc, isrc, copyright,
# record_company, upc, isrc, copyright, audio_info
# playlistName, playlistCuratorName, playlistSongIndex, codec
playlistSongNameFormat = "{playlistSongIndex:02d}. {artist} - {title}"
# Save lyrics as .lrc file
Expand Down
1 change: 1 addition & 0 deletions src/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ class Download(BaseModel):
codecAlternative: bool
codecPriority: list[str]
atmosConventToM4a: bool
audioInfoFormat: str
songNameFormat: str
dirPathFormat: str
playlistDirPathFormat: str
Expand Down
14 changes: 11 additions & 3 deletions src/metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
from src.models.song_data import Datum
from src.utils import ttml_convent_to_lrc

NOT_INCLUDED_FIELD = ["cover", "playlistIndex", "bit_depth", "sample_rate", "sample_rate_kHz"]


class SongMetadata(BaseModel):
title: Optional[str] = None
Expand All @@ -26,16 +28,17 @@ class SongMetadata(BaseModel):
upc: Optional[str] = None
isrc: Optional[str] = None
playlistIndex: Optional[int] = None
bit_depth: Optional[int] = None
sample_rate: Optional[int] = None
sample_rate_kHz: Optional[str] = None

def to_itags_params(self, embed_metadata: list[str]):
tags = []
for key, value in self.model_dump().items():
if not value:
continue
if key in embed_metadata and value:
if "playlist" in key:
continue
if key == "cover":
if key in NOT_INCLUDED_FIELD:
continue
if key == "lyrics":
lrc = ttml_convent_to_lrc(value)
Expand Down Expand Up @@ -73,3 +76,8 @@ async def get_cover(self, cover_format: str, cover_size: str):

def set_playlist_index(self, index: int):
self.playlistIndex = index

def set_bit_depth_and_sample_rate(self, bit_depth: int, sample_rate: int):
self.bit_depth = bit_depth
self.sample_rate = sample_rate
self.sample_rate_kHz = str(sample_rate / 1000)
9 changes: 7 additions & 2 deletions src/mp4.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ async def get_available_codecs(m3u8_url: str) -> Tuple[list[str], list[str]]:


async def extract_media(m3u8_url: str, codec: str, song_metadata: SongMetadata,
codec_priority: list[str], alternative_codec: bool = False) -> Tuple[str, list[str], str]:
codec_priority: list[str], alternative_codec: bool = False) -> Tuple[str, list[str], str, Optional[int], Optional[int]]:
parsed_m3u8 = m3u8.loads(await download_m3u8(m3u8_url), uri=m3u8_url)
specifyPlaylist = find_best_codec(parsed_m3u8, codec)
if not specifyPlaylist and alternative_codec:
Expand Down Expand Up @@ -65,7 +65,12 @@ async def extract_media(m3u8_url: str, codec: str, song_metadata: SongMetadata,
for key in skds:
if key.endswith(key_suffix) or key.endswith(CodecKeySuffix.KeySuffixDefault):
keys.append(key)
return stream.segment_map[0].absolute_uri, keys, selected_codec
if codec == Codec.ALAC:
sample_rate, bit_depth = specifyPlaylist.media[0].extras.values()
sample_rate, bit_depth = int(sample_rate), int(bit_depth)
else:
sample_rate, bit_depth = None, None
return stream.segment_map[0].absolute_uri, keys, selected_codec, bit_depth, sample_rate


async def extract_song(raw_song: bytes, codec: str) -> SongInfo:
Expand Down
32 changes: 19 additions & 13 deletions src/rip.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ async def rip_song(song: Song, auth_params: GlobalAuthParams, codec: str, config
song_metadata.set_playlist_index(playlist.songIdIndexMapping.get(song.id))
logger.info(f"Ripping song: {song_metadata.artist} - {song_metadata.title}")
if not await exist_on_storefront_by_song_id(song.id, song.storefront, auth_params.storefront,
auth_params.anonymousAccessToken, config.region.language):
auth_params.anonymousAccessToken, config.region.language):
logger.error(
f"Unable to download song {song_metadata.artist} - {song_metadata.title}. "
f"This song does not exist in storefront {auth_params.storefront.upper()} "
Expand Down Expand Up @@ -67,22 +67,26 @@ async def rip_song(song: Song, auth_params: GlobalAuthParams, codec: str, config
f"Failed to download song: {song_metadata.artist} - {song_metadata.title}. Audio does not exist")
return
if not specified_m3u8 and not song_data.attributes.extendedAssetUrls.enhancedHls:
logger.error(f"Failed to download song: {song_metadata.artist} - {song_metadata.title}. Lossless audio does not exist")
logger.error(
f"Failed to download song: {song_metadata.artist} - {song_metadata.title}. Lossless audio does not exist")
return
if not specified_m3u8 and config.download.getM3u8FromDevice:
device_m3u8 = await device.get_m3u8(song.id)
if device_m3u8:
specified_m3u8 = device_m3u8
logger.info(f"Use m3u8 from device for song: {song_metadata.artist} - {song_metadata.title}")
if specified_m3u8:
song_uri, keys, codec_id = await extract_media(specified_m3u8, codec, song_metadata,
config.download.codecPriority,
config.download.codecAlternative)
song_uri, keys, codec_id, bit_depth, sample_rate = await extract_media(
specified_m3u8, codec, song_metadata, config.download.codecPriority, config.download.codecAlternative)
else:
song_uri, keys, codec_id = await extract_media(song_data.attributes.extendedAssetUrls.enhancedHls, codec,
song_metadata,
config.download.codecPriority,
config.download.codecAlternative)
song_uri, keys, codec_id, bit_depth, sample_rate = await extract_media(
song_data.attributes.extendedAssetUrls.enhancedHls, codec, song_metadata,
config.download.codecPriority, config.download.codecAlternative)
if all([bool(bit_depth), bool(sample_rate)]):
song_metadata.set_bit_depth_and_sample_rate(bit_depth, sample_rate)
if not force_save and check_song_exists(song_metadata, config.download, codec, playlist):
logger.info(f"Song: {song_metadata.artist} - {song_metadata.title} already exists")
return
logger.info(f"Downloading song: {song_metadata.artist} - {song_metadata.title}")
codec = get_codec_from_codec_id(codec_id)
raw_song = await download_song(song_uri)
Expand Down Expand Up @@ -119,10 +123,12 @@ async def rip_album(album: Album, auth_params: GlobalAuthParams, codec: str, con
album_info = await get_album_info(album.id, auth_params.anonymousAccessToken, album.storefront,
config.region.language)
logger.info(f"Ripping Album: {album_info.data[0].attributes.artistName} - {album_info.data[0].attributes.name}")
if not await exist_on_storefront_by_album_id(album.id, album.storefront, auth_params.storefront, auth_params.anonymousAccessToken, config.region.language):
logger.error(f"Unable to download album {album_info.data[0].attributes.artistName} - {album_info.data[0].attributes.name}. "
f"This album does not exist in storefront {auth_params.storefront.upper()} "
f"and no device is available to decrypt it")
if not await exist_on_storefront_by_album_id(album.id, album.storefront, auth_params.storefront,
auth_params.anonymousAccessToken, config.region.language):
logger.error(
f"Unable to download album {album_info.data[0].attributes.artistName} - {album_info.data[0].attributes.name}. "
f"This album does not exist in storefront {auth_params.storefront.upper()} "
f"and no device is available to decrypt it")
return
async with asyncio.TaskGroup() as tg:
for track in album_info.data[0].relationships.tracks.data:
Expand Down
12 changes: 10 additions & 2 deletions src/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,14 @@ def playlist_metadata_to_params(playlist: PlaylistInfo):
"playlistCuratorName": playlist.data[0].attributes.curatorName}


def get_audio_info_str(metadata, codec: str, config: Download):
if all([bool(metadata.bit_depth), bool(metadata.sample_rate), bool(metadata.sample_rate_kHz)]):
return config.audioInfoFormat.format(bit_depth=metadata.bit_depth, sample_rate=metadata.sample_rate,
sample_rate_kHz=metadata.sample_rate_kHz, codec=codec)
else:
return ""


def get_path_safe_dict(param: dict):
new_param = deepcopy(param)
for key, val in new_param.items():
Expand All @@ -159,13 +167,13 @@ def get_song_name_and_dir_path(codec: str, config: Download, metadata, playlist:
safe_meta = get_path_safe_dict(metadata.model_dump())
safe_pl_meta = get_path_safe_dict(playlist_metadata_to_params(playlist))
song_name = config.playlistSongNameFormat.format(codec=codec, playlistSongIndex=metadata.playlistIndex,
**safe_meta)
audio_info=get_audio_info_str(metadata, codec, config), **safe_meta)
dir_path = Path(config.playlistDirPathFormat.format(codec=codec,
**safe_meta,
**safe_pl_meta))
else:
safe_meta = get_path_safe_dict(metadata.model_dump())
song_name = config.songNameFormat.format(codec=codec, **safe_meta)
song_name = config.songNameFormat.format(codec=codec, audio_info=get_audio_info_str(metadata, codec, config), **safe_meta)
dir_path = Path(config.dirPathFormat.format(codec=codec, **safe_meta))
if sys.platform == "win32":
song_name = get_valid_filename(song_name)
Expand Down

0 comments on commit 0d98109

Please sign in to comment.