emily/lidarr-dl
1
0
Fork 0
Code Issues Pull requests Activity Actions

Merge pull request #251 from hockeygoalie35/alt-notify

Browse source 
ARLChecker 2.0 - ntfy and Pushover support
This commit is contained in:
RandomNinjaAtk 2024-04-29 17:52:14 +00:00 committed by GitHub
commit f39d1c0540
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 111 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lidarr/ARLChecker
View file

@ -1,6 +1,6 @@
#!/usr/bin/with-contenv bash
### Default values
scriptVersion="1.6"
scriptVersion="2.0"
scriptName="ARLChecker"
sleepInterval='24h'
### Import Settings

8
lidarr/extended.conf
View file

@ -76,9 +76,15 @@ plexUrl="" # ONLY used if PlexNotify.bash is used, example: http://x.x.x.x:324
plexToken="" # ONLY used if PlexNotify.bash is used
##### DEEZER ARLCHECKER
arlUpdateInterval='24h' # Interval to check ARL Validity (default 24 hours). Reboot container after changing. s = seconds, m = minutes, h = hours, d = days
arlUpdateInterval="24h" # Interval to check ARL Validity (default 24 hours). Reboot container after changing. s = seconds, m = minutes, h = hours, d = days
telegramBotEnable="false" # Enable/Disable Telegram Bot to notify if ARL expires. Otherwise check text file in custom-services.d/python for status.
telegramBotToken="" # Get token from BotFather during bot creation. If you use a notify channel for Lidarr, you can probably use the same bot, as this script only takes temporary control.
telegramUserChatID="" # Get your userid by chatting: t.me/userinfobot
pushoverEnable="false" # Pushover Notification Support
pushoverUserKey=""
pushoverAppAPIKey=""
ntfyEnable="false" # ntfy Notification Support
ntfyServerTopic="" # Put your server + topic address ex: https://ntfy.mydomain.com/lidarr
ntfyUserToken="" # create a new user token using the ntfy cli commands. See ntfy documentation on how to do that.

140
lidarr/python/ARLChecker.py
View file

