MarshalX
commented
1 year ago код с пастбина:
from datetime import datetime
from typing import List
import json
from yandex_music import ClientAsync, Sequence, Track
class DescriptionSeed:
def __init__(self, value: str, tag: str, type: str, **kwargs):
self.value = value
self.tag = tag
self.type = type
def get_full_name(self, separator=':'):
return f'{self.type}{separator}{self.tag}'
def get_id_from(self):
return f'radio-mobile-{self.get_full_name("-")}-default'
class StationSession:
def __init__(self, radio_session_id: str, batch_id: str, pumpkin: bool, description_seed: DescriptionSeed, accepted_seeds: List[DescriptionSeed], **kwargs):
self.radio_session_id = radio_session_id
self.batch_id = batch_id
self.pumpkin = pumpkin
# Костыльно, но мне лень делать по другому :)
self.description_seed = DescriptionSeed(**description_seed)
self.accepted_seeds = [DescriptionSeed(**seed) for seed in accepted_seeds]
class PlaybackStatistics:
def __init__(self, total_played_seconds: float, skipped: bool) -> None:
self.total_played_seconds = total_played_seconds
self.skipped = skipped
class Station:
def __init__(self, client: ClientAsync, seeds: str | List[str]):
'''
Attributes:
seed: str | List[str]
В API указывается как seeds в массиве, но не смог найти условие, когда в длинна seeds не 1. Скорее всего это позволяет смешивать различные станции
Пример seed: 'track:{track_id}', 'user:onyourwave'
'''
self.client = client
self.seeds = [seeds] if isinstance(seeds, str) else seeds
self.current_track_number = -1
self.current_track_id = ''
def __get_rotor_link(self, path) -> str:
# Получение URL с сессией
return f'{self.client.base_url}/rotor/session/{self.session_info.radio_session_id}{path}'
async def __load_new_sequence(self):
'''
Загрузка новой последовательности треков.
В queue ОБЯЗАТЕЛЬНО долен быть первый трек из последовательности
'''
self.sequence = Sequence.de_list((await self.client.request.post(self.__get_rotor_link('/tracks'), json={
"queue": [
self.sequence[0].track.id
]
}))['sequence'], self.client)
def __get_current_timestamp(self) -> str:
# Конвертирует время в формате UTC. Пример: "2023-05-01T10:35:14.604531+07:00"
return datetime.now().astimezone().strftime('%Y-%m-%dT%H:%M:%S.%f%Z:00')
async def __send_feedback(self, type: str, **kwargs):
'''
Метод для отправки связи
Аргументы:
**kwargs: dict - параметры запроса. указаны после типов запросов
type: str
radioStarted: отправлять перед запуском станции ОДИН РАЗ
from: str - id станции. Пример: user:onyourwave
trackStarted: начало каждого трека
trackId: str - id трека
skip: пропуск трека
trackId: str - id трека
totalPlayedSeconds: float - количество секунд
trackFinished: отправлять при завершении трека
trackId: str - id трека
totalPlayedSeconds: float - количество секунд. В приложении ставится 0.1. Не знаю, обязательно ли это нужно
'''
await self.client.request.post(self.__get_rotor_link('/feedback'), json={
'event': {
'type': type,
'timestamp': self.__get_current_timestamp(),
**kwargs
},
'batchId': self.session_info.batch_id
})
async def new_session(self):
session = await self.client.request.post(f'{self.client.base_url}/rotor/session/new', json = {
'seeds': self.seeds,
'includeTracksInResponse': True
})
self.session_info = StationSession(**session)
self.sequence = Sequence.de_list(session['sequence'], self.client)
await self.__send_feedback('radioStarted', **{
'from': self.session_info.description_seed.get_id_from()
})
def set_playback_statistics(self, playback_statistics: PlaybackStatistics):
self.playback_statistics = playback_statistics
def __get_current_track(self) -> Track | None:
return self.sequence[self.current_track_number].track if 0 <= self.current_track_number < len(self.sequence) else None
async def next_track(self) -> Track:
# Заканчиваем трек, если был
if self.current_track_number != -1:
await self.__send_feedback('skip' if self.playback_statistics.skipped else 'trackFinished', **{
'trackId': self.current_track_id,
'totalPlayedSeconds': self.playback_statistics.total_played_seconds if self.playback_statistics.skipped else 0.1
})
# Получаем следующий трек
self.current_track_number += 1
track = self.__get_current_track()
if track is None:
self.current_track_number = 0
await self.__load_new_sequence()
track = self.__get_current_track()
self.current_track_id = track.id
# Запуск трека
await self.__send_feedback('trackStarted', **{
'trackId': self.current_track_id,
})
return track
async def example(token: str):
client = ClientAsync(token)
await client.init()
station = Station(client, "user:onyourwave")
await station.new_session()
while True:
track = await station.next_track()
print(', '.join([artist.name for artist in track.artists]), '-', track.title)
station.set_playback_statistics(PlaybackStatistics(
total_played_seconds=float(s) if (s := input('total_played_seconds: ')).replace('.', '', 1).isdigit() else 0.0,
skipped=input('skipped? [Y/n]: ') in ['y', 'Y', '']
))
KirMozor
Для работы станции надо:
Реализация класса Station с использованием новых методов API: https://pastebin.com/dJaHQmTp