Python

Spotify Telegram Bio and Profile

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

Приветствие (Telegram Business): В заголовок ставится название трека, а в описание - исполнитель.
О себе (Bio): Туда записывается строка формата "Исполнитель - Трек".
3. Сохранение музыки
Это самая сложная часть, так как Spotify не позволяет скачивать MP3-файлы напрямую. Скрипт делает:
Берет точное название трека и автора из Spotify.
Находит это аудио на YouTube (через yt-dlp) и скачивает звуковую дорожку.
Берет качественную обложку из Spotify и "приклеивает" её к файлу.
Загружает готовый MP3 в твои "Сохраненные сообщения" и через API закрепляет его в раздел "Музыка" твоего профиля.​

​

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

​

Сode

​

import time
import os
import sys
import requests
import spotipy
from spotipy.oauth2 import SpotifyOAuth
from pyrogram import Client as PyroClient
from pyrogram.raw import functions, types
import yt_dlp

API_ID = 1
API_HASH = "1"
SESSION_NAME = "session_music_bot"

SPOTIPY_CLIENT_ID = '1'
SPOTIPY_CLIENT_SECRET = '1'
SPOTIPY_REDIRECT_URI = 'http://127.0.0.1:8888/callback'

scope = "user-read-currently-playing"
sp = spotipy.Spotify(auth_manager=SpotifyOAuth(client_id=SPOTIPY_CLIENT_ID,
client_secret=SPOTIPY_CLIENT_SECRET,
redirect_uri=SPOTIPY_REDIRECT_URI,
scope=scope))

app = PyroClient(SESSION_NAME, api_id=API_ID, api_hash=API_HASH)

def get_current_track():
try:
current = sp.current_user_playing_track()
if not current or not current['is_playing']:
return None

item = current['item']
if not item: return None

track_data = {
'id': item['id'],
'title': item['name'],
'artists': [artist['name'] for artist in item['artists']],
'cover_url': item['album']['images'][0]['url'] if item['album']['images'] else None,
'duration_ms': item['duration_ms']
}
return track_data
except Exception as e:
print(f"Spotify Error: {e}")
return None

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

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 download_from_youtube(query, filename):
name_no_ext = filename.replace('.mp3', '')

ydl_opts = {
'format': 'bestaudio/best',
'outtmpl': name_no_ext,
'quiet': True,
'noplaylist': True,
'extractor_args': {'youtube': {'player_client': ['android', 'web']}},
'postprocessors': [{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '192',
}],
}

try:
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([f"ytsearch1:{query}"])

final_name = name_no_ext + ".mp3"
if os.path.exists(final_name):
return final_name
return None
except Exception as e:
print(f"Download error: {e}")
return None


def save_music(track):
mp3_filename = f"{track['id']}.mp3"
cover_filename = f"{track['id']}.jpg"

try:
print(f"Downloading: {track['title']}")

if track['cover_url']:
img_data = requests.get(track['cover_url']).content
with open(cover_filename, 'wb') as handler:
handler.write(img_data)

search_query = f"{', '.join(track['artists'])} - {track['title']} audio"
downloaded_file = download_from_youtube(search_query, mp3_filename)

if not downloaded_file:
print("Could not download audio")
return

artists_str = ', '.join(track['artists'])
thumb = cover_filename if os.path.exists(cover_filename) else None

msg = app.send_audio(
chat_id="me",
audio=downloaded_file,
thumb=thumb,
performer=artists_str,
title=track['title'],
caption=f"{artists_str} - {track['title']} (via Spotify)"
)

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_filename): os.remove(mp3_filename)
if os.path.exists(cover_filename): os.remove(cover_filename)

def main():
print("Started monitoring Spotify")
with app:
last_id = None
while True:
try:
track = get_current_track()
if track:
if track['id'] != last_id:
print(f"New track: {track['title']}")
update_profile(track)
save_music(track)
last_id = track['id']
else:
pass

