Add deezer module

bot/modules/deezer/__init__.py

@@ -0,0 +1,9 @@
+from .deezer import Deezer
+from bot.utils.config import config
+
+
+deezer = Deezer(
+    arl=config.tokens.deezer.arl,
+)
+
+__all__ = ['deezer']

bot/modules/deezer/deezer.py

@@ -0,0 +1,16 @@
+import asyncio
+
+from .song import Songs
+from .engine import DeezerEngine
+from .driver import DeezerDriver
+from .downloader import DownloaderBuilder
+
+
+class Deezer(object):
+    def __init__(self, arl: str):
+        self.engine = asyncio.get_event_loop().run_until_complete(
+            DeezerEngine.from_arl(arl)
+        )
+        self.driver = DeezerDriver(engine=self.engine)
+        self.songs = Songs(driver=self.driver)
+        self.downloader = DownloaderBuilder(driver=self.driver)

bot/modules/deezer/downloader.py

@@ -0,0 +1,98 @@
+from attrs import define
+
+from io import BytesIO
+
+from .driver import DeezerDriver
+
+from . import track_formats
+from .util import UrlDecrypter, ChunkDecrypter
+from .song import FullSongItem
+
+
+@define
+class DeezerBytestream:
+    file: bytes
+    filename: str
+    song: FullSongItem
+
+    @classmethod
+    def from_bytestream(
+            cls,
+            bytestream: BytesIO,
+            filename: str,
+            full_song: FullSongItem
+    ):
+        bytestream.seek(0)
+        return cls(
+            file=bytestream.read(),
+            filename=filename,
+            song=full_song,
+        )
+
+
+@define
+class Downloader:
+    driver: DeezerDriver
+    song_id: int
+    track: dict
+    song: FullSongItem
+
+    @classmethod
+    async def build(
+            cls,
+            song_id: int,
+            driver: DeezerDriver
+    ):
+        track = await driver.reverse_get_track(song_id)
+        return cls(
+            song_id=song_id,
+            driver=driver,
+            track=track['results'],
+            song=await FullSongItem.from_deezer(track)
+        )
+
+    async def to_bytestream(self) -> DeezerBytestream:
+        quality = track_formats.MP3_128
+
+        decrypter = ChunkDecrypter.from_track_id(self.song_id)
+        i = 0
+        audio = BytesIO()
+
+        async for chunk in self.driver.engine.get_data_iter(
+                await self._get_download_url(quality=quality)
+        ):
+            if i % 3 > 0 or len(chunk) < 2 * 1024:
+                audio.write(chunk)
+            else:
+                audio.write(decrypter.decrypt_chunk(chunk))
+            i += 1
+
+        return DeezerBytestream.from_bytestream(
+            filename=self.song.full_name + track_formats.TRACK_FORMAT_MAP[quality].ext,
+            bytestream=audio,
+            full_song=self.song
+        )
+
+    async def _get_download_url(self, quality: str = 'MP3_128'):
+        md5_origin = self.track["MD5_ORIGIN"]
+        track_id = self.track["SNG_ID"]
+        media_version = self.track["MEDIA_VERSION"]
+
+        url_decrypter = UrlDecrypter(
+            md5_origin=md5_origin,
+            track_id=track_id,
+            media_version=media_version
+        )
+
+        return url_decrypter.get_url_for(track_formats.TRACK_FORMAT_MAP[quality])
+
+
+@define
+class DownloaderBuilder:
+    driver: DeezerDriver
+
+    async def from_id(self, song_id: int):
+        return await Downloader.build(
+            song_id=song_id,
+            driver=self.driver
+        )

bot/modules/deezer/driver.py

@@ -0,0 +1,36 @@
+from attrs import define
+
+from .engine import DeezerEngine
+
+from .util import clean_query
+
+
+@define
+class DeezerDriver:
+    engine: DeezerEngine
+
+    async def get_track(self, track_id: int | str):
+        data = await self.engine.call_legacy_api(
+            f'track/{track_id}'
+        )
+
+        return data
+
+    async def reverse_get_track(self, track_id: int | str):
+        return await self.engine.call_api(
+            'song.getData',
+            params={
+                'SNG_ID': str(track_id)
+            }
+        )
+
+    async def search(self, query: str, limit: int = 30):
+        data = await self.engine.call_legacy_api(
+            'search/track',
+            params={
+                'q': clean_query(query),
+                'limit': limit
+            }
+        )
+
+        return data['data']

bot/modules/deezer/engine.py

