Updated the transformer
This commit is contained in:
Choco
@@ -22,9 +22,9 @@ SOFTWARE.
|
||||
"""
|
||||
|
||||
__version__ = "1.4"
|
||||
__author__ = 'Vocard Development, Choco'
|
||||
__author__ = 'Vocard Development, ChocoMeow'
|
||||
__license__ = "MIT"
|
||||
__copyright__ = "Copyright 2023 - present (c) Vocard Development, Choco"
|
||||
__copyright__ = "Copyright 2023 - present (c) Vocard Development, ChocoMeow"
|
||||
|
||||
from .enums import SearchType, LoopType
|
||||
from .events import *
|
||||
@@ -35,4 +35,4 @@ from .player import Player, connect_channel
|
||||
from .pool import *
|
||||
from .queue import *
|
||||
from .placeholders import Placeholders, build_embed
|
||||
from .formatter import encode, decode
|
||||
from .transformer import encode, decode
|
||||
|
||||
@@ -1,242 +0,0 @@
|
||||
"""MIT License
|
||||
|
||||
Copyright (c) 2023 - present Vocard Development
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64, io, abc, struct, dataclasses
|
||||
|
||||
from typing import Union, BinaryIO, Optional, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from .objects import Track
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Codec:
|
||||
encoding: str
|
||||
error_handler: str
|
||||
|
||||
def encode(self, data: str) -> bytes:
|
||||
return data.encode(self.encoding, self.error_handler)
|
||||
|
||||
def decode(self, data: bytes) -> str:
|
||||
return data.decode(self.encoding, self.error_handler)
|
||||
|
||||
UTF8 = Codec("utf-8", "surrogatepass")
|
||||
|
||||
_FORMAT_BOOL = "?"
|
||||
_FORMAT_BYTE = "b"
|
||||
_FORMAT_INT = ">i"
|
||||
_FORMAT_LONG = ">q"
|
||||
_FORMAT_USHORT = ">H"
|
||||
|
||||
class HasStream(abc.ABC):
|
||||
@property
|
||||
@abc.abstractmethod
|
||||
def stream(self) -> BinaryIO:
|
||||
...
|
||||
|
||||
class Reader(HasStream):
|
||||
def __init__(self, stream: Union[BinaryIO, HasStream]) -> None:
|
||||
self._stream: BinaryIO = stream.stream if isinstance(stream, HasStream) else stream
|
||||
|
||||
@property
|
||||
def stream(self) -> BinaryIO:
|
||||
return self._stream
|
||||
|
||||
def read_bool(self) -> bool:
|
||||
return struct.unpack(_FORMAT_BOOL, self._stream.read(1))[0]
|
||||
|
||||
def read_byte(self) -> int:
|
||||
return struct.unpack(_FORMAT_BYTE, self._stream.read(1))[0]
|
||||
|
||||
def read_int(self) -> int:
|
||||
return struct.unpack(_FORMAT_INT, self._stream.read(4))[0]
|
||||
|
||||
def read_long(self) -> int:
|
||||
return struct.unpack(_FORMAT_LONG, self._stream.read(8))[0]
|
||||
|
||||
def read_ushort(self) -> int:
|
||||
return struct.unpack(_FORMAT_USHORT, self._stream.read(2))[0]
|
||||
|
||||
def read_utf(self) -> str:
|
||||
length = self.read_ushort()
|
||||
data = self._stream.read(length)
|
||||
return UTF8.decode(data)
|
||||
|
||||
def read_optional_utf(self) -> Optional[str]:
|
||||
if self.read_bool():
|
||||
return self.read_utf()
|
||||
else:
|
||||
return None
|
||||
|
||||
class Writer:
|
||||
_stream: BinaryIO
|
||||
|
||||
def __init__(self, stream: Union[BinaryIO, HasStream] = None ) -> None:
|
||||
if stream is None:
|
||||
stream = io.BytesIO()
|
||||
elif isinstance(stream, HasStream):
|
||||
stream = stream.stream
|
||||
|
||||
self._stream = stream
|
||||
self._string_codec: Codec = UTF8
|
||||
|
||||
@property
|
||||
def stream(self) -> BinaryIO:
|
||||
return self._stream
|
||||
|
||||
def write_bool(self, data: bool) -> None:
|
||||
self._stream.write(struct.pack(_FORMAT_BOOL, data))
|
||||
|
||||
def write_byte(self, data: int) -> None:
|
||||
self._stream.write(struct.pack(_FORMAT_BYTE, data))
|
||||
|
||||
def write_int(self, data: int) -> None:
|
||||
self._stream.write(struct.pack(_FORMAT_INT, data))
|
||||
|
||||
def write_long(self, data: int) -> None:
|
||||
self._stream.write(struct.pack(_FORMAT_LONG, data))
|
||||
|
||||
def write_ushort(self, data: int) -> None:
|
||||
self._stream.write(struct.pack(_FORMAT_USHORT, data))
|
||||
|
||||
def write_utf(self, data: str) -> None:
|
||||
data = self._string_codec.encode(data)
|
||||
self.write_ushort(len(data))
|
||||
self._stream.write(data)
|
||||
|
||||
def write_optional_utf(self, data: Optional[str]) -> None:
|
||||
if data is None:
|
||||
self.write_bool(False)
|
||||
else:
|
||||
self.write_bool(True)
|
||||
self.write_utf(data)
|
||||
|
||||
class MessageInput(HasStream):
|
||||
def __init__(self, stream: Union[BinaryIO, HasStream]) -> None:
|
||||
self._stream: Reader = Reader(stream)
|
||||
self._flags: int = 0
|
||||
self._size: int = 0
|
||||
|
||||
@property
|
||||
def stream(self) -> BinaryIO:
|
||||
return self._stream.stream
|
||||
|
||||
@property
|
||||
def flags(self) -> int:
|
||||
return self._flags
|
||||
|
||||
def next(self) -> Optional[Reader]:
|
||||
value = self._stream.read_int()
|
||||
self._flags = (value & 0xC0000000) >> 30
|
||||
self._size = value & 0x3FFFFFFF
|
||||
|
||||
if not self._size:
|
||||
return None
|
||||
|
||||
data = self._stream.stream.read(self._size)
|
||||
|
||||
return Reader(io.BytesIO(data))
|
||||
|
||||
class MessageOutput(HasStream):
|
||||
_stream: Writer
|
||||
_body_stream: io.BytesIO
|
||||
|
||||
def __init__(self, stream: Union[BinaryIO, HasStream]) -> None:
|
||||
self._string_codec: Codec = UTF8
|
||||
|
||||
self._stream = Writer(stream)
|
||||
self._body_stream = io.BytesIO()
|
||||
|
||||
@property
|
||||
def stream(self) -> BinaryIO:
|
||||
return self._stream.stream
|
||||
|
||||
def start(self) -> Writer:
|
||||
self._body_stream.truncate(0)
|
||||
self._body_stream.seek(0)
|
||||
return Writer(self._body_stream)
|
||||
|
||||
def commit(self, flags: int = None) -> None:
|
||||
data = self._body_stream.getvalue()
|
||||
header = len(data)
|
||||
if flags:
|
||||
header |= flags << 30
|
||||
|
||||
self._stream.write_int(header)
|
||||
self._stream.stream.write(data)
|
||||
|
||||
def finish(self) -> None:
|
||||
self._stream.write_int(0)
|
||||
|
||||
class TrackDecoder:
|
||||
"""TrackDecoder for track messages."""
|
||||
|
||||
def decode(self, stream: MessageInput):
|
||||
"""Decode an entire message and return the `Track`."""
|
||||
|
||||
body_reader = stream.next()
|
||||
if not body_reader:
|
||||
raise ValueError("empty stream")
|
||||
|
||||
version = body_reader.read_byte()
|
||||
|
||||
return {
|
||||
"title": body_reader.read_utf(),
|
||||
"author": body_reader.read_utf(),
|
||||
"length": body_reader.read_long(),
|
||||
"identifier": body_reader.read_utf(),
|
||||
"is_stream": body_reader.read_bool(),
|
||||
"uri": body_reader.read_optional_utf(),
|
||||
"artworkUrl": None if version not in [0, 3] else body_reader.read_optional_utf(),
|
||||
"isrc": None if version != 3 else body_reader.read_optional_utf(),
|
||||
"sourceName": body_reader.read_utf(),
|
||||
"position": body_reader.read_long()
|
||||
}
|
||||
|
||||
class TrackEncoder:
|
||||
def encode(self, stream: MessageOutput, track: Track) -> None:
|
||||
body_writer = stream.start()
|
||||
|
||||
body_writer.write_byte(0)
|
||||
body_writer.write_utf(track.title)
|
||||
body_writer.write_utf(track.author)
|
||||
body_writer.write_long(track.length)
|
||||
body_writer.write_utf(track.identifier)
|
||||
body_writer.write_bool(track.is_stream)
|
||||
body_writer.write_optional_utf(track.uri)
|
||||
body_writer.write_optional_utf(track.thumbnail)
|
||||
body_writer.write_utf(track.source)
|
||||
body_writer.write_long(0)
|
||||
|
||||
stream.commit()
|
||||
|
||||
def decode(data: Union[str, bytes]) -> dict:
|
||||
decoded = base64.b64decode(data)
|
||||
stream = MessageInput(io.BytesIO(decoded))
|
||||
return TrackDecoder().decode(stream)
|
||||
|
||||
def encode(track) -> bytes:
|
||||
buf = io.BytesIO()
|
||||
stream = MessageOutput(buf)
|
||||
TrackEncoder().encode(stream, track)
|
||||
return base64.b64encode(buf.getvalue()).decode("utf-8")
|
||||
@@ -33,7 +33,7 @@ from function import (
|
||||
time as ctime
|
||||
)
|
||||
|
||||
from .formatter import encode
|
||||
from .transformer import encode
|
||||
|
||||
YOUTUBE_REGEX = re.compile(r'(https?://)?(www\.)?youtube\.(com|nl)/watch\?v=([-\w]+)')
|
||||
|
||||
@@ -108,7 +108,7 @@ class Track:
|
||||
@property
|
||||
def track_id(self) -> str:
|
||||
if not self._track_id:
|
||||
self._track_id = encode(self)
|
||||
self._track_id = encode(self.info)
|
||||
|
||||
return self._track_id
|
||||
|
||||
|
||||
@@ -582,11 +582,11 @@ class Player(VoiceProtocol):
|
||||
|
||||
data = {
|
||||
"encodedTrack": track.track_id,
|
||||
"position": str(start if start else track.position)
|
||||
"position": str(start or 0)
|
||||
}
|
||||
|
||||
if end or track.end_time:
|
||||
data["endTime"] = str(end if end else track.end_time)
|
||||
data["endTime"] = str(end or track.end_time)
|
||||
|
||||
await self.send(method=RequestMethod.PATCH, query=f"noReplace={ignore_if_playing}", data=data)
|
||||
if self._node.yt_ratelimit:
|
||||
|
||||
322
voicelink/transformer.py
Normal file
322
voicelink/transformer.py
Normal file
@@ -0,0 +1,322 @@
|
||||
"""
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2017-present Devoxin
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
"""
|
||||
|
||||
import struct
|
||||
|
||||
from io import BytesIO
|
||||
from base64 import b64decode, b64encode
|
||||
from typing import Any, Callable, Dict, Mapping, Optional, Tuple, Final
|
||||
|
||||
V2_KEYSET = {'title', 'author', 'length', 'identifier', 'isStream', 'uri', 'sourceName', 'position'}
|
||||
V3_KEYSET = V2_KEYSET | {'artworkUrl', 'isrc'}
|
||||
|
||||
class _MissingObj:
|
||||
__slots__ = ()
|
||||
|
||||
def __repr__(self):
|
||||
return '...'
|
||||
|
||||
MISSING: Any = _MissingObj()
|
||||
|
||||
class DataReader:
|
||||
__slots__ = ('_buf', '_mark')
|
||||
|
||||
def __init__(self, base64_str: str):
|
||||
self._buf: Final[BytesIO] = BytesIO(b64decode(base64_str))
|
||||
self._mark: Optional[int] = None
|
||||
|
||||
@property
|
||||
def remaining(self) -> int:
|
||||
return self._buf.getbuffer().nbytes - self._buf.tell()
|
||||
|
||||
def mark(self) -> None:
|
||||
self._mark = self._buf.tell()
|
||||
|
||||
def rewind(self) -> None:
|
||||
if self._mark is None or not isinstance(self._mark, int):
|
||||
raise IOError('Cannot rewind buffer without a marker!')
|
||||
|
||||
if self._mark < 0:
|
||||
raise IOError('Cannot rewind buffer to a negative position!')
|
||||
|
||||
self._buf.seek(self._mark)
|
||||
self._mark = None
|
||||
|
||||
def _read(self, count: int) -> bytes:
|
||||
return self._buf.read(count)
|
||||
|
||||
def read_byte(self) -> bytes:
|
||||
return self._read(1)
|
||||
|
||||
def read_boolean(self) -> bool:
|
||||
result, = struct.unpack('B', self.read_byte())
|
||||
return result != 0
|
||||
|
||||
def read_unsigned_short(self) -> int:
|
||||
result, = struct.unpack('>H', self._read(2))
|
||||
return result
|
||||
|
||||
def read_int(self) -> int:
|
||||
result, = struct.unpack('>i', self._read(4))
|
||||
return result
|
||||
|
||||
def read_long(self) -> int:
|
||||
result, = struct.unpack('>Q', self._read(8))
|
||||
return result
|
||||
|
||||
def read_nullable_utf(self, utfm: bool = False) -> Optional[str]:
|
||||
exists = self.read_boolean()
|
||||
|
||||
if not exists:
|
||||
return None
|
||||
|
||||
return self.read_utfm() if utfm else self.read_utf().decode()
|
||||
|
||||
def read_utf(self) -> bytes:
|
||||
text_length = self.read_unsigned_short()
|
||||
return self._read(text_length)
|
||||
|
||||
def read_utfm(self) -> str:
|
||||
text_length = self.read_unsigned_short()
|
||||
utf_string = self._read(text_length)
|
||||
return read_utfm(text_length, utf_string)
|
||||
|
||||
class DataWriter:
|
||||
__slots__ = ('_buf',)
|
||||
|
||||
def __init__(self):
|
||||
self._buf: Final[BytesIO] = BytesIO()
|
||||
|
||||
def _write(self, data):
|
||||
self._buf.write(data)
|
||||
|
||||
def write_byte(self, byte):
|
||||
self._buf.write(byte)
|
||||
|
||||
def write_boolean(self, boolean: bool):
|
||||
enc = struct.pack('B', 1 if boolean else 0)
|
||||
self.write_byte(enc)
|
||||
|
||||
def write_unsigned_short(self, short: int):
|
||||
enc = struct.pack('>H', short)
|
||||
self._write(enc)
|
||||
|
||||
def write_int(self, integer: int):
|
||||
enc = struct.pack('>i', integer)
|
||||
self._write(enc)
|
||||
|
||||
def write_long(self, long_value: int):
|
||||
enc = struct.pack('>Q', long_value)
|
||||
self._write(enc)
|
||||
|
||||
def write_nullable_utf(self, utf_string: Optional[str]):
|
||||
self.write_boolean(bool(utf_string))
|
||||
|
||||
if utf_string:
|
||||
self.write_utf(utf_string)
|
||||
|
||||
def write_utf(self, utf_string: str):
|
||||
utf = utf_string.encode('utf8')
|
||||
byte_len = len(utf)
|
||||
|
||||
if byte_len > 65535:
|
||||
raise OverflowError('UTF string may not exceed 65535 bytes!')
|
||||
|
||||
self.write_unsigned_short(byte_len)
|
||||
self._write(utf)
|
||||
|
||||
def finish(self) -> bytes:
|
||||
with BytesIO() as track_buf:
|
||||
byte_len = self._buf.getbuffer().nbytes
|
||||
flags = byte_len | (1 << 30)
|
||||
enc_flags = struct.pack('>i', flags)
|
||||
track_buf.write(enc_flags)
|
||||
|
||||
self._buf.seek(0)
|
||||
track_buf.write(self._buf.read())
|
||||
self._buf.close()
|
||||
|
||||
track_buf.seek(0)
|
||||
return track_buf.read()
|
||||
|
||||
def decode_probe_info(reader: DataReader) -> Mapping[str, Any]:
|
||||
probe_info = reader.read_utf().decode()
|
||||
return {'probe_info': probe_info}
|
||||
|
||||
def decode_lavasrc_fields(reader: DataReader) -> Mapping[str, Any]:
|
||||
if reader.remaining <= 8:
|
||||
return {}
|
||||
|
||||
album_name = reader.read_nullable_utf()
|
||||
album_url = reader.read_nullable_utf()
|
||||
artist_url = reader.read_nullable_utf()
|
||||
artist_artwork_url = reader.read_nullable_utf()
|
||||
preview_url = reader.read_nullable_utf()
|
||||
is_preview = reader.read_boolean()
|
||||
|
||||
return {
|
||||
'album_name': album_name,
|
||||
'album_url': album_url,
|
||||
'artist_url': artist_url,
|
||||
'artist_artwork_url': artist_artwork_url,
|
||||
'preview_url': preview_url,
|
||||
'is_preview': is_preview
|
||||
}
|
||||
|
||||
DEFAULT_DECODER_MAPPING: Dict[str, Callable[[DataReader], Mapping[str, Any]]] = {
|
||||
'http': decode_probe_info,
|
||||
'local': decode_probe_info,
|
||||
'deezer': decode_lavasrc_fields,
|
||||
'spotify': decode_lavasrc_fields,
|
||||
'applemusic': decode_lavasrc_fields
|
||||
}
|
||||
|
||||
def read_utfm(utf_len: int, utf_bytes: bytes) -> str:
|
||||
chars = []
|
||||
count = 0
|
||||
|
||||
while count < utf_len:
|
||||
char = utf_bytes[count] & 0xff
|
||||
if char > 127:
|
||||
break
|
||||
|
||||
count += 1
|
||||
chars.append(chr(char))
|
||||
|
||||
while count < utf_len:
|
||||
char = utf_bytes[count] & 0xff
|
||||
shift = char >> 4
|
||||
|
||||
if 0 <= shift <= 7:
|
||||
count += 1
|
||||
chars.append(chr(char))
|
||||
elif 12 <= shift <= 13:
|
||||
count += 2
|
||||
if count > utf_len:
|
||||
raise UnicodeDecodeError('utf8', b'', 0, utf_len, 'malformed input: partial character at end')
|
||||
char2 = utf_bytes[count - 1]
|
||||
if (char2 & 0xC0) != 0x80:
|
||||
raise UnicodeDecodeError('utf8', b'', 0, utf_len, f'malformed input around byte {count}')
|
||||
|
||||
char_shift = ((char & 0x1F) << 6) | (char2 & 0x3F)
|
||||
chars.append(chr(char_shift))
|
||||
elif shift == 14:
|
||||
count += 3
|
||||
if count > utf_len:
|
||||
raise UnicodeDecodeError('utf8', b'', 0, utf_len, 'malformed input: partial character at end')
|
||||
|
||||
char2 = utf_bytes[count - 2]
|
||||
char3 = utf_bytes[count - 1]
|
||||
|
||||
if (char2 & 0xC0) != 0x80 or (char3 & 0xC0) != 0x80:
|
||||
raise UnicodeDecodeError('utf8', b'', 0, utf_len, f'malformed input around byte {(count - 1)}')
|
||||
|
||||
char_shift = ((char & 0x0F) << 12) | ((char2 & 0x3F) << 6) | ((char3 & 0x3F) << 0)
|
||||
chars.append(chr(char_shift))
|
||||
else:
|
||||
raise UnicodeDecodeError('utf8', b'', 0, utf_len, f'malformed input around byte {count}')
|
||||
|
||||
return ''.join(chars).encode('utf-16', 'surrogatepass').decode('utf-16')
|
||||
|
||||
def _read_track_common(reader: DataReader) -> Tuple[str, str, int, str, bool, Optional[str]]:
|
||||
title = reader.read_utfm()
|
||||
author = reader.read_utfm()
|
||||
length = reader.read_long()
|
||||
identifier = reader.read_utf().decode()
|
||||
is_stream = reader.read_boolean()
|
||||
uri = reader.read_nullable_utf()
|
||||
return (title, author, length, identifier, is_stream, uri)
|
||||
|
||||
def _write_track_common(track: Dict[str, Any], writer: DataWriter):
|
||||
writer.write_utf(track['title'])
|
||||
writer.write_utf(track['author'])
|
||||
writer.write_long(track['length'])
|
||||
writer.write_utf(track['identifier'])
|
||||
writer.write_boolean(track['isStream'])
|
||||
writer.write_nullable_utf(track['uri'])
|
||||
|
||||
def decode(
|
||||
track: str,
|
||||
source_decoders: Mapping[str, Callable[[DataReader], Mapping[str, Any]]] = MISSING
|
||||
) -> dict:
|
||||
|
||||
decoders = DEFAULT_DECODER_MAPPING.copy()
|
||||
|
||||
if source_decoders is not MISSING:
|
||||
decoders.update(source_decoders)
|
||||
|
||||
reader = DataReader(track)
|
||||
|
||||
flags = (reader.read_int() & 0xC0000000) >> 30
|
||||
version, = struct.unpack('B', reader.read_byte()) if flags & 1 != 0 else (1,)
|
||||
|
||||
title, author, length, identifier, is_stream, uri = _read_track_common(reader)
|
||||
extra_fields = {}
|
||||
|
||||
if version == 3:
|
||||
extra_fields['artworkUrl'] = reader.read_nullable_utf()
|
||||
extra_fields['isrc'] = reader.read_nullable_utf()
|
||||
|
||||
source = reader.read_utf().decode()
|
||||
source_specific_fields = {}
|
||||
|
||||
if source in decoders:
|
||||
source_specific_fields.update(decoders[source](reader))
|
||||
|
||||
position = reader.read_long()
|
||||
|
||||
return {
|
||||
'title': title,
|
||||
'author': author,
|
||||
'length': length,
|
||||
'identifier': identifier,
|
||||
'isStream': is_stream,
|
||||
'uri': uri,
|
||||
'isSeekable': not is_stream,
|
||||
'sourceName': source,
|
||||
'position': position,
|
||||
**extra_fields
|
||||
}
|
||||
|
||||
def encode(
|
||||
track: Dict[str, Any],
|
||||
source_encoders: Mapping[str, Callable[[DataWriter, Dict[str, Any]], None]] = MISSING
|
||||
) -> str:
|
||||
assert V3_KEYSET <= track.keys()
|
||||
|
||||
writer = DataWriter()
|
||||
version = struct.pack('B', 3)
|
||||
writer.write_byte(version)
|
||||
_write_track_common(track, writer)
|
||||
writer.write_nullable_utf(track['artworkUrl'])
|
||||
writer.write_nullable_utf(track['isrc'])
|
||||
writer.write_utf(track['sourceName'])
|
||||
|
||||
if source_encoders is not MISSING and track['sourceName'] in source_encoders:
|
||||
source_encoders[track['sourceName']](writer, track)
|
||||
|
||||
writer.write_long(track['position'])
|
||||
|
||||
enc = writer.finish()
|
||||
return b64encode(enc).decode()
|
||||
@@ -203,5 +203,4 @@ class Ping:
|
||||
((self._host, self._port), None))
|
||||
s_runtime = 1000 * (cost_time)
|
||||
|
||||
return s_runtime
|
||||
|
||||
return s_runtime
|
||||
Reference in New Issue
Block a user