WorkbenchConfig.py

"""Class to encapsulate Workbench configuration definitions."""

import shlex
import tempfile
from getpass import getpass
from workbench_utils import *
from rich.console import Console
from rich.table import Table


class WorkbenchConfig:
    def __init__(self, args):
        self.args = args
        self.args.config = os.path.expanduser(self.args.config)
        self.user_mods = self.get_user_config()
        if "password" not in self.user_mods and "username" in self.user_mods:
            self.user_mods["password"] = self.get_credentials()
        if "password" not in self.user_mods and "username" not in self.user_mods:
            credentials_from_file = self.get_credentials()
            self.user_mods["username"] = credentials_from_file[0]
            self.user_mods["password"] = credentials_from_file[1]
        self.path_check()
        self.config = self.get_config()
        logging.basicConfig(
            filename=self.config["log_file_path"],
            level=logging.INFO,
            filemode="a",
            format="%(asctime)s - %(levelname)s - %(message)s",
            datefmt="%d-%b-%y %H:%M:%S",
        )

    def get_credentials(self):
        """If the password is not set in the config file, in the credentials file, or in the
        environment variable, prompt the user for the password. If neither the username or
        password are set in the config file, look for a credentials file at the configured
        'credentials_file_path' location.

        Returns
        -------
        string|tuple
            The password from the user prompt or ISLANDORA_WORKBENCH_PASSWORD environment variable,
            or if the "credentials_file_path" config setting is present, a tuple containing the
            username and password from the specified credentials file.
        """
        yaml = YAML()
        with open(self.args.config, "r") as stream:
            try:
                password_cfg = yaml.load(stream)
            except YAMLError as exc:
                message = (
                    f"There appears to be a YAML syntax error in your configuration file, {self.args.config}. Remove "
                    f"the username and\npassword, and run the file through https://codebeautify.org/yaml-validator/ "
                    f"or your YAML validator of choice.\n"
                    f"(Check 'workbench.log' file for additional error details.)"
                )
                logging.error(message)
                sys.exit(message)

        if "password" not in password_cfg:
            if "ISLANDORA_WORKBENCH_PASSWORD" in os.environ:
                return os.environ["ISLANDORA_WORKBENCH_PASSWORD"]
            elif (
                "username" not in password_cfg
                and "credentials_file_path" in password_cfg
            ):
                credentials_file_path = os.path.expanduser(
                    password_cfg["credentials_file_path"]
                )
                if os.path.exists(credentials_file_path) is False:
                    message = (
                        'Error: Credentials file "'
                        + credentials_file_path
                        + '" not found.'
                    )
                    logging.error(message)
                    sys.exit(message)
                else:
                    # We open the file to determine how many lines it contains.
                    with open(credentials_file_path, "r") as stream:
                        tmp_credentials_content_file = stream.read()
                        tmp_credentials_content = (
                            tmp_credentials_content_file.splitlines()
                        )
                        if len(tmp_credentials_content) == 1:
                            # It's likely encrypted if it contains only a single line,
                            # since if it isn't, we're expecting two lines.
                            encrypted = True
                        else:
                            encrypted = False

                    # We open the file here to read its YAML.
                    credentials_yaml = YAML()
                    with open(credentials_file_path, "r") as stream:
                        try:
                            if encrypted is True:
                                if "credentials_key_file_path" in password_cfg:
                                    credentials_key_file_path = os.path.abspath(
                                        os.path.expanduser(
                                            password_cfg["credentials_key_file_path"]
                                        )
                                    )
                                else:
                                    credentials_key_file_path = None
                                decrypted_credentials = self.decrypt_credentials_file(
                                    os.path.abspath(credentials_file_path),
                                    credentials_key_file_path,
                                )
                                credentials = credentials_yaml.load(
                                    decrypted_credentials
                                )
                                return (
                                    credentials["username"],
                                    credentials["password"],
                                )
                            else:
                                credentials = credentials_yaml.load(stream)
                                return (
                                    credentials["username"],
                                    credentials["password"],
                                )
                        except YAMLError as exc:
                            print(
                                f"There appears to be a YAML syntax error in your credentials file, {credentials_file_path}. See workbench.log for details."
                            )
                            logging.basicConfig(
                                filename="workbench.log",
                                format="%(asctime)s - %(levelname)s - %(message)s",
                                datefmt="%d-%b-%y %H:%M:%S",
                            )
                            yaml_error = f"\nYAML parsing error in credentials file\nException type: {type(exc).__name__}\n{exc}"
                            logging.error(yaml_error)
                            sys.exit()
            else:
                try:
                    password = getpass(
                        f"Password for Drupal user {password_cfg['username']} (ctrl-c to exit):"
                    )
                    return password
                except KeyboardInterrupt:
                    print("\nExiting.")
                    sys.exit(0)

    def decrypt_credentials_file(
        self, path_to_credentials_file, path_to_credentials_key_file_path
    ):
        """Decrypt the credentials file.
        Parameters
        :param path_to_credentials_file: string - The absolute ath to the credentials file.
        :param path_to_credentials_key_file_path: string - The absolute path to the credentials key file, or None if the path is not defined in the Workbench configuration.
        :return: str - The decrypted file's contents.
        """
        try:
            if path_to_credentials_key_file_path is not None:
                # Check that key file path exists and is readable.
                if os.path.exists(path_to_credentials_key_file_path) is False:
                    message = (
                        'Error: Credentials file "'
                        + path_to_credentials_key_file_path
                        + '" not found.'
                    )
                    logging.error(message)
                    sys.exit(message)

                # Get the value in the key file.
                with open(path_to_credentials_key_file_path, "rb") as f:
                    encryption_key = f.read().strip()
            elif "ISLANDORA_WORKBENCH_ENCRYPTION_KEY" in os.environ:
                encryption_key = os.environ["ISLANDORA_WORKBENCH_ENCRYPTION_KEY"]
            else:
                try:
                    encryption_key = getpass(
                        "Enter the encryptionn key for your credentials file (ctrl-c to exit): "
                    )
                except KeyboardInterrupt:
                    print("\nExiting.")
                    sys.exit(0)
            fernet = Fernet(encryption_key)
            with open(path_to_credentials_file, "rb") as f:
                encrypted_credentials = f.read()
            decrypted_credentials = (
                fernet.decrypt(encrypted_credentials).decode().strip()
            )
            return decrypted_credentials
        except Exception as e:
            message = (
                f"Provided encryption key cannot decrypt the credentials file: {e}"
            )
            logging.error(message)
            sys.exit(message)

    # Get fully constructed config dictionary.
    def get_config(self):
        try:
            config = self.get_default_config()
            # user_mods = self.get_user_config()
        except KeyboardInterrupt:
            try:
                sys.exit(0)
            except SystemExit:
                os._exit(0)

        # Blend defaults with user mods
        for key, value in self.user_mods.items():
            if isinstance(value, dict):
                if key not in config:
                    config[key] = {}
                for subkey, subvalue in value.items():
                    config[key][subkey] = subvalue
            else:
                config[key] = value

        # Modify some conditional values.
        if "temp_dir" not in self.user_mods.keys():
            config["temp_dir"] = tempfile.gettempdir()
        if self.user_mods["task"] in ["add_media", "update", "delete", "export_csv"]:
            config["id_field"] = "node_id"
        elif config["task"] == "delete_media":
            config["id_field"] = "media_id"
        if self.user_mods["task"] == "create_terms":
            config["id_field"] = "term_name"
            config["allow_adding_terms"] = True
        # @todo: These two overrides aren't working. For now, they are set within workbench.update_terms().
        elif config["task"] == "update_terms":
            config["id_field"] = "term_id"

        config["host"] = config["host"].rstrip("/")
        if "csv_id_to_node_id_map_allowed_hosts" not in config:
            config["csv_id_to_node_id_map_allowed_hosts"] = ["", config["host"]]
        config["current_config_file_path"] = os.path.abspath(self.args.config)
        config["field_text_format_ids"] = self.get_field_level_text_output_formats()

        if "csv_id_to_node_id_map_path" not in config:
            config["csv_id_to_node_id_map_path"] = os.path.join(
                config["csv_id_to_node_id_map_dir"],
                config["csv_id_to_node_id_map_filename"],
            )

        if "path_to_python" not in config:
            config["path_to_python"] = sys.executable

        if "page_files_source_dir_field" not in config:
            config["page_files_source_dir_field"] = config["id_field"]

        if "paged_content_page_content_type" not in config:
            config["paged_content_page_content_type"] = config["content_type"]

        config["config_file"] = self.args.config

        _expanduser_keys = [
            "input_dir",
            "input_csv",
            "log_file_path",
            "output_csv",
            "credentials_file_path",
            "credentials_key_file_path",
            "export_csv_file_path",
            "export_file_directory",
            "contact_sheet_output_dir",
            "contact_sheet_css_path",
            "rollback_dir",
            "csv_id_to_node_id_map_dir",
            "csv_id_to_node_id_map_path",
            "sqlite_db_filename",
            "path_to_workbench_script",
            "path_to_python",
            "check_lock_file_path",
            "rollback_csv_file_path",
            "rollback_config_file_path",
        ]
        for _key in _expanduser_keys:
            if _key in config and isinstance(config[_key], str):
                config[_key] = os.path.expanduser(config[_key])

        def _expand_script(s):
            if "~" not in s:
                return s
            try:
                tokens = shlex.split(s)
            except ValueError:
                return s
            expanded = [
                os.path.expanduser(t) if t.startswith("~") else t for t in tokens
            ]
            return " ".join(expanded)

        _list_script_keys = [
            "bootstrap",
            "shutdown",
            "node_post_create",
            "node_post_update",
            "media_post_create",
            "run_scripts",
        ]
        for _key in _list_script_keys:
            if _key in config and isinstance(config[_key], list):
                config[_key] = [
                    _expand_script(s) if isinstance(s, str) else s for s in config[_key]
                ]

        if "preprocessors" in config and isinstance(config["preprocessors"], list):
            for _entry in config["preprocessors"]:
                if isinstance(_entry, dict):
                    for _field, _script in _entry.items():
                        if isinstance(_script, str):
                            _entry[_field] = _expand_script(_script)

        return config

    # Get user input as dictionary.
    def get_user_config(self):
        yaml_loader = YAML(typ="safe")
        with open(self.args.config, "r") as stream:
            try:
                loaded = yaml_loader.load(stream)
            except YAMLError as exc:
                print(
                    f"There appears to be a YAML syntax error in your configuration file, {self.args.config}. Remove "
                    f"the username and\npassword, and run the file through https://codebeautify.org/yaml-validator/ "
                    f"or your YAML validator of choice.\n"
                    f"(Check 'workbench.log' file for additional error details.)"
                )
                # We are using logging.basicConfig() here to set a temporary logger here because this method will
                # run inside WorkbenchConfig.__init__() before the WorkbenchConfig class logger is configured.
                # Also by using the logger we don't have to output the YAML parsing errors to the CLI
                logging.basicConfig(
                    filename="workbench.log",
                    format="%(asctime)s - %(levelname)s - %(message)s",
                    datefmt="%d-%b-%y %H:%M:%S",
                )
                yaml_error = f"\nYAML parsing error in configuration file\nException type: {type(exc).__name__}\n{exc}"
                logging.error(yaml_error)

                sys.exit()

        # 'media_file_fields' has been replaced with 'media_fields' and 'media_type_file_fields'.
        # This is aliasing code that can be removed at some point in the future.
        if "media_file_fields" in loaded:
            media_fields = self.get_media_fields()
            for media_field in loaded["media_file_fields"]:
                for media_type, media_field_val in media_field.items():
                    media_fields[media_type] = media_field_val
            loaded["media_fields"] = media_fields
            loaded["media_type_file_fields"] = media_fields
        if os.path.isabs(self.args.config):
            loaded["config_file_path"] = self.args.config
        else:
            loaded["config_file_path"] = os.path.join(os.getcwd(), self.args.config)

        if (
            "export_csv_file_path" in loaded
            and loaded["export_csv_file_path"] is not None
        ):
            if (
                "export_file_directory" in loaded
                and loaded["export_file_directory"] is not None
                and os.path.isabs(loaded["export_file_directory"]) is False
            ):
                loaded["export_file_directory"] = os.path.abspath(
                    loaded["export_file_directory"]
                )

        return loaded

    # Returns standard media fields.
    @staticmethod
    def get_media_fields():
        return dict(
            {
                "file": "field_media_file",
                "document": "field_media_document",
                "image": "field_media_image",
                "audio": "field_media_audio_file",
                "video": "field_media_video_file",
                "extracted_text": "field_media_file",
                "fits_technical_metadata": "field_media_file",
                "remote_video": "field_media_oembed_video",
            }
        )

    # Returns standard media extensions for given media type.
    @staticmethod
    def get_media_types():
        return [
            {"image": ["png", "gif", "jpg", "jpeg"]},
            {"document": ["pdf", "doc", "docx", "ppt", "pptx"]},
            {"file": ["tif", "tiff", "jp2", "zip", "tar", "hocr"]},
            {"audio": ["mp3", "wav", "aac"]},
            {
                "video": [
                    "mp4",
                    "mov",
                    "wmv",
                    "avi",
                    "mts",
                    "flv",
                    "f4v",
                    "swf",
                    "mkv",
                    "webm",
                    "ogv",
                    "mpeg",
                ]
            },
            {"extracted_text": ["txt"]},
        ]

    # Returns standard field name for media track files for given media type.
    @staticmethod
    def get_media_track_file_fields():
        return {"audio": "field_track", "video": "field_track"}

    # Gets the field->text output format mapping dict from the optional 'field_text_format_ids'
    # config setting. If the setting is absent, returns an empty dict.
    def get_field_level_text_output_formats(self):
        user_config = self.get_user_config()
        field_text_output_map = {}
        if "field_text_format_ids" in user_config:
            for map_entry in user_config["field_text_format_ids"]:
                for fieldname, text_format_id in map_entry.items():
                    field_text_output_map[fieldname] = text_format_id
        return field_text_output_map

    # Returns the standard allowed oEmbed provider URLs for a given media type. These
    # are used to identify URLs in the 'file' CSV column as being remote media.
    @staticmethod
    def get_oembed_media_types():
        return [{"remote_video": ["https://www.youtube.com/", "https://youtu.be"]}]

    # Returns default configs, to be updated by user-supplied config.
    def get_default_config(self):
        return {
            "input_dir": "input_data",
            "input_csv": "metadata.csv",
            "media_use_tid": "http://pcdm.org/use#OriginalFile",
            # 'drupal_filesystem' is used only in Drupal 8.x - 9.1; after that,
            # the filesystem is automatically detected from the media's configuration.
            "drupal_filesystem": "fedora://",
            "id_field": "id",
            "content_type": "islandora_object",
            "delimiter": ",",
            "subdelimiter": "|",
            "log_file_path": "workbench.log",
            "log_file_mode": "a",
            "allow_missing_files": False,
            # See Issue 620. Allows the "--check" function to keep running past errors that can be handled
            #  well with batch operations, rather than having to fix it in real-time when one error is hit.
            "perform_soft_checks": False,
            "update_mode": "replace",
            "max_node_title_length": 255,
            "max_image_alt_text_length": 255,
            "paged_content_from_directories": False,
            "paged_content_from_directories_parents_exist": False,
            "delete_media_with_nodes": True,
            "allow_adding_terms": False,
            "columns_with_term_names": [],
            "protected_vocabularies": [],
            "nodes_only": False,
            "log_response_time": False,
            "adaptive_pause_threshold": 2,
            "log_response_time_sample": False,
            "log_request_url": False,
            "log_json": False,
            "log_response_body": False,
            "log_response_status_code": False,
            "log_headers": False,
            "log_term_creation": True,
            "log_file_name_and_line_number": False,
            "progress_bar": False,
            "show_percentage_of_csv_input_processed": False,
            "prompt_user_before_delete_task": False,
            "run_scripts_threads": 1,
            "run_scripts_log_script_output": True,
            "user_agent": "Islandora Workbench",
            "allow_redirects": True,
            "secure_ssl_only": True,
            "google_sheets_csv_filename": "google_sheet.csv",
            "google_sheets_gid": "0",
            "excel_worksheet": "Sheet1",
            "excel_csv_filename": "excel.csv",
            "ignore_csv_columns": list(),
            "use_node_title_for_media": False,
            "use_nid_in_media_title": False,
            "use_node_title_for_media_title": False,
            "field_for_remote_filename": False,
            "remote_file_cookie_name": "",
            "remote_file_cookie_value": "",
            "keep_filename_parent_directory": True,
            "field_for_media_title": False,
            "delete_tmp_upload": False,
            "input_data_zip_archives": [],
            "delete_zip_archive_after_extraction": True,
            "list_missing_drupal_fields": False,
            "secondary_tasks": None,
            "sqlite_db_filename": "workbench_temp_data.db",
            "csv_id_to_node_id_map_dir": tempfile.gettempdir(),
            "csv_id_to_node_id_map_filename": "csv_id_to_node_id_map.db",
            "fixity_algorithm": None,
            "validate_fixity_during_check": False,
            "output_csv_include_input_csv": False,
            "export_file_url_instead_of_download": False,
            "timestamp_rollback": False,
            "rollback_dir": None,
            "rollback_file_include_node_info": False,
            "enable_http_cache": True,
            "http_cache_storage": "memory",
            "http_cache_storage_expire_after": 1200,
            "http_max_retries": 3,
            "http_retry_on_status_codes": [500, 502, 503, 504],
            "http_backoff_factor": 1,
            "http_retry_allowed_methods": [
                "HEAD",
                "GET",
                "POST",
                "PUT",
                "PATCH",
                "DELETE",
            ],
            "validate_terms_exist": True,
            "validate_parent_node_exists": True,
            "media_types": self.get_media_types(),
            "preprocessors": {},
            "check": self.args.check,
            "get_csv_template": self.args.get_csv_template,
            "paged_content_sequence_separator": "-",
            "paged_content_page_weight_multiplier": "1",
            "media_type_file_fields": self.get_media_fields(),
            "media_track_file_fields": self.get_media_track_file_fields(),
            "media_fields": self.get_media_fields(),
            "delete_media_by_node_media_use_tids": [],
            "update_media_by_node_media_use_tids": [],
            "export_csv_term_mode": "tid",
            "export_csv_file_path": None,
            "export_csv_field_list": [],
            "export_file_directory": None,
            "export_file_media_use_term_id": "http://pcdm.org/use#OriginalFile",
            "standalone_media_url": False,
            "require_entity_reference_views": True,
            "csv_start_row": 0,
            "csv_stop_row": None,
            "path_to_workbench_script": os.path.join(os.getcwd(), "workbench"),
            "oembed_providers": self.get_oembed_media_types(),
            "contact_sheet_output_dir": "contact_sheet_output",
            "contact_sheet_css_path": os.path.join(
                "assets", "contact_sheet", "contact-sheet.css"
            ),
            "page_title_template": "$parent_title, page $weight",
            "csv_headers": "names",
            "clean_csv_values_skip": [],
            "text_format_id": "basic_html",
            "ignore_existing_parent_ids": True,
            "query_csv_id_to_node_id_map_for_parents": False,
            "ignore_duplicate_parent_ids": True,
            "redirect_status_code": 301,
            "csv_value_templates": [],
            "csv_value_templates_for_paged_content": [],
            "csv_value_templates_rand_length": 5,
            "allow_csv_value_templates_if_field_empty": [],
            "remind_user_to_run_check": False,
            "completion_message": None,
            "media_type_by_media_use": False,
            "paged_content_ignore_files": ["Thumbs.db"],
            "include_password_in_rollback_config_file": False,
            "remove_password_from_config_file": False,
            "recovery_mode_starting_from_node_id": False,
            "viewer_override_fieldname": "field_viewer_override",
            "check_for_workbench_updates": True,
            "use_workbench_permissions": False,
            "show_shutdown_script_output": False,
            "show_bootstrap_script_output": False,
        }

    # Tests validity and existence of configuration file path.
    def path_check(self):
        # Check existence of configuration file.
        if not os.path.exists(self.args.config):
            # Since the main logger gets its log file location from this file, we
            # need to define a local logger to write to the default log file location,
            # 'workbench.log'.
            logging.basicConfig(
                filename="workbench.log",
                format="%(asctime)s - %(levelname)s - %(message)s",
                datefmt="%d-%b-%y %H:%M:%S",
            )
            message = 'Error: Configuration file "' + self.args.config + '" not found.'
            logging.error(message)
            sys.exit(message)

    # Convenience function for debugging - Prints config to console screen.
    def print_config(self):
        table = Table(title="Workbench Configuration")
        table.add_column("Parameter", justify="left")
        table.add_column("Value", justify="left")
        for key, value in self.config.items():
            table.add_row(key, str(value))
        console = Console()
        console.print(table)