@@ -0,0 +1,93 @@
+import aiohttp
+
+from aiohttp import ClientResponse
+
+from attrs import define
+
+
+HTTP_HEADERS = {
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+                  "(KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36",
+    "Content-Language": "en-US",
+    "Cache-Control": "max-age=0",
+    "Accept": "*/*",
+    "Accept-Charset": "utf-8,ISO-8859-1;q=0.7,*;q=0.3",
+    "Accept-Language": "en-US,en;q=0.9,en-US;q=0.8,en;q=0.7",
+    "Connection": 'keep-alive'
+}
+
+
+@define
+class DeezerEngine:
+    cookies: dict
+    token: str = None
+
+    @classmethod
+    async def from_arl(cls, arl: str):
+        cookies = {'arl': arl}
+        data, cookies = await cls(cookies).call_api(
+            'deezer.getUserData', get_cookies=True
+        )
+
+        data = data['results']
+        token = data['checkForm']
+
+        return cls(
+            cookies=cookies,
+            token=token
+        )
+
+    async def call_legacy_api(
+            self, request_point: str, params: dict = None
+    ):
+        async with aiohttp.ClientSession(cookies=self.cookies) as session:
+            async with session.get(
+                    f"https://api.deezer.com/{request_point}",
+                    params=params,
+                    headers=HTTP_HEADERS
+            ) as r:
+                return await r.json()
+
+    @staticmethod
+    async def _iter_exact_chunks(response: ClientResponse, chunk_size: int = 2048):
+        buffer = b""
+        async for chunk in response.content.iter_any():
+            buffer += chunk
+            while len(buffer) >= chunk_size:
+                yield buffer[:chunk_size]
+                buffer = buffer[chunk_size:]
+        if buffer:
+            yield buffer
+
+    async def get_data_iter(self, url: str):
+        async with aiohttp.ClientSession(
+                cookies=self.cookies,
+                headers=HTTP_HEADERS
+        ) as session:
+            r = await session.get(
+                url,
+                allow_redirects=True
+            )
+            async for chunk in self._iter_exact_chunks(r):
+                yield chunk
+
+    async def call_api(
+            self, method: str, params: dict = None,
+            get_cookies: bool = False
+    ):
+        async with aiohttp.ClientSession(cookies=self.cookies) as session:
+            async with session.post(
+                    f"https://www.deezer.com/ajax/gw-light.php",
+                    params={
+                        'method': method,
+                        'api_version': '1.0',
+                        'input': '3',
+                        'api_token': self.token or 'null',
+                    },
+                    headers=HTTP_HEADERS,
+                    json=params
+            ) as r:
+                if not get_cookies:
+                    return await r.json()
+                else:
+                    return await r.json(), r.cookies

bot/modules/deezer/song.py

@@ -0,0 +1,92 @@
+from attrs import define
+
+from .driver import DeezerDriver
+
+
+@define
+class SongItem:
+    name: str
+    id: int
+    artist: str
+    preview_url: str | None
+    thumbnail: str
+
+    @classmethod
+    def from_deezer(cls, song_item: dict):
+        return cls(
+            name=song_item['title'],
+            id=song_item['id'],
+            artist=song_item['artist']['name'],
+            preview_url=song_item.get('preview'),
+            thumbnail=song_item['album']['cover_medium']
+        )
+
+    @property
+    def full_name(self):
+        return f"{self.artist} - {self.name}"
+
+    def __str__(self):
+        return self.full_name
+
+
+@define
+class FullSongItem:
+    name: str
+    id: int
+    artists: list[str]
+    preview_url: str | None
+    duration: int
+    thumbnail: str
+
+    @classmethod
+    async def from_deezer(cls, song_item: dict):
+        if song_item.get('results'):
+            song_item = song_item['results']
+
+        return cls(
+            name=song_item['SNG_TITLE'],
+            id=song_item['SNG_ID'],
+            artists=[artist['ART_NAME'] for artist in song_item['ARTISTS']],
+            preview_url=(song_item.get('MEDIA').get('HREF')
+                         if type(song_item.get('MEDIA')) is dict and
+                         song_item.get('MEDIA').get('TYPE') == 'preview'
+                         else None),
+            thumbnail=f'https://e-cdns-images.dzcdn.net/images/cover/'
+                      f'{song_item["ALB_PICTURE"]}/320x320.jpg',
+            duration=int(song_item['DURATION'])
+        )
+
+    @property
+    def all_artists(self):
+        return ', '.join(self.artists)
+
+    @property
+    def full_name(self):
+        return f"{self.all_artists} - {self.name}"
+
+    def __str__(self):
+        return self.full_name
+
+
+@define
+class Songs(object):
+    driver: DeezerDriver
+
+    async def search(self, query: str, limit: int = 30) -> list[SongItem] | None:
+        r = await self.driver.search(query, limit=limit)
+
+        if r is None:
+            return None
+
+        return [SongItem.from_deezer(item) for item in r]
+
+    async def search_one(self, query: str) -> SongItem | None:
+        return (await self.search(query, limit=1) or [None])[0]
+
+    async def from_id(self, song_id: int) -> FullSongItem | None:
+        r = await self.driver.reverse_get_track(song_id)
+
+        if r is None:
+            return None
+
+        return await FullSongItem.from_deezer(r)

