137 lines
5.9 KiB
Python
137 lines
5.9 KiB
Python
from tbotconfig import *
|
|
|
|
# Telegram things:
|
|
from telegram import InlineQueryResultArticle, InputTextMessageContent, Update
|
|
from telegram.constants import ParseMode
|
|
from telegram.ext import Application, CommandHandler, ContextTypes, InlineQueryHandler
|
|
|
|
# Other needful stuff
|
|
import asyncio
|
|
import html
|
|
import json
|
|
import logging
|
|
import traceback
|
|
from uuid import uuid4
|
|
import urllib.request
|
|
import urllib
|
|
import re
|
|
import spotipy
|
|
import spotipy.oauth2
|
|
from dpath import util
|
|
import requests
|
|
import time
|
|
import os
|
|
from pathlib import Path
|
|
|
|
logging.basicConfig(
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
level=logging.INFO
|
|
)
|
|
|
|
# Actual bot stuffs
|
|
#
|
|
# Define a few command handlers. These usually take the two arguments bot and update. Error handlers also receive the raised TelegramError object in error.
|
|
|
|
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
"""Send a message when the command /start is issued."""
|
|
await context.bot.send_message(
|
|
chat_id=update.effective_chat.id,
|
|
text="'Hello there!\n/help will give a short introduction to this bot.\nIts an inline bot, so you shouldn\'nt really be using it here.\nThis is a bot made by @DailytheNoob, check out the code on daviddaily.dev/david/telegram-music-bot'"
|
|
)
|
|
|
|
|
|
async def help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
"""Send a message when the command /help is issued."""
|
|
await context.bot.update.message.reply_markdown_v2(f"Attention! I am an inline bot only!\n\nStart your message with @MusicServiceBot and then the name of the song you want me to search spotify for. Wait for a few seconds and you should get {num_results} results back.\nIf spotify provides a preview for the song, when you tap the desired result you'll get a 30 second preview for the song, otherwise its just the name of the song. Once you pick the right result you'll get a button for Spotify and more links that goes to a link aggregation site that will have links to most other services.")
|
|
|
|
|
|
async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|
"""Handle the inline query. This is run when you type: @botusername <query>"""
|
|
query = update.inline_query.query
|
|
|
|
if query == "":
|
|
return
|
|
|
|
results = []
|
|
|
|
sp_info = sp_search(query)
|
|
|
|
for i in range(len(sp_info)):
|
|
"""Spotify"""
|
|
sp_title = util.get(sp_info, f"/tracks/items/[{i}]/name")
|
|
sp_artist = util.get(sp_info, f"/tracks/items/[{i}]/artists/[0]/name")
|
|
sp_albname = util.get(sp_info, f"/tracks/items/[{i}]/album/name")
|
|
sp_albdate = util.get(sp_info, f"/tracks/items/[{i}]/album/release_date")
|
|
sp_art = util.get(sp_info, f"/tracks/items/[{i}]/album/images/[2]/url")
|
|
sp_audio = util.get(sp_info, f"/tracks/items/[{i}]/preview_url")
|
|
sp_link = util.get(sp_info, f"/tracks/items/[{i}]/external_urls/spotify")
|
|
songlink = "https://song.link/{0}".format(sp_link)
|
|
reply_markup = InlineKeyboardMarkup([[InlineKeyboardButton("Spotify", url = sp_link), InlineKeyboardButton("More", url = songlink)]])
|
|
description = f"By {sp_artist} on the album {sp_albname}, released {sp_albdate}"
|
|
|
|
if "None" in str(sp_audio):
|
|
message_content = f"Check out \"{sp_title}\" by {sp_artist} on the album \"{sp_albname}\""
|
|
results.append(InlineQueryResultArticle(id = i, title = sp_title, description = description, input_message_content = InputTextMessageContent(message_content), thumb_url = sp_art, reply_markup = reply_markup),)
|
|
else:
|
|
message_content = f"Listen to \"{sp_title}\" by {sp_artist} on the album \"{sp_albname}\""
|
|
results.append(InlineQueryResultAudio(id = i, audio_url = sp_audio, title = sp_title, performer = sp_artist, audio_duration = 30, caption = message_content, reply_markup = reply_markup),)
|
|
|
|
update.inline_query.answer(results)
|
|
|
|
|
|
async def sp_search(query):
|
|
return sp.search(q=query, limit=num_results)
|
|
|
|
|
|
async def error_handler(update) -> None:
|
|
"""Log the error and send a telegram message to notify the developer."""
|
|
# Log the error before we do anything else, so we can see it even if something breaks.
|
|
logger.error(msg="Exception while handling an update:", exc_info=context.error)
|
|
|
|
# This can be your own ID, or one for a developer group/channel.
|
|
# You can use the /start command of this bot to see your chat id.
|
|
DEVELOPER_CHAT_ID = 175042676
|
|
# traceback.format_exception returns the usual python message about an exception, but as a
|
|
# list of strings rather than a single string, so we have to join them together.
|
|
tb_list = traceback.format_exception(None, context.error, context.error.__traceback__)
|
|
tb_string = ''.join(tb_list)
|
|
|
|
# Build the message with some markup and additional information about what happened.
|
|
# You might need to add some logic to deal with messages longer than the 4096 character limit.
|
|
message = (
|
|
f'An exception was raised while handling an update\n'
|
|
f'<pre>update = {html.escape(json.dumps(update.to_dict(), indent=2, ensure_ascii=False))}'
|
|
'</pre>\n\n'
|
|
f'<pre>context.chat_data = {html.escape(str(context.chat_data))}</pre>\n\n'
|
|
f'<pre>context.user_data = {html.escape(str(context.user_data))}</pre>\n\n'
|
|
f'<pre>{html.escape(tb_string)}</pre>'
|
|
)
|
|
|
|
# Finally, send the message
|
|
context.bot.send_message(chat_id=DEVELOPER_CHAT_ID, text=message, parse_mode=ParseMode.HTML)
|
|
|
|
|
|
def main() -> None:
|
|
#Configure how many results to fetch
|
|
num_results = 5
|
|
|
|
# Spotify:
|
|
spotify_credentials = spotipy.oauth2.SpotifyClientCredentials(SPOTIPY_CLIENT_ID, SPOTIPY_CLIENT_SECRET)
|
|
sp = spotipy.Spotify(client_credentials_manager=spotify_credentials)
|
|
|
|
# Create the Application and pass it your bot's token.
|
|
application = Application.builder().token(BOT_TOKEN).build()
|
|
|
|
# on different commands - answer in Telegram
|
|
application.add_handler(CommandHandler("start", start))
|
|
application.add_handler(CommandHandler("help", help))
|
|
|
|
# on non command i.e message - echo the message on Telegram
|
|
application.add_handler(InlineQueryHandler(inline_query))
|
|
|
|
# Run the bot until the user presses Ctrl-C
|
|
application.run_polling()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |