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

1.6 dev: debugging setup complete. refactor logging, version fetching to functions.

Browse source
This commit is contained in:
hockeygoalie35 2024-02-16 23:50:21 -05:00
parent 3f914e6774
commit 1e053a8064
2 changed files with 135 additions and 130 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 #!/usr/bin/with-contenv bash
### Default values ### Default values
scriptVersion="1.6 dev" scriptVersion="1.6dev"
scriptName="ARLChecker" scriptName="ARLChecker"
sleepInterval='24h' sleepInterval='24h'
### Import Settings ### Import Settings

263
lidarr/python/ARLChecker.py
View file

@ -11,40 +11,15 @@ import logging
import os import os
from datetime import datetime from datetime import datetime
# TODO: Breakout check function to be able to test new ARL tokens
CUSTOM_SERVICES_PATH = '/custom-services.d/' CUSTOM_SERVICES_PATH = '/custom-services.d/'
CUSTOM_INIT_PATH = '/custom-cont_init.d/' CUSTOM_INIT_PATH = '/custom-cont_init.d/'
EXTENDED_CONF_PATH = '/config/extended.conf' EXTENDED_CONF_PATH = '/config/extended.conf'
NOT_FOUND_PATH = '/config/extended/logs/notfound' NOT_FOUND_PATH = '/config/extended/logs/notfound'
FAILED_DOWNLOADS_PATH = '/config/extended/logs/downloaded/failed/deezer' FAILED_DOWNLOADS_PATH = '/config/extended/logs/downloaded/failed/deezer'
STATUS_FALLBACK_LOCATION = '/custom-services.d/python/ARLStatus.txt' STATUS_FALLBACK_LOCATION = '/custom-services.d/python/ARLStatus.txt'
LOG_FILES_DIRECTORY = '/config/logs'
# Pull script version from bash script. will likely change this to a var passthrough
with open(CUSTOM_SERVICES_PATH+"ARLChecker", "r") as r:
for line in r:
if 'scriptVersion' in line:
VERSION = re.search(r'"([A-Za-z0-9_\./\\-]*)"', line)[0].replace('"','')
# Get current log file
path = '/config/logs'
latest_file = max([os.path.join(path, f) for f in os.listdir(path) if 'ARLChecker' in f],key=os.path.getctime)
# Logging Setup
logging.basicConfig(
format=f'%(asctime)s :: ARLChecker :: {VERSION} :: %(levelname)s :: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.INFO,
handlers=[
logging.StreamHandler(stdout),
logging.FileHandler(latest_file, mode="a")
]
)
logger = logging.getLogger(__name__)
# Initialize colorama
init(autoreset=True)
# Web agent used to access Deezer # 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' USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/110.0'
@ -93,6 +68,7 @@ class DeezerPlatformProvider:
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.log = logging.getLogger('ARLChecker')
self.session = Session() self.session = Session()
self.session.headers.update({'User-Agent': USER_AGENT}) self.session.headers.update({'User-Agent': USER_AGENT})
@ -105,7 +81,7 @@ class DeezerPlatformProvider:
) )
res.raise_for_status() res.raise_for_status()
except Exception as error: except Exception as error:
logger.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.LIGHTWHITE_EX)
raise ConnectionError() raise ConnectionError()
self.session.cookies.clear() self.session.cookies.clear()
@ -113,17 +89,17 @@ class DeezerPlatformProvider:
try: try:
res = res.json() res = res.json()
except Exception as error: except Exception as error:
logger.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.LIGHTWHITE_EX)
raise ParseError() raise ParseError()
if 'error' in res and res['error']: if 'error' in res and res['error']:
logger.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.LIGHTWHITE_EX)
raise ServiceError() raise ServiceError()
res = res['results'] res = res['results']
if res['USER']['USER_ID'] == 0: if res['USER']['USER_ID'] == 0:
logger.error(Fore.RED+"ARL Token Expired. Update the token in extended.conf"+Fore.LIGHTWHITE_EX) self.log.error(Fore.RED+"ARL Token Expired. Update the token in extended.conf"+Fore.LIGHTWHITE_EX)
raise AuthError() raise AuthError()
return Account(username, secret, res['COUNTRY'], Plan( return Account(username, secret, res['COUNTRY'], Plan(
@ -138,19 +114,11 @@ class DeezerPlatformProvider:
class LidarrExtendedAPI: class LidarrExtendedAPI:
# sets new token to extended.conf # sets new token to extended.conf
def __init__(self, new_arl_token, root_path=''): def __init__(self):
# :param new_ark_token: if running with the -n flag, sets the supplied string as the new token self.root = ''
# :param root_path: change root of all path params to param, for debugging outside the Docker container. self.log = logging.getLogger('ARLChecker')
# TODO: change these to be changed in main self.newARLToken = None
self.root = root_path self.currentARLToken = None
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 + EXTENDED_CONF_PATH
self.newARLToken = new_arl_token
self.arlToken = None
self.arlLineText = None self.arlLineText = None
self.arlLineIndex = None self.arlLineIndex = None
self.fileText = None self.fileText = None
@ -160,23 +128,19 @@ class LidarrExtendedAPI:
self.telegram_user_chat_id = None self.telegram_user_chat_id = None
self.telegramBotEnableLineText = None self.telegramBotEnableLineText = None
self.telegramBotEnableLineIndex = None self.telegramBotEnableLineIndex = None
self.bot = None self.bot = None
self.parse_extended_conf()
def parse_extended_conf(self): def parse_extended_conf(self):
deezer_active = False self.currentARLToken = None
self.arlToken = None
arl_token_match = None arl_token_match = None
deezer_active = False
re_search_pattern = r'"([^"]*)"' re_search_pattern = r'"([^"]*)"'
try: # Try to open extended.conf and read all text into a var. try: # Try to open extended.conf and read all text into a var.
with open(self.extendedConfDir, 'r', encoding='utf-8') as file: with open(self.root+EXTENDED_CONF_PATH, 'r', encoding='utf-8') as file:
self.fileText = file.readlines() self.fileText = file.readlines()
file.close() file.close()
except: except:
logger.error(f"Could not find {self.extendedConfDir}") self.log.error(f"Could not find {self.root+EXTENDED_CONF_PATH}")
exit(1) exit(1)
# Ensure Deezer is enabled and ARL token is populated # Ensure Deezer is enabled and ARL token is populated
for line in self.fileText: for line in self.fileText:
@ -190,14 +154,14 @@ class LidarrExtendedAPI:
# ARL Token wrong flag error handling. # ARL Token wrong flag error handling.
if arl_token_match is None: if arl_token_match is None:
logger.error("ARL Token not found in extended.conf. Exiting") self.log.error("ARL Token not found in extended.conf. Exiting")
exit(1) exit(1)
elif deezer_active is False: elif deezer_active is False:
logger.error("Deezer not set as an active downloader in extended.conf. Exiting") self.log.error("Deezer not set as an active downloader in extended.conf. Exiting")
file.close() file.close()
exit(1) exit(1)
self.arlToken = arl_token_match[0] self.currentARLToken = arl_token_match[0]
logger.info('ARL Found in extended.conf') self.log.info('ARL Found in extended.conf')
for line in self.fileText: for line in self.fileText:
if 'telegramBotEnable=' in line: if 'telegramBotEnable=' in line:
@ -211,36 +175,36 @@ class LidarrExtendedAPI:
if self.enable_telegram_bot: if self.enable_telegram_bot:
logger.info('Telegram bot is enabled.') self.log.info('Telegram bot is enabled.')
if self.telegram_bot_token is None or self.telegram_user_chat_id is None: 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') self.log.error('Telegram bot token or user chat ID not set in extended.conf. Exiting')
exit(1) exit(1)
else: else:
logger.info('Telegram bot is disabled. Set the flag in extended.conf to enable.') self.log.info('Telegram bot is disabled. Set the flag in extended.conf to enable.')
# Uses DeezerPlatformProvider to check if the token is valid # Uses DeezerPlatformProvider to check if the token is valid
def check_token(self, token=None): def check_token(self, token=None):
logger.info('Checking ARL Token Validity...') self.log.info('Checking ARL Token Validity...')
if token == '""': if token == '""':
logger.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.LIGHTWHITE_EX)
self.report_status("NOT SET") self.report_status("NOT SET")
exit(0) exit(0)
if token is None: if token is None:
print('Invalid ARL Token Entry') self.log.error('Invalid ARL Token Entry (None Object)')
return False return False
try: try:
deezer_check = DeezerPlatformProvider() deezer_check = DeezerPlatformProvider()
account = deezer_check.login('', token.replace('"','')) account = deezer_check.login('', token.replace('"',''))
if account.plan: if account.plan:
logger.info(Fore.GREEN + f'Deezer Account Found.'+ Fore.LIGHTWHITE_EX) self.log.info(Fore.GREEN + f'Deezer Account Found.'+ Fore.LIGHTWHITE_EX)
logger.info('-------------------------------') self.log.info('-------------------------------')
logger.info(f'Plan: {account.plan.name}') self.log.info(f'Plan: {account.plan.name}')
logger.info(f'Expiration: {account.plan.expires}') self.log.info(f'Expiration: {account.plan.expires}')
logger.info(f'Active: {Fore.GREEN+"Y" if account.plan.active else "N"}'+Fore.LIGHTWHITE_EX) self.log.info(f'Active: {Fore.GREEN+"Y" if account.plan.active else "N"}'+Fore.LIGHTWHITE_EX)
logger.info(f'Download: {Fore.GREEN+"Y" if account.plan.download else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX) self.log.info(f'Download: {Fore.GREEN+"Y" if account.plan.download else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX)
logger.info(f'Lossless: {Fore.GREEN+"Y" if account.plan.lossless else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX) self.log.info(f'Lossless: {Fore.GREEN+"Y" if account.plan.lossless else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX)
logger.info(f'Explicit: {Fore.GREEN+"Y" if account.plan.explicit else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX) self.log.info(f'Explicit: {Fore.GREEN+"Y" if account.plan.explicit else Fore.RED+"N"}'+Fore.LIGHTWHITE_EX)
logger.info('-------------------------------') self.log.info('-------------------------------')
self.report_status('VALID') self.report_status('VALID')
return True return True
except Exception as e: except Exception as e:
@ -249,17 +213,18 @@ class LidarrExtendedAPI:
if self.telegram_bot_running: if self.telegram_bot_running:
return False return False
if self.enable_telegram_bot: if self.enable_telegram_bot:
logger.info('Starting Telegram bot...Check Telegram and follow instructions.') self.log.info('Starting Telegram bot...Check Telegram and follow instructions.')
self.telegram_bot_running = True self.telegram_bot_running = True
self.start_telegram_bot() self.start_telegram_bot()
exit(420) exit(420)
def set_new_token(self): # Re-writes extended.conf with previously read-in text, replacing w/ new ARL 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) self.fileText[self.arlLineIndex] = self.arlLineText.replace(self.currentARLToken, self.newARLToken)
with open(self.extendedConfDir, 'w', encoding='utf-8') as file: with open(self.root+EXTENDED_CONF_PATH, 'w', encoding='utf-8') as file:
file.writelines(self.fileText) file.writelines(self.fileText)
file.close() file.close()
logger.info("New ARL token written to extended.conf") self.log.info("New ARL token written to extended.conf")
self.parse_extended_conf()
# After new token is set, clean up notfound and failed downloads to bypass the default 30 day wait # After new token is set, clean up notfound and failed downloads to bypass the default 30 day wait
def clear_not_found(self): def clear_not_found(self):
@ -270,7 +235,7 @@ class LidarrExtendedAPI:
os.remove(file_to_delete) os.remove(file_to_delete)
def report_status(self, status): def report_status(self, status):
f = open(STATUS_FALLBACK_LOCATION, "w") f = open(self.root+STATUS_FALLBACK_LOCATION, "w")
now = datetime.strftime(datetime.now(),"%b-%d-%Y at %H:%M:%S") 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.write(f"{now}: ARL Token is {status}.{' Please update arlToken in extended.conf' if status=='EXPIRED' else ''}")
f.close() f.close()
@ -281,24 +246,26 @@ class LidarrExtendedAPI:
def disable_telegram_bot(self): def disable_telegram_bot(self):
compiled = re.compile(re.escape('true'), re.IGNORECASE) compiled = re.compile(re.escape('true'), re.IGNORECASE)
self.fileText[self.telegramBotEnableLineIndex] = compiled.sub('false', self.telegramBotEnableLineText) self.fileText[self.telegramBotEnableLineIndex] = compiled.sub('false', self.telegramBotEnableLineText)
with open(self.extendedConfDir, 'w', encoding='utf-8') as file: with open(self.root+EXTENDED_CONF_PATH, 'w', encoding='utf-8') as file:
file.writelines(self.fileText) file.writelines(self.fileText)
file.close() file.close()
logger.info("Telegram Bot Disabled.") self.log.info("Telegram Bot Disabled.")
class TelegramBotControl: class TelegramBotControl:
def __init__(self, parent,telegram_bot_token,telegram_user_chat_id): def __init__(self, parent, telegram_bot_token, telegram_user_chat_id):
self.log = logging.getLogger('ARLChecker')
self.parent = parent
self.telegram_bot_token = telegram_bot_token
self.telegram_chat_id = telegram_user_chat_id
# Send initial notification
async def send_expired_token_notification(application): 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) 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)
# TODO: Get Chat ID/ test on new bot # 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 # start bot control
self.application = ApplicationBuilder().token(self.telegram_bot_token).post_init(send_expired_token_notification).build() self.application = ApplicationBuilder().token(self.telegram_bot_token).post_init(send_expired_token_notification).build()
token_handler = CommandHandler('set_token', self.set_token) token_handler = CommandHandler('set_token', self.set_token)
@ -309,14 +276,12 @@ class TelegramBotControl:
self.application.add_handler(disable_handler) self.application.add_handler(disable_handler)
self.application.run_polling(allowed_updates=Update.ALL_TYPES) self.application.run_polling(allowed_updates=Update.ALL_TYPES)
async def disable_bot(self, update):
async def disable_bot(self, update, context: ContextTypes.DEFAULT_TYPE):
self.parent.disable_telegram_bot() 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') 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() self.application.stop_running()
async def cancel(self, update):
async def cancel(self, update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text('Canceling...ARLToken is still expired.') await update.message.reply_text('Canceling...ARLToken is still expired.')
try: try:
self.application.stop_running() self.application.stop_running()
@ -331,14 +296,11 @@ class TelegramBotControl:
await update.message.reply_text('Invalid Entry... please try again.') await update.message.reply_text('Invalid Entry... please try again.')
return return
print(new_token) print(new_token)
logger.info("Testing ARL Token Validity...") self.log.info("Testing ARL Token Validity...")
token_validity = self.parent.check_token(new_token) token_validity = self.parent.check_token(new_token)
if token_validity: if token_validity:
await context.bot.send_message(chat_id=update.effective_chat.id, text="ARL valid, applying...") await context.bot.send_message(chat_id=update.effective_chat.id, text="ARL valid, applying...")
self.parent.newARLToken = '"'+new_token+'"' 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") await context.bot.send_message(chat_id=update.effective_chat.id, text="Checking configuration")
# reparse extended.conf # reparse extended.conf
self.parent.parse_extended_conf() self.parent.parse_extended_conf()
@ -350,59 +312,102 @@ class TelegramBotControl:
except Exception: except Exception:
pass pass
else:# If Token invalid else: # If Token invalid
await update.message.reply_text(text="Token expired or inactive. try another token.") await update.message.reply_text(text="Token expired or inactive. try another token.")
return return
def parse_arguments():
def main(arlToken = None):
parser = ArgumentParser(prog='Account Checker', description='Check if Deezer ARL Token is valid') 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('-c', '--check', help='Check if currently set ARL Token is active/valid',required=False, default=False, action='store_true')
parser.add_argument('-t', '--test', help='Test new token for validity', required=False,default=False, action='store_true')
parser.add_argument('-d', '--debug', help='For debug and development', required=False, default=False,action='store_true')
parser.add_argument('-n', '--new', help='Set new ARL Token',type = str, required=False, default=False) parser.add_argument('-n', '--new', help='Set new ARL Token',type = str, required=False, default=False)
if not argv[1:]: if not argv[1:]:
parser.print_help() parser.print_help()
parser.exit() parser.exit()
args = parser.parse_args() return 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.LIGHTWHITE_EX)
elif 'The token' in str(e):
logger.error(Fore.RED + "Check your Bot Token in extended.conf."+Fore.LIGHTWHITE_EX)
else:
print(e)
exit(1)
elif args.new: def get_version(root):
if args.new == '': # Pull script version from bash script. will likely change this to a var passthrough
print("Please pass new ARL token as an argument") with open(root+CUSTOM_SERVICES_PATH+"ARLChecker", "r") as r:
exit(96) for line in r:
if 'scriptVersion' in line:
arlToken_instance.newARLToken = '"'+args.new+'"' return re.search(r'"([A-Za-z0-9_\./\\-]*)"', line)[0].replace('"','')
arlToken_instance.set_new_token() logging.error('Script Version not found! Exiting...')
exit(1)
else:
parser.print_help()
def enable_debug(debug_enable = False): def get_active_log(root):
if debug_enable is True: # Get current log file
root = './env/' path = root + LOG_FILES_DIRECTORY
else: latest_file = max([os.path.join(path, f) for f in os.listdir(path) if 'ARLChecker' in f], key=os.path.getctime)
root = '' return latest_file
return root
def init_logging(version, log_file_path):
# Logging Setup
logging.basicConfig(
format=f'%(asctime)s :: ARLChecker :: {version} :: %(levelname)s :: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.INFO,
handlers=[
logging.StreamHandler(stdout),
logging.FileHandler(log_file_path, mode="a")
]
)
logger = logging.getLogger('ARLChecker')
# Initialize colorama
init(autoreset=True)
logger.info('Logger initialized')
return logger
def main():
root = ''
args = parse_arguments()
arl_checker_instance = LidarrExtendedAPI()
if args.debug:
root = './env'
log = init_logging(get_version(root), get_active_log(root))
arl_checker_instance.root = root
#arl_checker_instance.log = log
try:
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(arl_checker_instance.currentARLToken)
except Exception as e:
if 'Chat not found' 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)
elif args.new:
if args.new == '':
print("Please pass new ARL token as an argument")
exit(96)
arl_checker_instance.newARLToken = '"'+args.new+'"'
arl_checker_instance.set_new_token()
else:
args.print_help()
except Exception as e:
logging.error(e, exc_info=True)
exit(1)
if __name__ == '__main__': if __name__ == '__main__':
main('FAKETOKEN') main()