cogs/functions.py

import asyncio
from unittest import result
import yt_dlp
import os
from shazamio import Shazam
import aiohttp
import base64
from functools import partial

CLIENT_ID = "550edc8d6d294381a0f4dac9c8c9fea5"
CLIENT_SECRET = "fade5c01cc8b40acb79ad1f598074b4c"

async def music_recognition(filepath):
    shazam = Shazam()
    out = await shazam.recognize(filepath)
    songname = out['track']['title']
    artist = out["track"]["subtitle"]
    ydl_opts = {
        "quiet": True,
        "skip_download": True,
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(f"ytsearch1:{songname} {artist}", download=False)
        video = info["entries"][0]
        title = video["title"]
        yturl = video["webpage_url"]
        thumbnail = video["thumbnail"]
        duration = video["duration"]
        spotifyshit = await spotify_search(f"{songname} {artist}")
        spotifylink = spotifyshit["tracks"]["items"][0]["external_urls"]["spotify"]
        return {
            "songname": songname,
            "artist": artist,
            "title": title,
            "yturl": yturl,
            "thumbnail": thumbnail,
            "duration": duration,
            "spotifylink": spotifylink
        }


async def spotify_auth():
    auth_url = "https://accounts.spotify.com/api/token"
    auth_string = f"{CLIENT_ID}:{CLIENT_SECRET}"
    auth_header = base64.b64encode(auth_string.encode()).decode()

    headers = {
        "Authorization": f"Basic {auth_header}",
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "grant_type": "client_credentials"
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(auth_url, headers=headers, data=data) as resp:
            response = await resp.json()
            return response["access_token"]
  
async def spotify_search(query: str, search_type="track"):
    token = await spotify_auth()

    url = "https://api.spotify.com/v1/search"
    params = {
        "q": query,
        "type": search_type,
        "limit": 1
    }

    headers = {
        "Authorization": f"Bearer {token}"
    }

    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers, params=params) as resp:
            data = await resp.json()
            return data


async def download_mp3(query: str):
    def _download():
        ydl_opts = {
            "quiet": True,
            "format": "bestaudio/best",
            "outtmpl": "./assets/audio.%(ext)s",
            "postprocessors": [{
                "key": "FFmpegExtractAudio",
                "preferredcodec": "mp3",
                "preferredquality": "192",
            }],
        }
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            ydl.extract_info(f"ytsearch1:{query}", download=True)
        return "./assets/audio.mp3"
    return await asyncio.to_thread(_download)