Python

Yandex Music Telegram Bio and Profile

Этот скрипт автоматическая связка между Яндекс.Музыкой и профилем Telegram. Он работает в фоновом режиме и выполняет три основные задачи при смене трека:
1. Мониторинг
Код каждые 15 секунд проверяет состояние очереди воспроизведения в Яндексе. Если обнаруживается новая композиция, запускается процесс обновления.
2. Обновление внешнего вида профиля
Скрипт изменяет два поля в твоем аккаунте:
Приветствие (Telegram Business): В заголовок ставится название трека, а в описание - исполнитель.

О себе (Bio): Туда записывается строка формата "Исполнитель - Трек".

3. Сохранение музыки в профиль
Это технически самая сложная часть. Скрипт скачивает аудиофайл и обложку альбома с серверов Яндекса, загружает их в "Избранное" Telegram, а затем использует специальный API-метод, чтобы закрепить этот файл в раздел "Музыка" твоего профиля.
В итоге твой профиль всегда отображает актуальный статус, а прослушанные треки автоматически собираются в плейлист на странице аккаунта.
​

Как выглядит со стороны

Code

import time
import os
import sys
from yandex_music import Client
from pyrogram import Client as PyroClient
from pyrogram.raw import functions, types

YANDEX_TOKEN = '1'
API_ID = 1
API_HASH = "1"

ym = Client(YANDEX_TOKEN).init()
app = PyroClient("session_music_bot", api_id=API_ID, api_hash=API_HASH)

def get_track():
try:
queues = ym.queues_list()
if not queues: return None
queues.sort(key=lambda x: x.modified, reverse=True)
for q in queues:
try:
curr = ym.queue(q.id)
if curr.current_index is None: continue
track_id = curr.get_current_track()
if not track_id: continue
return track_id.fetch_track()
except: continue
return None
except: return None

def update_profile(track):
title = track.title
artists = ', '.join(track.artists_name())

try:
t_intro = title[:25] + "..." if len(title) > 28 else title
d_intro = artists[:70]

app.invoke(
functions.account.UpdateBusinessIntro(
intro=types.InputBusinessIntro(
title=t_intro,
description=d_intro,
sticker=None
)
)
)
print(f"intro updated: {t_intro}")
except Exception as e:
print(f"intro error: {e}")

try:
bio = f"{artists} - {title}"
if len(bio) > 70: bio = bio[:67] + "..."
app.update_profile(bio=bio)
except: pass

def save_music(track):
mp3 = f"{track.id}.mp3"
cover = f"{track.id}.jpg"

try:
print(f"downloading: {track.title}")
track.download(mp3, codec='mp3', bitrate_in_kbps=192)
track.download_cover(cover, size='400x400')

thumb = cover if os.path.exists(cover) else None
artists = ', '.join(track.artists_name())

msg = app.send_audio(
chat_id="me",
audio=mp3,
thumb=thumb,
performer=artists,
title=track.title,
caption=f"{artists} - {track.title}"
)

media = app.invoke(
functions.messages.GetMessages(
id=[types.InputMessageID(id=msg.id)]
)
)

msg_obj = media.messages[0]
if hasattr(msg_obj, 'media') and hasattr(msg_obj.media, 'document'):
doc = msg_obj.media.document
input_doc = types.InputDocument(
id=doc.id,
access_hash=doc.access_hash,
file_reference=doc.file_reference
)

try:
app.invoke(functions.account.SaveMusic(id=input_doc))
print("saved to profile music")
except Exception as e:
print(f"save_music api error: {e}")

except Exception as e:
print(f"download/upload error: {e}")
finally:
if os.path.exists(mp3): os.remove(mp3)
if os.path.exists(cover): os.remove(cover)

def main():
print("started monitoring")
with app:
last_id = None
while True:
try:
track = get_track()
if track:
if track.id != last_id:
print(f"new track: {track.title}")
update_profile(track)
save_music(track)
last_id = track.id
time.sleep(15)
except KeyboardInterrupt:
break
except Exception as e:
print(f"error: {e}")
time.sleep(15)

if __name__ == '__main__':
main()

Python

import time

import os

import sys

from yandex_music import Client

from pyrogram import Client as PyroClient

from pyrogram.raw import functions, types



YANDEX_TOKEN = '1'

API_ID = 1

API_HASH = "1"



ym = Client(YANDEX_TOKEN).init()

app = PyroClient("session_music_bot", api_id=API_ID, api_hash=API_HASH)



def get_track():

    try:

        queues = ym.queues_list()

        if not queues: return None

        queues.sort(key=lambda x: x.modified, reverse=True)

        for q in queues:

            try:

                curr = ym.queue(q.id)

                if curr.current_index is None: continue

                track_id = curr.get_current_track()

                if not track_id: continue

                return track_id.fetch_track()

            except: continue

        return None

    except: return None



def update_profile(track):

    title = track.title

    artists = ', '.join(track.artists_name())

   

    try:

        t_intro = title[:25] + "..." if len(title) > 28 else title

        d_intro = artists[:70]

       

        app.invoke(

            functions.account.UpdateBusinessIntro(

                intro=types.InputBusinessIntro(

                    title=t_intro,

                    description=d_intro,

                    sticker=None

                )

            )

        )

        print(f"intro updated: {t_intro}")

    except Exception as e:

        print(f"intro error: {e}")



    try:

        bio = f"{artists} - {title}"

        if len(bio) > 70: bio = bio[:67] + "..."

        app.update_profile(bio=bio)

    except: pass



def save_music(track):

    mp3 = f"{track.id}.mp3"

    cover = f"{track.id}.jpg"

   

    try:

        print(f"downloading: {track.title}")

        track.download(mp3, codec='mp3', bitrate_in_kbps=192)

        track.download_cover(cover, size='400x400')

       

        thumb = cover if os.path.exists(cover) else None

        artists = ', '.join(track.artists_name())



        msg = app.send_audio(

            chat_id="me",

            audio=mp3,

            thumb=thumb,

            performer=artists,

            title=track.title,

            caption=f"{artists} - {track.title}"

        )



        media = app.invoke(

            functions.messages.GetMessages(

                id=[types.InputMessageID(id=msg.id)]

            )

        )

       

        msg_obj = media.messages[0]

        if hasattr(msg_obj, 'media') and hasattr(msg_obj.media, 'document'):

            doc = msg_obj.media.document

            input_doc = types.InputDocument(

                id=doc.id,

                access_hash=doc.access_hash,

                file_reference=doc.file_reference

            )

           

            try:

                app.invoke(functions.account.SaveMusic(id=input_doc))

                print("saved to profile music")

            except Exception as e:

                print(f"save_music api error: {e}")



    except Exception as e:

        print(f"download/upload error: {e}")

    finally:

        if os.path.exists(mp3): os.remove(mp3)

        if os.path.exists(cover): os.remove(cover)



def main():

    print("started monitoring")

    with app:

        last_id = None

        while True:

            try:

                track = get_track()

                if track:

                    if track.id != last_id:

                        print(f"new track: {track.title}")

                        update_profile(track)

                        save_music(track)

                        last_id = track.id

                time.sleep(15)

            except KeyboardInterrupt:

                break

            except Exception as e:

                print(f"error: {e}")

                time.sleep(15)



if __name__ == '__main__':

main()