lib/config/config_callbacks.py

"""
MIT License

Copyright (C) 2021 ROCKY4546
https://github.com/rocky4546

This file is part of Cabernet

Permission is hereby granted, free of charge, to any person obtaining a copy of this software
and associated documentation files (the “Software”), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify, merge, publish, distribute,
sublicense, and/or sell copies of the Software, and to permit persons to whom the Software
is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or
substantial portions of the Software.
"""

import getpass
import importlib
import importlib.resources
import logging
import logging.config
import os
import pathlib
import platform
import urllib.request
import uuid

import lib.common.utils as utils
import lib.common.encryption as encryption
import lib.config.config_defn as config_defn
import lib.clients.hdhr.hdhr_server as hdhr_server
from lib.db.db_config_defn import DBConfigDefn
from lib.db.db_scheduler import DBScheduler
from lib.db.db_channels import DBChannels
from lib.clients.web_handler import WebHTTPHandler

try:
    import cryptography
    from cryptography.hazmat.backends import default_backend
    from cryptography.hazmat.primitives import hashes
    from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
    from cryptography.fernet import Fernet

    CRYPTO_LOADED = True
except ImportError:
    CRYPTO_LOADED = False

ENCRYPT_STRING = 'ENC::'


def noop(_config_obj, _section, _key):
    pass


def logging_refresh(_config_obj, _section, _key):
    logging.config.fileConfig(fname=_config_obj.data['paths']['config_file'], disable_existing_loggers=False)
    req = urllib.request.Request('http://{}:{}/logreset'.format(
        _config_obj.data['web']['plex_accessible_ip'], str(_config_obj.data['web']['plex_accessible_port'])))
    with urllib.request.urlopen(req) as resp:
        content = resp.read()


def logging_enable(_config_obj, _section, _key):
    # update the config_file to enable or disable the log
    # [logger_root]
    # handlers = loghandler, filehandler
    handler_list = []
    if _config_obj.data['handler_filehandler']['enabled']:
        handler_list.append('filehandler')
    if _config_obj.data['handler_loghandler']['enabled']:
        handler_list.append('loghandler')
    handlers = ','.join(handler_list)
    _config_obj.write('logger_root', 'handlers', handlers)
    logging_refresh(_config_obj, _section, _key)


def set_version(_config_obj, _section, _key):
    _config_obj.data[_section][_key] \
        = utils.get_version_str()


def set_system(_config_obj, _section, _key):
    _config_obj.data[_section][_key] \
        = platform.system()


def set_python_version(_config_obj, _section, _key):
    _config_obj.data[_section][_key] \
        = platform.python_version()


def set_user(_config_obj, _section, _key):
    _config_obj.data[_section][_key] \
        = getpass.getuser()


def set_os(_config_obj, _section, _key):
    _config_obj.data[_section][_key] \
        = platform.version()


def set_path(_config_obj, _section, _key, _base_dir, _folder):
    if not _config_obj.data[_section][_key]:
        _config_obj.data[_section][_key] = pathlib.Path(_base_dir).joinpath(_folder)
    else:
        _config_obj.data[_section][_key] = pathlib.Path(_config_obj.data[_section][_key])
    if not _config_obj.data[_section][_key].is_dir():
        _config_obj.data[_section][_key].mkdir()
    _config_obj.data[_section][_key] = str(os.path.abspath(_config_obj.data[_section][_key]))


def set_data_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        set_path(_config_obj, _section, _key,
                 _config_obj.data['paths']['main_dir'], 'data')


def set_logs_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        set_path(_config_obj, _section, _key,
                 _config_obj.data['paths']['data_dir'], 'logs')


def set_thumbnails_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        set_path(_config_obj, _section, _key,
                 _config_obj.data['paths']['data_dir'], 'thumbnails')


def set_temp_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        set_path(_config_obj, _section, _key,
                 _config_obj.data['paths']['data_dir'], 'tmp')


def set_database_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        set_path(_config_obj, _section, _key,
                 _config_obj.data['paths']['data_dir'], 'db')


def set_backup_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        set_path(_config_obj, _section, _key,
                 _config_obj.data['paths']['data_dir'], 'backups')


def set_configdefn_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        _config_obj.data[_section][_key] = _config_obj.defn_json.defn_path


