Yandex.Music बॉट क्लाइंट के हुड के तहत

import struct
import base64

from ext.query_types import QueryType


class BadDataVersion(Exception):
    pass


class CallbackData:
    ACTUAL_VERSION = 7
    BASE_FORMAT = '<BB'

    def __init__(self, type_: QueryType, data=None, version=None):
        self.type = type_
        self.version = version or CallbackData.ACTUAL_VERSION
        self.data = data

        if self.data is not None and not isinstance(self.data, list):
            self.data = [self.data]

        if self.data is not None:
            self.data = self._normalize_data_to_format(self.data)

    def __repr__(self):
        return f'<CallbackData> Type: {self.type} Version: {self.version} Data: {self.data}'

    def _normalize_data_to_format(self, data, bytes_object=False):
        normalized_data = data.copy()
        for i, c in enumerate(self.type.format):
            cast = str
            if c.lower() in 'bhilqn':
                cast = int
            elif c in 'efd':
                cast = float
            elif c == '?':
                cast = bool

            casted = cast(data[i])
            if bytes_object and cast == str:
                casted = casted.encode('utf-8')
            normalized_data[i] = casted

        return normalized_data

    @staticmethod
    def decode_type(callback_data):
        decoded = base64.b85decode(callback_data.encode('utf-8'))
        type_, version = struct.unpack(CallbackData.BASE_FORMAT, decoded[:2])

        if CallbackData.ACTUAL_VERSION != version:
            raise BadDataVersion()

        return QueryType(type_), version, decoded

    @classmethod
    def decode(cls, type_, version, decoded):
        start, data = 2, []

        if start < len(decoded):
            format_iter = iter(type_.format)

            while True:
                if start >= len(decoded):
                    break

                format_ = next(format_iter, type_.format[-1])

                decode_str = format_ in 'ps'
                if decode_str:
                    # struct.calcsize('b') = 1
                    length = list(struct.unpack('b', decoded[start: start + 1]))[0]
                    start += 1

                    format_ = f'{length}{format_}'

                step = struct.calcsize(format_)

                unpacked = list(struct.unpack(f'{format_}', decoded[start: start + step]))
                if decode_str:
                    unpacked[0] = unpacked[0].decode('UTF-8')

                data += unpacked
                start += step

        return cls(type_, data if data else None, version)

    def encode(self):
        encode = struct.pack(self.BASE_FORMAT, self.type.value, self.version)

        if self.data is not None:
            format_iter = iter(self.type.format)
            normalized_data = self._normalize_data_to_format(self.data, bytes_object=True)

            for data in normalized_data:
                format_ = next(format_iter, self.type.format[-1])

                if format_ in 'ps':
                    #        .  'b'  
                    # -      ,    > 36 
                    encode += struct.pack('b', len(data))
                    encode += struct.pack(f'{len(data)}{format_}', data)
                else:
                    encode += struct.pack(f'{format_}', data)

        return base64.b85encode(encode).decode('utf-8')

import copy
import time

from typing import Union
from collections import OrderedDict

import requests

from yandex_music.utils.request import Request as YandexMusicRequest


class LimitedSizeDict(OrderedDict):
    def __init__(self, *args, **kwargs):
        self.size_limit = kwargs.pop("size_limit", None)
        OrderedDict.__init__(self, *args, **kwargs)
        self._check_size_limit()

    def __setitem__(self, key, value):
        OrderedDict.__setitem__(self, key, value)
        self._check_size_limit()

    def _check_size_limit(self):
        if self.size_limit is not None:
            while len(self) > self.size_limit:
                self.popitem(last=False)


class CachedItem:
    def __init__(self, response: requests.Response):
        self.timestamp = time.time()
        self.response = response


class Cache:
    __singleton: 'Cache' = None

    def __init__(self, lifetime: Union[str, int], size: Union[str, int]):
        Cache.__singleton = self

        self.lifetime = int(lifetime) * 60
        self.size = int(size)

        self.storage: LimitedSizeDict = LimitedSizeDict(size_limit=int(size))

    def get(self, *args, **kwargs):
        hash_ = Cache.get_hash(*args, **kwargs)

        return self.storage[hash_].response

    def update(self, response: requests.Response, *args, **kwargs):
        hash_ = Cache.get_hash(*args, **kwargs)
        self.storage.update({hash_: CachedItem(response)})

    def need_to_fetch(self, *args, **kwargs):
        hash_ = Cache.get_hash(*args, **kwargs)
        cached_item = self.storage.get(hash_)

        if not cached_item:
            return True

        if time.time() - cached_item.timestamp > self.lifetime:
            return True

        return False

    @classmethod
    def get_instance(cls) -> 'Cache':
        if cls.__singleton is not None:
            return cls.__singleton
        else:
            raise RuntimeError(f'{cls.__name__} not initialized')

    @staticmethod
    def freeze_dict(d: dict) -> Union[frozenset, tuple, dict]:
        if isinstance(d, dict):
            return frozenset((key, Cache.freeze_dict(value)) for key, value in d.items())
        elif isinstance(d, list):
            return tuple(Cache.freeze_dict(value) for value in d)

        return d

    @staticmethod
    def get_hash(*args, **kwargs):
        return hash((args, Cache.freeze_dict(kwargs)))