@ -1,4 +1,5 @@
import re
import requests
from dataclasses import dataclass
from requests import Session
from argparse import ArgumentParser
@ -80,7 +81,7 @@ class DeezerPlatformProvider:
)
res.raise_for_status()
except Exception as error:
self.log.error(Fore.RED + 'Could not connect! Service down, API changed, wrong credentials or code-related issue.' + Fore.LIGHTWHITE_EX)
self.log.error(Fore.RED + 'Could not connect! Service down, API changed, wrong credentials or code-related issue.' + Fore.RESET)
raise ConnectionError()
self.session.cookies.clear()
@ -88,11 +89,11 @@ class DeezerPlatformProvider:
try:
res = res.json()
except Exception as error:
self.log.error(Fore.RED + "Could not parse JSON response from DEEZER!" + Fore.LIGHTWHITE_EX)
self.log.error(Fore.RED + "Could not parse JSON response from DEEZER!" + Fore.RESET)
raise ParseError()
if 'error' in res and res['error']:
self.log.error(Fore.RED + "Deezer returned the following error:{}".format(res["error"]) + Fore.LIGHTWHITE_EX)
self.log.error(Fore.RED + "Deezer returned the following error:{}".format(res["error"]) + Fore.RESET)
raise ServiceError()
res = res['results']
@ -124,9 +125,15 @@ class LidarrExtendedAPI:
self.telegram_bot_running = False
self.telegram_bot_token = None
self.telegram_user_chat_id = None
self.telegramBotEnableLineText = None
self.telegramBotEnableLineIndex = None
self.telegram_bot_enable_line_text = None
self.telegram_bot_enable_line_index = None
self.bot = None
self.enable_pushover_notify = False
self.pushover_user_key = None
self.pushover_app_api_key = None
self.enable_ntfy_notify = False
self.ntfy_sever_topic = None
self.ntfy_user_token = None
def parse_extended_conf(self):
self.currentARLToken = None
@ -163,27 +170,48 @@ class LidarrExtendedAPI:
for line in self.fileText:
if 'telegramBotEnable=' in line:
self.telegramBotEnableLineText = line
self.telegramBotEnableLineIndex = self.fileText.index(self.telegramBotEnableLineText)
self.telegram_bot_enable_line_text = line
self.telegram_bot_enable_line_index = self.fileText.index(self.telegram_bot_enable_line_text)
self.enable_telegram_bot = re.search(re_search_pattern, line)[0].replace('"', '').lower() in 'true'
if 'telegramBotToken=' in line:
self.telegram_bot_token = re.search(re_search_pattern, line)[0].replace('"', '')
if 'telegramUserChatID=' in line:
self.telegram_user_chat_id = re.search(re_search_pattern, line)[0].replace('"', '')
if 'pushoverEnable=' in line:
self.enable_pushover_notify = re.search(re_search_pattern, line)[0].replace('"', '').lower() in 'true'
if 'pushoverUserKey=' in line:
self.pushover_user_key = re.search(re_search_pattern, line)[0].replace('"', '')
if 'pushoverAppAPIKey=' in line:
self.pushover_app_api_key = re.search(re_search_pattern, line)[0].replace('"', '')
if 'ntfyEnable=' in line:
self.enable_ntfy_notify = re.search(re_search_pattern, line)[0].replace('"', '').lower() in 'true'
if 'ntfyServerTopic=' in line:
self.ntfy_sever_topic = re.search(re_search_pattern, line)[0].replace('"', '')
if 'ntfyUserToken=' in line:
self.ntfy_user_token = re.search(re_search_pattern, line)[0].replace('"', '')
if self.enable_telegram_bot:
self.log.info('Telegram bot is enabled.')
self.log.info(Fore.GREEN+'Telegram bot is enabled.'+Fore.RESET)
if self.telegram_bot_token is None or self.telegram_user_chat_id is None:
self.log.error('Telegram bot token or user chat ID not set in extended.conf. Exiting')
exit(1)
else:
self.log.info('Telegram bot is disabled. Set the flag in extended.conf to enable.')
self.log.info('Telegram bot is disabled.')
# Report Notify/Bot Enable
if self.enable_pushover_notify:
self.log.info(Fore.GREEN+'Pushover notify is enabled.'+Fore.RESET)
else:
self.log.info('Pushover notify is disabled.')
if self.enable_ntfy_notify:
self.log.info(Fore.GREEN+'ntfy notify is enabled.'+Fore.RESET)
else:
self.log.info('ntfy notify is disabled.')
def check_token_wrapper(self): # adds Lidarr_extended specific logging and actions around check_token
self.log.info("Checking ARL Token from extended.conf")
if self.currentARLToken == '""':
self.log.info(Fore.YELLOW+"No ARL Token set in Extended.conf"+Fore.LIGHTWHITE_EX)
self.log.info(Fore.YELLOW+"No ARL Token set in Extended.conf"+Fore.RESET)
self.report_status("NOT SET")
exit(0)
if self.currentARLToken is None:
@ -194,11 +222,15 @@ class LidarrExtendedAPI:
self.report_status('VALID') # For text fallback method
else:
self.report_status('EXPIRED')
self.log.error(Fore.RED + 'Update the token in extended.conf' + Fore.LIGHTWHITE_EX)
self.log.error(Fore.RED + 'Update the token in extended.conf' + Fore.RESET)
if self.telegram_bot_running: # Don't re-start the telegram bot if it's already running after bot invalid token entry
return False
if self.enable_pushover_notify:
pushover_notify(self.pushover_app_api_key, self.pushover_user_key, '---\U0001F6A8WARNING\U0001F6A8-----\nARL TOKEN EXPIRED\n Update arlToken in extended.conf"\n You can find a new ARL at:\nhttps://rentry.org/firehawk52#deezer-arls')
if self.enable_ntfy_notify:
ntfy_notify(self.ntfy_sever_topic, '---\U0001F6A8WARNING\U0001F6A8-----\nARL TOKEN EXPIRED\n Update arlToken in extended.conf\n You can find a new ARL at:\nhttps://rentry.org/firehawk52#deezer-arls', self.ntfy_user_token)
if self.enable_telegram_bot:
self.log.info(Fore.YELLOW + 'Starting Telegram bot...Check Telegram and follow instructions.' + Fore.LIGHTWHITE_EX)
self.log.info(Fore.YELLOW + 'Starting Telegram bot...Check Telegram and follow instructions.' + Fore.RESET)
self.telegram_bot_running = True
self.start_telegram_bot()
exit(420)
@ -226,11 +258,20 @@ class LidarrExtendedAPI:
f.close()
def start_telegram_bot(self):
self.bot = TelegramBotControl(self, self.telegram_bot_token, self.telegram_user_chat_id)
try:
self.bot = TelegramBotControl(self, self.telegram_bot_token, self.telegram_user_chat_id)
except Exception as e:
if 'Chat not found' in str(e) or 'Chat_id' in str(e):
self.log.error(
Fore.RED + "Telegram Bot: Chat not found. Check your chat ID in extended.conf, or start a chat with your bot." + Fore.RESET)
elif 'The token' in str(e):
self.log.error(Fore.RED + "Telegram Bot: Check your Bot Token in extended.conf." + Fore.RESET)
else:
self.log.error('Telegram Bot: ' + str(e))
def disable_telegram_bot(self):
compiled = re.compile(re.escape('true'), re.IGNORECASE)
self.fileText[self.telegramBotEnableLineIndex] = compiled.sub('false', self.telegramBotEnableLineText)
self.fileText[self.telegram_bot_enable_line_index] = compiled.sub('false', self.telegram_bot_enable_line_text)
with open(self.root+EXTENDED_CONF_PATH, 'w', encoding='utf-8') as file:
file.writelines(self.fileText)
file.close()
@ -247,7 +288,7 @@ class TelegramBotControl:
# Send initial notification
async def send_expired_token_notification(application):
await application.bot.sendMessage(chat_id=self.telegram_chat_id, text='---\U0001F6A8WARNING\U0001F6A8-----\nARL TOKEN EXPIRED\n Update Token by running "/set_token <TOKEN>"\n You can find a new ARL at:\nhttps://rentry.org/firehawk52#deezer-arls\n\n\n Other Commands:\n/cancel - Cancel this session\n/disable - Disable Telegram Bot', disable_web_page_preview=True)
self.log.info(Fore.YELLOW + "Telegram Bot Sent ARL Token Expiry Message " + Fore.LIGHTWHITE_EX)
self.log.info(Fore.YELLOW + "Telegram Bot Sent ARL Token Expiry Message " + Fore.RESET)
# TODO: Get Chat ID/ test on new bot
# start bot control
@ -263,12 +304,12 @@ class TelegramBotControl:
async def disable_bot(self, update, context: ContextTypes.DEFAULT_TYPE):
self.parent.disable_telegram_bot()
await update.message.reply_text('Disabled Telegram Bot. \U0001F614\nIf you would like to re-enable,\nset telegramBotEnable to true\nin extended.conf')
self.log.info(Fore.YELLOW + 'Telegram Bot: Send Disable Bot Message :(' + Fore.LIGHTWHITE_EX)
self.log.info(Fore.YELLOW + 'Telegram Bot: Send Disable Bot Message :(' + Fore.RESET)
self.application.stop_running()
async def cancel(self, update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text('Canceling...ARLToken is still expired.')
self.log.info(Fore.YELLOW + 'Telegram Bot: Canceling...ARLToken is still expired.' + Fore.LIGHTWHITE_EX)
self.log.info(Fore.YELLOW + 'Telegram Bot: Canceling...ARLToken is still expired.' + Fore.RESET)
try:
self.application.stop_running()
except Exception:
@ -280,7 +321,7 @@ class TelegramBotControl:
await update.message.reply_text(text=text)
else:
await context.bot.send_message(chat_id=update.effective_chat.id, text=text)
self.log.info(Fore.YELLOW+"Telegram Bot: " + text + Fore.LIGHTWHITE_EX)
self.log.info(Fore.YELLOW+"Telegram Bot: " + text + Fore.RESET)
try:
new_token = update.message.text.split('/set_token ')[1]
if new_token == '':
@ -288,7 +329,7 @@ class TelegramBotControl:
except:
await update.message.reply_text('Invalid Entry... Please try again.')
return
self.log.info(Fore.YELLOW+f"Telegram Bot:Token received: {new_token}" + Fore.LIGHTWHITE_EX)
self.log.info(Fore.YELLOW+f"Telegram Bot:Token received: {new_token}" + Fore.RESET)
token_validity = check_token(new_token)
if token_validity:
await send_message("ARL valid, applying...")
@ -310,6 +351,37 @@ class TelegramBotControl:
return
def pushover_notify(api_token, user_key, message): # Send Notification to Pushover
log = logging.getLogger('ARLChecker') # Get Logging
log.info(Fore.YELLOW + 'Attempting Pushover notification' + Fore.RESET)
response = requests.post("https://api.pushover.net/1/messages.json", data={
"token": api_token,
"user": user_key,
"message": message
})
if response.json()['status'] == 1:
log.info(Fore.GREEN+"Pushover notification sent successfully"+Fore.RESET)
else:
for message_error in response.json()['errors']:
log.error(Fore.RED+f"Pushover Response: {message_error}"+ Fore.RESET)
def ntfy_notify(server_plus_topic, message, token): # Send Notification to ntfy topic
log = logging.getLogger('ARLChecker') # Get Logging
log.info(Fore.YELLOW + 'Attempting ntfy notification' + Fore.RESET)
try:
requests.post(server_plus_topic,
data=message.encode(encoding='utf-8'),
headers={"Authorization": f"Bearer {token}"}
)
log.info(Fore.GREEN + 'ntfy notification sent successfully' + Fore.RESET)
except Exception as e:
if "Failed to resolve" in str(e):
log.error(Fore.RED + "ntfy ERROR: Check if server and user token correct in extended.conf"+Fore.RESET)
else:
log.error(Fore.RED + "NTFY ERROR: "+str(e)+Fore.RESET)
def check_token(token=None):
log = logging.getLogger('ARLChecker')
log.info(f"ARL Token to check: {token}")
@ -318,19 +390,19 @@ def check_token(token=None):
deezer_check = DeezerPlatformProvider()
account = deezer_check.login('', token.replace('"', ''))
if account.plan:
log.info(Fore.GREEN + f'Deezer Account Found.' + Fore.LIGHTWHITE_EX)
log.info(Fore.GREEN + f'Deezer Account Found.' + Fore.RESET)
log.info('-------------------------------')
log.info(f'Plan: {account.plan.name}')
log.info(f'Expiration: {account.plan.expires}')
log.info(f'Active: {Fore.GREEN+"Y" if account.plan.active else "N"}'+Fore.LIGHTWHITE_EX)
log.info(f'Download: {Fore.GREEN+"Y" if account.plan.download else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX)
log.info(f'Lossless: {Fore.GREEN+"Y" if account.plan.lossless else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX)
log.info(f'Explicit: {Fore.GREEN+"Y" if account.plan.explicit else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX)
log.info(f'Active: {Fore.GREEN+"Y" if account.plan.active else "N"}'+Fore.RESET)
log.info(f'Download: {Fore.GREEN+"Y" if account.plan.download else Fore.RED+"N"}'+Fore.RESET)
log.info(f'Lossless: {Fore.GREEN+"Y" if account.plan.lossless else Fore.RED+"N"}'+Fore.RESET)
log.info(f'Explicit: {Fore.GREEN+"Y" if account.plan.explicit else Fore.RED+"N"}'+Fore.RESET)
log.info('-------------------------------')
return True
except Exception as e:
if type(e) is AuthError:
log.error(Fore.RED + 'ARL Token Invalid/Expired.' + Fore.LIGHTWHITE_EX)
log.error(Fore.RED + 'ARL Token Invalid/Expired.' + Fore.RESET)
return False
else:
log.error(e)
@ -383,7 +455,7 @@ def init_logging(version, log_file_path):
# Initialize colorama
init(autoreset=True)
logger.info(Fore.GREEN + 'Logger initialized'+Fore.LIGHTWHITE_EX)
logger.info(Fore.GREEN + 'Logger initialized'+Fore.RESET)
return logger
@ -397,7 +469,7 @@ def main():
try:
if args.test_token:
log.info(Fore.CYAN+"CLI Token Tester"+Fore.LIGHTWHITE_EX)
log.info(Fore.CYAN+"CLI Token Tester"+Fore.RESET)
check_token(args.test_token)
exit(0)
arl_checker_instance = LidarrExtendedAPI()
@ -405,16 +477,8 @@ def main():
if args.check is True:
if arl_checker_instance.currentARLToken == '':
log.error("ARL Token not set. re-run with -n flag")
try:
arl_checker_instance.parse_extended_conf()
arl_checker_instance.check_token_wrapper()
except Exception as e:
if 'Chat not found' in str(e) or 'Chat_id' in str(e):
log.error(Fore.RED + "Chat not found. Check your chat ID in extended.conf, or start a chat with your bot."+Fore.LIGHTWHITE_EX)
elif 'The token' in str(e):
log.error(Fore.RED + "Check your Bot Token in extended.conf."+Fore.LIGHTWHITE_EX)
else:
log.error(e)
arl_checker_instance.parse_extended_conf()
arl_checker_instance.check_token_wrapper()
elif args.new_token:
if args.new_token == '':

1
lidarr/readme.md
View file

@ -94,6 +94,7 @@ This configuration does its best to update everything automatically, but with ho
* Checks Deezer ARL set in extended.conf at set interval for validity
* Reports ARL status in text file
* Optional Telegram bot with ability to set token from the chat
* Optional Pushover and ntfy notification upon ARL token expiration
For more details, visit the [Wiki](https://github.com/RandomNinjaAtk/arr-scripts/wiki/Lidarr)