music-downloader/downloader/downloader.py

import hashlib
import config

from .utils import system
from .type import Album, Track
from genius.type import Album as gAlbum
import asyncio
from pathlib import Path

import genius
from .mp3 import set_track
from .slice import slice_audio


class Downloader:
    @staticmethod
    async def download_audio(url: str):
        if "youtu.be" in url:
            filename = url.split("/")[-1] + ".mp3"
        else:
            filename = url.split("?")[-1] + ".mp3"

        path = config.tmp_folder / filename
        if path.exists():
            return path

        command = f'youtube-dl -o "tmp/%(id)s.%(ext)s" -x --audio-format mp3 {url}'
        await system(command)
        return path

    @staticmethod
    async def download_album(album: Album):
        tasks = []
        if "watch" in album.audio_url or "youtu.be" in album.audio_url:
            _, genius_album = await asyncio.gather(
                asyncio.Task(Downloader.download_audio(album.audio_url)),
                genius.get_album(album.meta_url)
            )
        for track in album.tracks:
            tasks.append(
                asyncio.Task(Downloader.download_track(track, genius_album))
            )

        return genius_album, await asyncio.gather(*tasks)

    @staticmethod
    async def download_track(track: Track, album: gAlbum = None):
        hash = hashlib.md5(track.meta_url.split("/")[-1].encode()).hexdigest()

        async def get_audio():
            src = await Downloader.download_audio(track.audio_url)
            return await slice_audio(src, config.tmp_folder / f"{hash}.mp3", track.timecode)

        dst, meta = await asyncio.gather(
            asyncio.Task(get_audio()),
            asyncio.Task(genius.get_track(track.meta_url))
        )

        meta.album = album
        meta.number = album.track_urls[meta.url]
        await set_track(dst, meta)
        return dst, meta

    @staticmethod
    async def download(object):
        res: dict[Path, Track] = {}
        album = None
        if object.type == "album":
            album, tracks = await Downloader.download_album(object)
            for key, value in tracks:
                res[key] = value
        else:
            key, value = await Downloader.download_track(object)
            res[key] = value

        return album, res