time.sleep(15)
except KeyboardInterrupt:
break
except Exception as e:
print(f"Global error: {e}")
time.sleep(15)

if __name__ == '__main__':
main()

Python

import time

import os

import sys

import requests

import spotipy

from spotipy.oauth2 import SpotifyOAuth

from pyrogram import Client as PyroClient

from pyrogram.raw import functions, types

import yt_dlp



API_ID = 1

API_HASH = "1"

SESSION_NAME = "session_music_bot"



SPOTIPY_CLIENT_ID = '1'

SPOTIPY_CLIENT_SECRET = '1'

SPOTIPY_REDIRECT_URI = 'http://127.0.0.1:8888/callback'



scope = "user-read-currently-playing"

sp = spotipy.Spotify(auth_manager=SpotifyOAuth(client_id=SPOTIPY_CLIENT_ID,

                                               client_secret=SPOTIPY_CLIENT_SECRET,

                                               redirect_uri=SPOTIPY_REDIRECT_URI,

                                               scope=scope))



app = PyroClient(SESSION_NAME, api_id=API_ID, api_hash=API_HASH)



def get_current_track():

    try:

        current = sp.current_user_playing_track()

        if not current or not current['is_playing']:

            return None

       

        item = current['item']

        if not item: return None



        track_data = {

            'id': item['id'],

            'title': item['name'],

            'artists': [artist['name'] for artist in item['artists']],

            'cover_url': item['album']['images'][0]['url'] if item['album']['images'] else None,

            'duration_ms': item['duration_ms']

        }

        return track_data

    except Exception as e:

        print(f"Spotify Error: {e}")

        return None



def update_profile(track):

    title = track['title']

    artists = ', '.join(track['artists'])

   

    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 download_from_youtube(query, filename):

    name_no_ext = filename.replace('.mp3', '')

   

    ydl_opts = {

        'format': 'bestaudio/best',

        'outtmpl': name_no_ext,

        'quiet': True,

        'noplaylist': True,

        'extractor_args': {'youtube': {'player_client': ['android', 'web']}},

        'postprocessors': [{

            'key': 'FFmpegExtractAudio',

            'preferredcodec': 'mp3',

            'preferredquality': '192',

        }],

    }

   

    try:

        with yt_dlp.YoutubeDL(ydl_opts) as ydl:

            ydl.download([f"ytsearch1:{query}"])

       

        final_name = name_no_ext + ".mp3"

        if os.path.exists(final_name):

            return final_name

        return None

    except Exception as e:

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

        return None





def save_music(track):

    mp3_filename = f"{track['id']}.mp3"

    cover_filename = f"{track['id']}.jpg"

   

    try:

        print(f"Downloading: {track['title']}")

       

        if track['cover_url']:

            img_data = requests.get(track['cover_url']).content

            with open(cover_filename, 'wb') as handler:

                handler.write(img_data)



        search_query = f"{', '.join(track['artists'])} - {track['title']} audio"

        downloaded_file = download_from_youtube(search_query, mp3_filename)



        if not downloaded_file:

            print("Could not download audio")

            return



        artists_str = ', '.join(track['artists'])

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



        msg = app.send_audio(

            chat_id="me",

            audio=downloaded_file,

            thumb=thumb,

            performer=artists_str,

            title=track['title'],

            caption=f"{artists_str} - {track['title']} (via Spotify)"

        )



        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_filename): os.remove(mp3_filename)

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



def main():

    print("Started monitoring Spotify")

    with app:

        last_id = None

        while True:

            try:

                track = get_current_track()

                if track:

                    if track['id'] != last_id:

                        print(f"New track: {track['title']}")

                        update_profile(track)

                        save_music(track)

                        last_id = track['id']

                else:

                    pass

               

                time.sleep(15)

            except KeyboardInterrupt:

                break

            except Exception as e:

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

                time.sleep(15)



if __name__ == '__main__':

main()