class Request(YandexMusicRequest):
    def _request_wrapper(self, *args, **kwargs):
        use_cache = kwargs.get('use_cache', True)

        if 'use_cache' in kwargs:
            kwargs.pop('use_cache')

        if not use_cache:
            response = super()._request_wrapper(*args, **copy.deepcopy(kwargs))
        elif use_cache and Cache.get_instance().need_to_fetch(*args, **kwargs):
            response = super()._request_wrapper(*args, **copy.deepcopy(kwargs))
            Cache.get_instance().update(response, *args, **kwargs)
        else:
            response = Cache.get_instance().get(*args, **kwargs)

        return response

import uuid

import lomond


def get_auth_data():
    return {
        "event": {
            "header": {
                "messageId": str(uuid.uuid4()),
                "name": "SynchronizeState",
                "namespace": "System"
            },
            "payload": {
                "accept_invalid_auth": True,
                "auth_token": "5983ba91-339e-443c-8452-390fe7d9d308",
                "uuid": str(uuid.uuid4()).replace('-', ''),
            }
        }
    }


def get_asr_data():
    return {
        "event": {
            "header": {
                "messageId": str(uuid.uuid4()),
                "name": "Recognize",
                "namespace": "ASR",
                "streamId": 1
            },
            "payload": {
                "advancedASROptions": {
                    "manual_punctuation": False,
                    "partial_results": False
                },
                "disableAntimatNormalizer": False,
                "format": "audio/opus",
                "music_request2": {
                    "headers": {
                        "Content-Type": "audio/opus"
                    }
                },
                "punctuation": False,
                "tags": "PASS_AUDIO;",
                "topic": "queries"
            }
        }
    }


class Recognition:
    URI = 'wss://voiceservices.yandex.net/uni.ws'
    LANGS = ['', 'ru-RU', 'en-US']
    POLL_DELAY = 0.3

    def __init__(self, binary_data, status_msg):
        self.status_msg = status_msg
        self.websocket = lomond.WebSocket(self.URI)
        self.chunks = self.get_chunks_and_replace_encoder(binary_data)

    def get_track_id(self):
        for lang in Recognition.LANGS:
            asr_data = get_asr_data()
            if lang:
                asr_data['event']['payload'].update({'lang': lang})
                self.status_msg.edit_text(f'     {lang}...')
            else:
                self.status_msg.edit_text(f'      ...')

            for msg in self.websocket.connect(poll=self.POLL_DELAY):
                if msg.name == 'ready':
                    self.websocket.send_json(get_auth_data())
                    self.websocket.send_json(asr_data)

                    for chunk in self.chunks:
                        self.websocket.send_binary(chunk)

                if msg.name == 'text':
                    response = msg.json.get('directive')

                    if self.is_valid_response(response):
                        self.websocket.close()

                        return self.parse_track_id(response)
                    elif self.is_fatal_error(response):
                        self.websocket.close()

                        break

    def is_valid_response(self, response):
        if response.get('header', {}).get('name') == 'MusicResult' and \
                response.get('payload', {}).get('result') == 'success':
            self.status_msg.edit_text(f' ,  !')
            return True
        return False

    @staticmethod
    def is_fatal_error(response):
        if response.get('header', {}).get('name') == 'MusicResult' and \
                response.get('payload', {}).get('result') == 'music':
            return False
        return True

    @staticmethod
    def parse_track_id(response):
        return response['payload']['data']['match']['id']

    @staticmethod
    def get_chunks_and_replace_encoder(binary_data):
        chunks = []

        for chunk in binary_data.split(b'OggS')[1:]:
            if b'OpusTags' in chunk:
                pos = chunk.index(b'OpusTags') + 12
                size = len(chunk)
                chunk = chunk[:pos] + b'#\x00\x00\x00\x00ENCODER=SpeechKit Mobile SDK v3.28.0'
                chunk += b"\x00" * (size - len(chunk))

            chunks.append(b'\x00\x00\x00\x01OggS' + chunk)

        return chunks