HEX
HEX
Server: Apache/2.4.29 (Ubuntu)
System: Linux 2amigos-php74 5.4.0-1103-aws #111~18.04.1-Ubuntu SMP Tue May 23 20:04:10 UTC 2023 x86_64
User: squarehillcompany.com (1002)
PHP: 7.4.25
Disabled: pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
Upload Files
File: /var/www/vhosts/dial-copper-dev.2amigos.us/dialpad-copper-crm/api/conf/config.py
"""Module with configuration access BL
"""

import os
import re
from enum import Enum
from os.path import dirname, join as path_join
from pathlib import Path
from typing import Union

import yaml
from pycoppercrm.networking import (
    create_session, CopperCrmSession, SessionMode
)

from api.common.crypto import encrypt_content
from api.common.utils import extract_from_json

_SETTING_RE = re.compile(r'^(\w+\.)+\w+$')

_CONFIG_FILE_PATH = path_join(Path(dirname(__file__)), 'config.yml')


def _get_setting(settings: dict, setting_code: str) -> Union[str, list, int]:
    """Gets a setting by code e.g: db.conn_str = settings[db][conn_str]

    Args:
        settings (dict): settings dictionary
        setting_code (str): code of the setting to be splitted by dots

    Returns:
        Union[str, list, int]: Setting value
    """
    return extract_from_json(settings, setting_code)


def _dump_yaml(settings: dict, file_path: str):
    """Dumps a dict into a yml

    Args:
        settings (dict): settings to be written
        file_path (str): yaml file path
    """
    with open(file_path, 'w') as f:
        yaml.dump(settings, f)


def _load_yaml(file_path: str) -> dict:
    """Loads a yaml file

    Args:
        file_path (str): file path

    Returns:
        dict: loaded settings
    """
    with open(file_path, 'r') as f:
        return yaml.load(f.read(), yaml.SafeLoader)


class ConfigScope(Enum):
    """Enumeration with supported config scopes
    """
    # Unit tests
    UNIT_TEST = 1
    # Local or remote server
    SERVER = 2

    @property
    def is_server(self) -> bool:
        """Boolean flag to determine if the config scope is
           on server or unit tests/

           Returns:
                bool -- True if SERVER False otherwise
        """
        return self.value == 2


class ConfigWrapper:
    """Yaml Config accessor"""

    def __init__(self, settings: dict):
        self._settings = settings

    @property
    def config_scope(self) -> ConfigScope:
        """Current configuration scope

        Returns:
            ConfigScope: enumeration of current app running scope
        """
        # Defaulting to server to ensure runtime stability
        app_scope = os.environ.get("DC_APP_SCOPE", ConfigScope.SERVER.name)
        return ConfigScope[app_scope]
        # NOTE: See note on self.set_app_scope
        # return ConfigScope[self.get_setting('app.scope')]

    @property
    def db_cstr_cipher(self) -> str:
        """Encrypted db connection str

        Returns:
            dict: encrypted settings
        """
        scope = self.config_scope
        setting = 'app_crypto.app_server_db' if scope.is_server else 'app.unit_test_db'
        return self.get_setting(setting)

    @property
    def dialpad_oauth_url(self) -> str:
        """Convinience property to access the configured dialpad OAuth URL
        """
        return self.get_setting('dialpad.oauth_url')

    @property
    def dialpad_logout_url(self) -> str:
        """Convinience property to access the configured dialpad Logout URL
        """
        return self.get_setting('dialpad.logout_url')

    @property
    def dialpad_oauth_secret(self) -> str:
        """Convinience property to access the configured dialpad OAuth Secret
        """
        return self.get_setting('dialpad.oauth_secret')

    @property
    def gds_credentials_path(self) -> str:
        """Full path for Google data store credentials"""
        file_dir = Path(dirname(__file__))
        file_name = self.get_setting('app_crypto.app_gds_credentials_file')
        return path_join(file_dir, file_name)

    def get_setting(self, setting_code: str) -> Union[str, list, int]:
        """Gets a setting by code e.g: db.conn_str = settings[db][conn_str]

        Args:
            setting_code (str): code of the setting to be splitted by dots

        Returns:
            Union[str, list, int]: Setting value
        """
        if not _SETTING_RE.match(setting_code):
            raise ValueError(f'Invalid setting code: {setting_code}')

        return _get_setting(self._settings, setting_code)

    def update_setting(self, setting_code: str, new_value: str, encrypt=False) -> str:
        """Updates a setting in the config file

        Args:
            setting_code (str): code or path to the setting
            new_value (str): value to update
            encrypt (bool, optional): Encrypt using master.key. Defaults to False.

        Returns:
            str: old setting value
        """
        setting_path = setting_code.split('.')
        setting = self._settings.get(next(iter(setting_path)))

        for i in range(1, len(setting_path) - 1):
            setting = setting.get(setting_path[i])

        key = setting_path.pop()
        old = setting.get(key)
        setting[key] = new_value if not encrypt else encrypt_content(new_value)
        return old

    def save(self):
        """updates the settings with this class values
        """
        env_settings = {k: v for k, v in self._settings.items() if k != 'app'}
        main_settings = {k: v for k, v in self._settings.items() if k == 'app'}
        env_path = path_join(Path(dirname(__file__)), self.get_setting('app.full_config_file'))

        _dump_yaml(main_settings, _CONFIG_FILE_PATH)
        _dump_yaml(env_settings, env_path)

    def copper_session(self, copper_key: str, user_email: str) -> CopperCrmSession:
        """Builds a CopperCrm session based on the config

        Args:
            copper_key (str): Copper API Key
            user_email (str): Copper user email

        Returns:
            CopperCrmSession: CopperCrmSession
        """
        mode = SessionMode.SANDBOX if self.get_setting('copper.sandbox') else SessionMode.LIVE
        return create_session(copper_key, user_email, mode)

    @staticmethod
    def read_settings() -> dict:
        """Read settings from config.yml path

        Returns:
            dict: config.yml settings
        """
        main_config = _load_yaml(_CONFIG_FILE_PATH)
        file_path = _get_setting(main_config, 'app.full_config_file')
        env_path = path_join(Path(dirname(__file__)), file_path)
        env_config = _load_yaml(env_path)
        return {**main_config, **env_config}

    @classmethod
    def read_config(cls) -> 'ConfigWrapper':
        """Instance from api configuration file

        Returns:
            ConfigWrapper: App Settings accessor
        """
        return cls(cls.read_settings())


def set_app_scope(scope: ConfigScope):
    """Sets an scope in the config file

    Args:
        scope (ConfigScope): Running app scope
    """
    # NOTE: Google cloud doesn't support updates on local FS
    # Using env variables for this.
    os.environ["DC_APP_SCOPE"] = scope.name
    # wrapper = ConfigWrapper.read_config()
    # wrapper.update_setting('app.scope', scope.name)
    # wrapper.save()