bot/modules/deezer/track_formats.py

@@ -0,0 +1,50 @@
+from dataclasses import dataclass
+
+FLAC = "FLAC"
+MP3_128 = "MP3_128"
+MP3_256 = "MP3_256"
+MP3_320 = "MP3_320"
+MP4_RA1 = "MP4_RA1"
+MP4_RA2 = "MP4_RA2"
+MP4_RA3 = "MP4_RA3"
+
+FALLBACK_QUALITIES = [MP3_320, MP3_128, FLAC]
+FORMAT_LIST = [MP3_128, MP3_256, MP3_320, FLAC]
+
+
+@dataclass
+class TrackFormat:
+    code: int
+    ext: str
+
+
+TRACK_FORMAT_MAP = {
+    FLAC: TrackFormat(
+        code=9,
+        ext=".flac"
+    ),
+    MP3_128: TrackFormat(
+        code=1,
+        ext=".mp3"
+    ),
+    MP3_256: TrackFormat(
+        code=5,
+        ext=".mp3"
+    ),
+    MP3_320: TrackFormat(
+        code=3,
+        ext=".mp3"
+    ),
+    MP4_RA1: TrackFormat(
+        code=13,
+        ext=".mp4"
+    ),
+    MP4_RA2: TrackFormat(
+        code=14,
+        ext=".mp4"
+    ),
+    MP4_RA3: TrackFormat(
+        code=15,
+        ext=".mp3"
+    )
+}

bot/modules/deezer/util.py

@@ -0,0 +1,93 @@
+# https://pypi.org/project/music-helper/
+
+import re
+import hashlib
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
+from attrs import define
+
+from .track_formats import TrackFormat
+
+
+def clean_query(query):
+    query = re.sub(r"/ feat[.]? /g", " ", query)
+    query = re.sub(r"/ ft[.]? /g", " ", query)
+    query = re.sub(r"/\(feat[.]? /g", " ", query)
+    query = re.sub(r"/\(ft[.]? /g", " ", query)
+    query = re.sub(r"/&/g", "", query)
+    query = re.sub(r"/–/g", "-", query)
+    query = re.sub(r"/–/g", "-", query)
+
+    return query
+
+
+def get_text_md5(text, encoding="UTF-8"):
+    return hashlib.md5(str(text).encode(encoding)).hexdigest()
+
+
+@define
+class UrlDecrypter:
+    md5_origin: str
+    track_id: str
+    media_version: str
+
+    def get_url_for(self, track_format: TrackFormat):
+        step1 = (f'{self.md5_origin}¤{track_format.code}¤'
+                 f'{self.track_id}¤{self.media_version}')
+        m = hashlib.md5()
+        m.update(bytes([ord(x) for x in step1]))
+
+        step2 = f'{m.hexdigest()}¤{step1}¤'
+        step2 = step2.ljust(80, " ")
+
+        cipher = Cipher(
+            algorithm=algorithms.AES(
+                key=bytes('jo6aey6haid2Teih', 'ascii')
+            ),
+            mode=modes.ECB(),
+            backend=default_backend()
+        )
+
+        encryptor = cipher.encryptor()
+        step3 = encryptor.update(bytes([ord(x) for x in step2])).hex()
+
+        cdn = self.md5_origin[0]
+
+        return f'https://e-cdns-proxy-{cdn}.dzcdn.net/mobile/1/{step3}'
+
+
+@define
+class ChunkDecrypter:
+    cipher: Cipher
+
+    @classmethod
+    def from_track_id(cls, track_id: int):
+        cipher = Cipher(
+            algorithms.Blowfish(get_blowfish_key(str(track_id))),
+            modes.CBC(bytes([i for i in range(8)])),
+            default_backend()
+        )
+
+        return cls(
+            cipher=cipher
+        )
+
+    def decrypt_chunk(self, chunk: bytes):
+        decryptor = self.cipher.decryptor()
+        return decryptor.update(chunk) + decryptor.finalize()
+
+
+def get_blowfish_key(track_id: str):
+    secret = 'g4el58wc0zvf9na1'
+
+    m = hashlib.md5()
+    m.update(bytes([ord(x) for x in track_id]))
+    id_md5 = m.hexdigest()
+
+    blowfish_key = bytes(
+        [(ord(id_md5[i]) ^ ord(id_md5[i + 16]) ^ ord(secret[i])) for i in range(16)]
+    )
+
+    return blowfish_key

pyproject.toml

@@ -18,6 +18,7 @@ attrs = "^23.1.0"
 ytmusicapi = "^1.3.0"
 pytube = "^15.0.0"
 pydub = "^0.25.1"
+aiohttp = "^3.8.6"
 
 
 [build-system]