Add YTM, newer and better error handling

This commit is contained in:
David Daily 2021-01-14 14:59:07 -06:00
parent 78556ef6d2
commit fe29eb47e7
2 changed files with 78 additions and 45 deletions

View File

@ -13,6 +13,7 @@ dpath = "*"
gmusicapi = "*" gmusicapi = "*"
pathlib = "*" pathlib = "*"
uuid = "*" uuid = "*"
ytmusicapi = "*"
[requires] [requires]
python_version = "3.6" python_version = "3.6"

118
music.py
View File

@ -1,12 +1,14 @@
import logging
from tbotconfig import * from tbotconfig import *
# Telegram things: # Telegram things:
#from telegram import InlineKeyboardButton, InlineKeyboardMarkup from telegram import Update, ParseMode, InlineKeyboardMarkup, InlineKeyboardButton, InlineQueryResultArticle, InputTextMessageContent, InlineQueryResult, InlineQueryResultAudio
from telegram import InlineKeyboardMarkup, InlineKeyboardButton, InlineQueryResultArticle, InputTextMessageContent, InlineQueryResult, InlineQueryResultAudio
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackQueryHandler, ConversationHandler, InlineQueryHandler, ChosenInlineResultHandler, CallbackContext from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackQueryHandler, ConversationHandler, InlineQueryHandler, ChosenInlineResultHandler, CallbackContext
# Other needful stuff # Other needful stuff
import html
import json
import logging
import traceback
from uuid import uuid4 from uuid import uuid4
import urllib.request import urllib.request
import urllib import urllib
@ -14,29 +16,22 @@ import re
import spotipy import spotipy
import spotipy.oauth2 import spotipy.oauth2
import json import json
import dpath from dpath import util
import requests import requests
import time import time
import os import os
from pathlib import Path from pathlib import Path
from ytmusicapi import YTMusic
ytmusic = YTMusic()
#Configure how many results to fetch #Configure how many results to fetch
num_results = 5 num_results = 5
# Enable logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
logger = logging.getLogger(__name__)
# Spotify: # Spotify:
spotify_credentials = spotipy.oauth2.SpotifyClientCredentials(SPOTIPY_CLIENT_ID, SPOTIPY_CLIENT_SECRET) spotify_credentials = spotipy.oauth2.SpotifyClientCredentials(SPOTIPY_CLIENT_ID, SPOTIPY_CLIENT_SECRET)
sp = spotipy.Spotify(client_credentials_manager=spotify_credentials) sp = spotipy.Spotify(client_credentials_manager=spotify_credentials)
#
# Actual bot stuffs # 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. # Define a few command handlers. These usually take the two arguments bot and update. Error handlers also receive the raised TelegramError object in error.
@ -56,26 +51,23 @@ def inlinequery(update, context: CallbackContext):
spotify_credentials = spotipy.oauth2.SpotifyClientCredentials(SPOTIPY_CLIENT_ID, SPOTIPY_CLIENT_SECRET) spotify_credentials = spotipy.oauth2.SpotifyClientCredentials(SPOTIPY_CLIENT_ID, SPOTIPY_CLIENT_SECRET)
sp = spotipy.Spotify(client_credentials_manager=spotify_credentials) sp = spotipy.Spotify(client_credentials_manager=spotify_credentials)
if query: if query:
sp_info = sp.search(q=query, limit=num_results) sp_info = sp.search(q=query, limit=num_results)
for i in range(num_results): for i in range(num_results):
try: sp_title = util.get(sp_info, f"/tracks/items/[{i}]/name")
dpath.util.get(sp_info, f"/tracks/items/[{i}]/name") sp_artist = util.get(sp_info, f"/tracks/items/[{i}]/artists/[0]/name")
except: sp_albname = util.get(sp_info, f"/tracks/items/[{i}]/album/name")
continue sp_albdate = util.get(sp_info, f"/tracks/items/[{i}]/album/release_date")
else: sp_art = util.get(sp_info, f"/tracks/items/[{i}]/album/images/[2]/url")
sp_title = dpath.util.get(sp_info, f"/tracks/items/[{i}]/name") 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")
sp_artist = dpath.util.get(sp_info, f"/tracks/items/[{i}]/artists/[0]/name")
sp_albname = dpath.util.get(sp_info, f"/tracks/items/[{i}]/album/name")
sp_albdate = dpath.util.get(sp_info, f"/tracks/items/[{i}]/album/release_date")
sp_art = dpath.util.get(sp_info, f"/tracks/items/[{i}]/album/images/[2]/url")
sp_audio = dpath.util.get(sp_info, f"/tracks/items/[{i}]/preview_url")
sp_link = dpath.util.get(sp_info, f"/tracks/items/[{i}]/external_urls/spotify")
songlink = "https://song.link/{0}".format(sp_link) songlink = "https://song.link/{0}".format(sp_link)
yt_q = f"{sp_artist} - {sp_title}"
reply_markup = InlineKeyboardMarkup([[InlineKeyboardButton("Spotify", url = sp_link), InlineKeyboardButton("More", url = songlink)]]) yt_res = ytmusic.search(yt_q, 'songs', limit=1)
yt_id = util.get(yt_res, f"[1]/videoId")
yt_link = f"https://music.youtube.com/watch?v={yt_id}"
reply_markup = InlineKeyboardMarkup([[InlineKeyboardButton("YouTube", url = yt_link), InlineKeyboardButton("Spotify", url = sp_link), InlineKeyboardButton("More", url = songlink)]])
description = f"By {sp_artist} on the album {sp_albname}, released {sp_albdate}" description = f"By {sp_artist} on the album {sp_albname}, released {sp_albdate}"
@ -88,34 +80,74 @@ def inlinequery(update, context: CallbackContext):
update.inline_query.answer(results) update.inline_query.answer(results)
def error(update, context: CallbackContext):
# Log Errors caused by Updates
logger.warning(f"\nUpdate {update} caused error {context.error}\n")
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
)
logger = logging.getLogger(__name__)
# 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
def error_handler(update: Update, context: CallbackContext) -> 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)
# 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 start(update: Update, context: CallbackContext) -> None:
update.effective_message.reply_html(
'Use /bad_command to cause an error.\n'
f'Your chat id is <code>{update.effective_chat.id}</code>.'
)
def main(): def main():
# Create the Updater and pass it your bot's token. # Create the Updater and pass it your bot's token.
# Make sure to set use_context=True to use the new context based callbacks # Make sure to set use_context=True to use the new context based callbacks
# Post version 12 this will no longer be necessary # Post version 12 this will no longer be necessary
updater = Updater(TBOT_TOKEN, use_context=True) updater = Updater(BOT_TOKEN, use_context=True)
# Get the dispatcher to register handlers # Get the dispatcher to register handlers
bot = updater.dispatcher dispatcher = updater.dispatcher
# on different commands - answer in Telegram # Register the commands...
bot.add_handler(CommandHandler("start", start)) dispatcher.add_handler(CommandHandler('start', start))
bot.add_handler(CommandHandler("help", help)) dispatcher.add_handler(CommandHandler("help", help))
dispatcher.add_handler(InlineQueryHandler(inlinequery))
# on noncommand i.e message - echo the message on Telegram # ...and the error handler
bot.add_handler(InlineQueryHandler(inlinequery)) dispatcher.add_error_handler(error_handler)
# log all errors
bot.add_error_handler(error)
# Start the Bot # Start the Bot
updater.start_polling() updater.start_polling()
# Block until the user presses Ctrl-C or the process receives SIGINT, # Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since # SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully. # start_polling() is non-blocking and will stop the bot gracefully.
updater.idle() updater.idle()