diff --git a/lidarr/ARLChecker b/lidarr/ARLChecker index 7d2dbc5..6b7910d 100644 --- a/lidarr/ARLChecker +++ b/lidarr/ARLChecker @@ -1,6 +1,6 @@ #!/usr/bin/with-contenv bash ### Default values -scriptVersion="1.6" +scriptVersion="2.0" scriptName="ARLChecker" sleepInterval='24h' ### Import Settings diff --git a/lidarr/extended.conf b/lidarr/extended.conf index af2ffa6..d4f6ce7 100644 --- a/lidarr/extended.conf +++ b/lidarr/extended.conf @@ -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. diff --git a/lidarr/python/ARLChecker.py b/lidarr/python/ARLChecker.py index ba933c4..88f5738 100644 --- a/lidarr/python/ARLChecker.py +++ b/lidarr/python/ARLChecker.py @@ -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 "\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 == '': diff --git a/lidarr/readme.md b/lidarr/readme.md index 9ab1fbb..826d0a5 100644 --- a/lidarr/readme.md +++ b/lidarr/readme.md @@ -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)