Compare commits
No commits in common. 'c04ce9cb75337e539b6d4eb235b306683da5a504' and 'f06a7c2530fe1b09d22e878195399576c4120841' have entirely different histories.
c04ce9cb75
...
f06a7c2530
@ -1,69 +0,0 @@ |
|||||||
from typing import Optional |
|
||||||
|
|
||||||
import m3u8 |
|
||||||
from loguru import logger |
|
||||||
from pydantic import BaseModel |
|
||||||
|
|
||||||
from src.api import get_song_info, get_m3u8_from_api, download_m3u8 |
|
||||||
from src.config import Config, Device |
|
||||||
from src.exceptions import CodecNotFoundException |
|
||||||
from src.metadata import SongMetadata |
|
||||||
from src.types import GlobalAuthParams |
|
||||||
from src.url import Song |
|
||||||
from src.utils import get_codec_from_codec_id |
|
||||||
|
|
||||||
|
|
||||||
async def get_available_audio_quality(m3u8_url: str): |
|
||||||
parsed_m3u8 = m3u8.loads(await download_m3u8(m3u8_url), uri=m3u8_url) |
|
||||||
result = [] |
|
||||||
for playlist in parsed_m3u8.playlists: |
|
||||||
if get_codec_from_codec_id(playlist.stream_info.audio): |
|
||||||
result.append(AudioQuality(codec_id=playlist.stream_info.audio, |
|
||||||
codec=get_codec_from_codec_id(playlist.stream_info.audio), |
|
||||||
bitrate=playlist.stream_info.bandwidth, |
|
||||||
average_bitrate=playlist.stream_info.average_bandwidth, |
|
||||||
channels=playlist.media[0].channels, |
|
||||||
sample_rate=playlist.media[0].extras.get("sample_rate", None), |
|
||||||
bit_depth=playlist.media[0].extras.get("bit_depth", None))) |
|
||||||
return result |
|
||||||
|
|
||||||
|
|
||||||
class AudioQuality(BaseModel): |
|
||||||
codec_id: str |
|
||||||
codec: str |
|
||||||
bitrate: int |
|
||||||
average_bitrate: int |
|
||||||
channels: str |
|
||||||
sample_rate: Optional[int] = None |
|
||||||
bit_depth: Optional[int] = None |
|
||||||
|
|
||||||
|
|
||||||
async def get_available_song_audio_quality(song: Song, config: Config, auth_params: GlobalAuthParams, |
|
||||||
device: Device) -> tuple[SongMetadata, list[AudioQuality]]: |
|
||||||
specified_m3u8 = None |
|
||||||
token = auth_params.anonymousAccessToken |
|
||||||
song_data = await get_song_info(song.id, token, song.storefront, config.region.language) |
|
||||||
song_metadata = SongMetadata.parse_from_song_data(song_data) |
|
||||||
if config.m3u8Api.enable: |
|
||||||
m3u8_url = await get_m3u8_from_api(config.m3u8Api.endpoint, song.id, config.m3u8Api.enable) |
|
||||||
if m3u8_url: |
|
||||||
specified_m3u8 = m3u8_url |
|
||||||
logger.info(f"Use m3u8 from API for song: {song_metadata.artist} - {song_metadata.title}") |
|
||||||
if not song_data.attributes.extendedAssetUrls: |
|
||||||
logger.error( |
|
||||||
f"Failed to get audio quality fo song: {song_metadata.artist} - {song_metadata.title}. Audio does not exist") |
|
||||||
raise CodecNotFoundException |
|
||||||
if not song_data.attributes.extendedAssetUrls.enhancedHls: |
|
||||||
logger.error( |
|
||||||
f"Failed to get audio quality for song: {song_metadata.artist} - {song_metadata.title}. Lossless audio does not exist") |
|
||||||
raise CodecNotFoundException |
|
||||||
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: |
|
||||||
audio_qualities = await get_available_audio_quality(specified_m3u8) |
|
||||||
else: |
|
||||||
audio_qualities = await get_available_audio_quality(song_data.attributes.extendedAssetUrls.enhancedHls) |
|
||||||
return song_metadata, audio_qualities |
|
Loading…
Reference in new issue