import re from pathlib import Path from dataclasses import dataclass from requests import Session from argparse import ArgumentParser from sys import argv from colorama import Fore, init from telegram import Update from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler import logging import os from datetime import datetime # Pull script version from previous VERSION = '0.0.0' with open("./custom-services.d/ARLChecker", "r") as r: for line in r: if 'scriptVersion' in line: VERSION = re.search(r'"([A-Za-z0-9_\./\\-]*)"', line)[0].replace('"','') # Logging Setup logging.basicConfig( format=f'%(asctime)s :: ARLChecker :: {VERSION} :: %(levelname)s :: %(message)s', datefmt='%Y-%m-%d %H:%M:%S', level=logging.INFO ) logger = logging.getLogger(__name__) # Initialize colorama init(autoreset=True) # Web agent used to access Deezer USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/110.0' @dataclass class Plan: name: str expires: str active: bool download: bool lossless: bool explicit: bool @dataclass class Account: id: int token: str country: str plan: Plan class AuthError(Exception): pass class ParseError(Exception): pass class ServiceError(Exception): pass class DeezerPlatformProvider: NAME = 'Deezer' BASE_URL = 'http://www.deezer.com' API_PATH = '/ajax/gw-light.php' SESSION_DATA = { 'api_token': 'null', 'api_version': '1.0', 'input': '3', 'method': 'deezer.getUserData' } def __init__(self): super().__init__() self.session = Session() self.session.headers.update({'User-Agent': USER_AGENT}) def login(self, username, secret): try: res = self.session.post( self.BASE_URL + self.API_PATH, cookies={'arl': secret}, data=self.SESSION_DATA ) res.raise_for_status() except Exception as error: logger.error(Fore.RED + 'Could not connect! Service down, API changed, wrong credentials or code-related issue.' + Fore.WHITE) raise ConnectionError() self.session.cookies.clear() try: res = res.json() except Exception as error: logger.error(Fore.RED + "Could not parse JSON response from DEEZER!" + Fore.WHITE) raise ParseError() if 'error' in res and res['error']: logger.error(Fore.RED + "Deezer returned the following error:{}".format(res["error"]) + Fore.WHITE) raise ServiceError() res = res['results'] if res['USER']['USER_ID'] == 0: logger.error(Fore.RED+"ARL Token Expired. Update the token in extended.conf"+Fore.WHITE) raise AuthError() return Account(username, secret, res['COUNTRY'], Plan( res['OFFER_NAME'], 'Unknown', True, True, res['USER']['OPTIONS']['web_sound_quality']['lossless'], res['USER']['EXPLICIT_CONTENT_LEVEL'] )) class LidarrExtendedAPI: # sets new token to extended.conf def __init__(self, new_arl_token): workingDir = Path(os.getcwd()) print(workingDir) #self.parentDir = str(workingDir.parents[1]) self.parentDir = str(workingDir.parents[3]) print(self.parentDir) self.extendedConfDir = self.parentDir + '/config/extended.conf' self.newARLToken = new_arl_token self.arlToken = None self.arlLineText = None self.arlLineIndex = None self.fileText = None self.enable_telegram_bot = False self.telegram_bot_running = False self.telegram_bot_token = None self.telegram_user_chat_id = None self.telegramBotEnableLineText = None self.telegramBotEnableLineIndex = None self.bot = None self.parse_extended_conf() def parse_extended_conf(self): deezer_active = False self.arlToken = None arl_token_match = None re_search_pattern = r'"([^"]*)"' try: # Try to open extended.conf and read all text into a var. with open(self.extendedConfDir, 'r', encoding='utf-8') as file: self.fileText = file.readlines() file.close() except: logger.error(f"Could not find {self.extendedConfDir}") exit(1) # Ensure Deezer is enabled and ARL token is populated for line in self.fileText: if 'dlClientSource="deezer"' in line or 'dlClientSource="both"' in line: deezer_active = True if 'arlToken=' in line: self.arlLineText = line self.arlLineIndex = self.fileText.index(self.arlLineText) arl_token_match = re.search(re_search_pattern, line) break # ARL Token wrong flag error handling. if arl_token_match is None: logger.error("ARL Token not found in extended.conf. Exiting") exit(1) elif deezer_active is False: logger.error("Deezer not set as an active downloader in extended.conf. Exiting") file.close() exit(1) self.arlToken = arl_token_match[0] logger.info('ARL Found in extended.conf') for line in self.fileText: if 'telegramBotEnable=' in line: self.telegramBotEnableLineText = line self.telegramBotEnableLineIndex = self.fileText.index(self.telegramBotEnableLineText) 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 self.enable_telegram_bot: logger.info('Telegram bot is enabled.') if self.telegram_bot_token is None or self.telegram_user_chat_id is None: logger.error('Telegram bot token or user chat ID not set in extended.conf. Exiting') exit(1) else: logger.info('Telegram bot is disabled. Set the flag in extended.conf to enable.') # Uses DeezerPlatformProvider to check if the token is valid def check_token(self, token=None): logger.info('Checking ARL Token Validity...') if token == '""': logger.info(Fore.YELLOW+"No ARL Token set in Extended.conf"+Fore.WHITE) self.report_status("NOT SET") exit(0) if token is None: print('Invalid ARL Token Entry') return False try: deezer_check = DeezerPlatformProvider() account = deezer_check.login('', token.replace('"','')) if account.plan: logger.info(Fore.GREEN + f'Deezer Account Found.'+ Fore.WHITE) logger.info('-------------------------------') logger.info(f'Plan: {account.plan.name}') logger.info(f'Expiration: {account.plan.expires}') logger.info(f'Active: {Fore.GREEN+"Y" if account.plan.active else "N"}'+Fore.WHITE) logger.info(f'Download: {Fore.GREEN+"Y" if account.plan.download else Fore.RED+"N"}'+Fore.WHITE) logger.info(f'Lossless: {Fore.GREEN+"Y" if account.plan.lossless else Fore.RED+"N"}'+Fore.WHITE) logger.info(f'Explicit: {Fore.GREEN+"Y" if account.plan.explicit else Fore.RED+"N"}'+Fore.WHITE) logger.info('-------------------------------') self.report_status('VALID') return True except Exception as e: print(e) self.report_status('EXPIRED') if self.telegram_bot_running: return False if self.enable_telegram_bot: logger.info('Starting Telegram bot...Check Telegram and follow instructions.') self.telegram_bot_running = True self.start_telegram_bot() exit(420) def set_new_token(self): # Re-writes extended.conf with previously read-in text, replacing w/ new ARL self.fileText[self.arlLineIndex] = self.arlLineText.replace(self.arlToken, self.newARLToken) with open(self.extendedConfDir, 'w', encoding='utf-8') as file: file.writelines(self.fileText) file.close() logger.info("New ARL token written to extended.conf") # After new token is set, clean up notfound and failed downloads to bypass the default 30 day wait def clear_not_found(self): paths = [self.parentDir + '/config/extended/logs/notfound',self.parentDir+'/config/extended/logs/downloaded/failed/deezer'] for path in paths: for file in os.listdir(path): file_to_delete = os.path.join(path,file) os.remove(file_to_delete) def report_status(self, status): f = open("/custom-services.d/python/ARLStatus.txt", "w") now = datetime.strftime(datetime.now(),"%b-%d-%Y at %H:%M:%S") f.write(f"{now}: ARL Token is {status}.{' Please update arlToken in extended.conf' if status=='EXPIRED' else ''}") f.close() def start_telegram_bot(self): self.bot = TelegramBotControl(self,self.telegram_bot_token,self.telegram_user_chat_id) def disable_telegram_bot(self): compiled = re.compile(re.escape('true'), re.IGNORECASE) self.fileText[self.telegramBotEnableLineIndex] = compiled.sub('false', self.telegramBotEnableLineText) with open(self.extendedConfDir, 'w', encoding='utf-8') as file: file.writelines(self.fileText) file.close() logger.info("Telegram Bot Disabled.") class TelegramBotControl: def __init__(self, parent,telegram_bot_token,telegram_user_chat_id): 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) # TODO: Get Chat ID/ test on new bot self.parent = parent self.telegram_bot_token = telegram_bot_token self.telegram_chat_id = telegram_user_chat_id # start bot control self.application = ApplicationBuilder().token(self.telegram_bot_token).post_init(send_expired_token_notification).build() token_handler = CommandHandler('set_token', self.set_token) cancel_handler = CommandHandler('cancel', self.cancel) disable_handler = CommandHandler('disable', self.disable_bot) self.application.add_handler(token_handler) self.application.add_handler(cancel_handler) self.application.add_handler(disable_handler) self.application.run_polling(allowed_updates=Update.ALL_TYPES) 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.application.stop_running() async def cancel(self, update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text('Canceling...ARLToken is still expired.') try: self.application.stop_running() except Exception: pass async def set_token(self, update, context: ContextTypes.DEFAULT_TYPE): try: new_token = update.message.text.split('/set_token ')[1] if new_token == '': raise Exception except: await update.message.reply_text('Invalid Entry... please try again.') return print(new_token) logger.info("Testing ARL Token Validity...") token_validity = self.parent.check_token(new_token) if token_validity: await context.bot.send_message(chat_id=update.effective_chat.id, text="ARL valid, applying...") self.parent.newARLToken = '"'+new_token+'"' self.parent.set_new_token() self.parent.arlToken = self.parent.newARLToken # TODO Fix this garbage - move functionality out of telegram stuff await context.bot.send_message(chat_id=update.effective_chat.id, text="Checking configuration") # reparse extended.conf self.parent.parse_extended_conf() token_validity = self.parent.check_token(self.parent.arlToken) if token_validity: await context.bot.send_message(chat_id=update.effective_chat.id, text="ARL Updated! \U0001F44D") try: await self.application.stop_running() except Exception: pass else:# If Token invalid await update.message.reply_text(text="Token expired or inactive. try another token.") return def main(arlToken = None): parser = ArgumentParser(prog='Account Checker', description='Check if Deezer ARL Token is valid') parser.add_argument('-c', '--check', help='Check if current ARL Token is active/valid',required=False, default=False, action='store_true') parser.add_argument('-n', '--new', help='Set new ARL Token',type = str, required=False, default=False) if not argv[1:]: parser.print_help() parser.exit() args = parser.parse_args() arlToken_instance = LidarrExtendedAPI(arlToken) if args.check is True: if arlToken_instance.arlToken == '': print("ARL Token not set. re-run with -n flag") exit(1) try: arlToken_instance.check_token(arlToken_instance.arlToken) except Exception as e: if 'Chat not found' in str(e): logger.error(Fore.RED + "Chat not found. Check your chat ID in extended.conf, or start a chat with your bot."+Fore.WHITE) elif 'The token' in str(e): logger.error(Fore.RED + "Check your Bot Token in extended.conf."+Fore.WHITE) else: print(e) exit(1) elif args.new: if args.new == '': print("Please pass new ARL token as an argument") exit(96) arlToken_instance.newARLToken = '"'+args.new+'"' arlToken_instance.set_new_token() else: parser.print_help() if __name__ == '__main__': main('FAKETOKEN')