def set_main_path(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        _config_obj.data['paths']['main_dir'] = _config_obj.script_dir


def set_ffmpeg_path(_config_obj, _section, _key):
    if not _config_obj.data[_section][_key]:
        if platform.system() in ['Windows']:
            base_ffmpeg_dir \
                = pathlib.Path(_config_obj.script_dir).joinpath('ffmpeg/bin')
            if base_ffmpeg_dir.is_dir():
                _config_obj.data[_section][_key] \
                    = str(pathlib.Path(base_ffmpeg_dir).joinpath('ffmpeg.exe'))
            else:
                _config_obj.data[_section][_key] = 'ffmpeg.exe'
                _config_obj.logger.notice(
                    'ffmpeg_path does not exist in [cabernet]/ffmpeg/bin, will use PATH env to find ffmpeg.exe')
        else:
            _config_obj.data[_section][_key] = 'ffmpeg'


def set_ffprobe_path(_config_obj, _section, _key):
    if not _config_obj.data[_section][_key]:
        if platform.system() in ['Windows']:
            base_ffmpeg_dir \
                = pathlib.Path(_config_obj.script_dir).joinpath('ffmpeg/bin')
            if base_ffmpeg_dir.is_dir():
                _config_obj.data[_section][_key] \
                    = str(pathlib.Path(base_ffmpeg_dir).joinpath('ffprobe.exe'))
            else:
                _config_obj.data[_section][_key] = 'ffprobe.exe'
                _config_obj.logger.notice(
                    'ffprobe_path does not exist in [cabernet]/ffmpeg/bin, will use PATH env to find ffprobe.exe')
        else:
            _config_obj.data[_section][_key] = 'ffprobe'


def set_streamlink_path(_config_obj, _section, _key):
    if not _config_obj.data[_section][_key]:
        if platform.system() in ['Windows']:
            _config_obj.data[_section][_key] \
                = 'streamlink.exe'
            _config_obj.logger.notice(
                'streamlink_path does not exist in PATH to find streamlink.exe')
        else:
            streamlink_file = os.path.expanduser('~/.local/bin/streamlink')
            if os.path.isfile(streamlink_file):
                _config_obj.data[_section][_key] = streamlink_file
            else:
                _config_obj.data[_section][_key] = 'streamlink'


def set_pdata(_config_obj, _section, _key):
    if not _config_obj.data[_section][_key]:
        _config_obj.data[_section][_key] = \
            utils.PLUGIN_DATA + config_defn.PLUGIN_DATA


def check_encryption(_config_obj, _section, _key):
    if not CRYPTO_LOADED:
        return 'python cryptography module not installed, unable to encrypt'


def load_encrypted_setting(_config_obj, _section, _key):
    if CRYPTO_LOADED and _config_obj.data['main']['use_encryption']:
        if _config_obj.data['main']['encrypt_key'] is None:
            _config_obj.data['main']['encrypt_key'] = encryption.set_fernet_key().decode('utf-8')

        if _config_obj.data[_section][_key] is not None:
            if _config_obj.data[_section][_key].startswith(ENCRYPT_STRING):
                # encrypted
                _config_obj.data[_section][_key] \
                    = encryption.decrypt(
                    _config_obj.data[_section][_key],
                    _config_obj.data['main']['encrypt_key'])
                if _config_obj.data[_section][_key] is None:
                    _config_obj.logger.error(
                        'Unable to decrypt password. ' +
                        'Try updating password in config file in clear text')
            else:
                # not encrypted
                clear_pwd = _config_obj.data[_section][_key]
                encrypted_pwd = encryption.encrypt(
                    _config_obj.data[_section][_key],
                    _config_obj.data['main']['encrypt_key'])
                _config_obj.write(_section, _key, encrypted_pwd)
                _config_obj.data[_section][_key] = clear_pwd


def set_ip(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] == '0.0.0.0':
        _config_obj.data['web']['bind_ip'] = '0.0.0.0'
        if _config_obj.data['web']['plex_accessible_ip'] == '0.0.0.0':
            _config_obj.data['web']['plex_accessible_ip'] \
                = utils.get_ip()
    else:
        _config_obj.data['web']['bind_ip'] \
            = _config_obj.data[_section][_key]
        if _config_obj.data['web']['plex_accessible_ip'] == '0.0.0.0':
            _config_obj.data['web']['plex_accessible_ip'] \
                = _config_obj.data[_section][_key]


def set_netmask(_config_obj, _section, _key):
    if not _config_obj.data[_section][_key]:
        _config_obj.data[_section][_key] = \
            '{}/32'.format(_config_obj.data['web']['plex_accessible_ip'])


def enable_hdhr(_config_obj, _section, _key):
    if not _config_obj.data[_section][_key]:
        if _config_obj.data['hdhomerun']['udp_netmask'] is None:
            _config_obj.data[_section][_key] = True
            return 'ERROR:: [hdhomerun][udp_netmask] must be set when HDHomeRun is enabled, reverted save'


def enable_ssdp(_config_obj, _section, _key):
    if not _config_obj.data[_section][_key]:
        if _config_obj.data['ssdp']['udp_netmask'] is None:
            _config_obj.data[_section][_key] = True
            return 'ERROR:: [ssdp][udp_netmask] must be set when HDHomeRun is enabled, reverted save'


def set_hdhomerun_id(_config_obj, _section, _key):
    if _config_obj.data[_section][_key] is None:
        _config_obj.write(
            _section, _key, hdhr_server.hdhr_gen_device_id())
    elif not hdhr_server.hdhr_validate_device_id(
            _config_obj.data[_section][_key]):
        _config_obj.write(
            _section, _key, hdhr_server.hdhr_gen_device_id())


def set_uuid(_config_obj, _section, _key):
    if _config_obj.data["main"]["uuid"] is None:
        _config_obj.write('main', 'uuid', str(uuid.uuid1()).upper())


def update_instance_label(_config_obj, _section, _key):
    value = _config_obj.data[_section][_key]
    db_confdefn = DBConfigDefn(_config_obj.data)
    areas = db_confdefn.get_area_by_section(_section)
    if len(areas) > 1:
        results = 'WARNING: There is more than one section named {}'.format(_section)
    elif len(areas) == 0:
        return
    else:
        results = None
    section_data = db_confdefn.get_one_section_dict(areas[0], _section)
    section_data[_section]['label'] = value
    db_confdefn.add_section(areas[0], _section, section_data[_section])
    # when instance label is updated, all tasks for that instance are removed
    # a restart is needed. Note added to config label
    db_scheduler = DBScheduler(_config_obj.data)
    namespace, instance = _section.split('_', 1)
    tasks = db_scheduler.get_tasks_by_name(namespace, instance)
    for task in tasks:
        WebHTTPHandler.sched_queue.put({'cmd': 'deltask', 'taskid': task['taskid']})
    return results


def update_channel_num(_config_obj, _section, _key):
    starting_num = _config_obj.data[_section][_key]
    init_num = starting_num
    is_changed = False
    namespace_l, instance = _section.split('_', 1)
    db_channels = DBChannels(_config_obj.data)
    namespaces = db_channels.get_channel_names()
    namespace = {n['namespace']: n for n in namespaces if namespace_l == n['namespace'].lower()}.keys()
    if len(namespace) == 0:
        return 'ERROR: Bad namespace'
    namespace = list(namespace)[0]
    ch_list = db_channels.get_channels(namespace, instance)
    for ch in ch_list.values():
        if starting_num != -1:
            if ch[0]['display_number'] != starting_num:
                ch[0]['display_number'] = starting_num
                db_channels.update_channel_number(ch[0])
            starting_num += 1

    if init_num == -1:
        return 'Renumbered channels back to default'.format(_section, _key)
    else:
        return 'Renumbered channels starting at {}'.format(_section, _key, init_num)


def set_theme_folders(_defn, _config, _section, _key):
    """
    To make this work, the themes folder must be static and known
    since theme folder list is set before the paths are initialized
    """
    theme_list = []
    themes_path = 'lib.web.htdocs.modules.themes'
    for folder in sorted(importlib.resources.contents(themes_path)):
        if folder.startswith('__'):
            continue
        try:
            importlib.resources.read_text(themes_path, folder)
        except (IsADirectoryError, PermissionError):
            theme_list.append(folder)
        except UnicodeDecodeError:
            continue
    _defn['general']['sections']['display']['settings']['theme']['values'] = theme_list
    theme_default = _defn['general']['sections']['display']['settings']['theme']['default']
    if theme_default not in theme_list:
        _defn['general']['sections']['display']['settings']['theme']['default'] = theme_list[0]