Repository: mjordan/islandora_workbench

Issue:
Reformat Workbench fields to use more inheritance
It would be good to add a super class so you can start adding type hinting to ensure the correct "thing" is returned and can be used more abstractly.

Discussion:

Solution patch:
diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml
index 2c569d9..d72da4c 100644
--- a/.github/workflows/integration-test.yml
+++ b/.github/workflows/integration-test.yml
@@ -60,8 +60,7 @@ jobs:
       - name: Run islandora_tests  
         run: python tests/islandora_tests.py
 
-        # See https://github.com/mjordan/islandora_workbench/issues/890
-        # - name: Run islandora_tests_recovery_mode
-          # run: python tests/islandora_tests_recovery_mode.py
-          # env:
-          # REQUESTS_CA_BUNDLE: ${{ github.workspace }}/isle-site-template/certs/rootCA.pem        
+      - name: Run islandora_tests_recovery_mode
+        run: python tests/islandora_tests_recovery_mode.py
+        env:
+          REQUESTS_CA_BUNDLE: ${{ github.workspace }}/isle-site-template/certs/rootCA.pem        
diff --git a/README.md b/README.md
index fef2aa0..af85ae0 100644
--- a/README.md
+++ b/README.md
@@ -9,6 +9,7 @@ Note that this tool is not related in any way to the Drupal contrib module calle
 * Allows creation of Islandora nodes and media, updating of nodes, and deletion of nodes and media from CSV files
 * Allows creation of paged/compound content
 * Can run from anywhere - it communicates with Drupal via HTTP interfaces
+* Can be built as a Docker container, and run from within that container
 * Provides robust data validation functionality
 * Supports a variety of Drupal entity field types (text, integer, term reference, typed relation, geolocation)
 * Can provide a CSV file template based on Drupal content type
@@ -17,7 +18,7 @@ Note that this tool is not related in any way to the Drupal contrib module calle
 * Allows creation of new taxonomy terms from CSV field data, including complex and hierarchical terms
 * Allows the assignment of URL aliases
 * Allows creation of URL redirects
-* Allows adding alt text to images
+* Allows adding alt text to images (and updating exisiting alt text)
 * Supports transmission fixity auditing for media files
 * Cross platform (written in Python, tested on Linux, Mac, and Windows)
 * Well tested
diff --git a/WorkbenchConfig.py b/WorkbenchConfig.py
index 8d29b14..f31bfeb 100644
--- a/WorkbenchConfig.py
+++ b/WorkbenchConfig.py
@@ -211,6 +211,7 @@ class WorkbenchConfig:
             "perform_soft_checks": False,
             "update_mode": "replace",
             "max_node_title_length": 255,
+            "max_image_alt_text_length": 255,
             "paged_content_from_directories": False,
             "delete_media_with_nodes": True,
             "allow_adding_terms": False,
@@ -254,6 +255,7 @@ class WorkbenchConfig:
             "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,
@@ -317,6 +319,7 @@ class WorkbenchConfig:
             "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",
         }
 
     # Tests validity and existence of configuration file path.
diff --git a/scripts/make_local_tests.py b/scripts/make_local_tests.py
index 837cc8a..883e2a8 100644
--- a/scripts/make_local_tests.py
+++ b/scripts/make_local_tests.py
@@ -15,6 +15,7 @@ import sys
 import shutil
 import glob
 import argparse
+import re
 
 current_dir = os.getcwd()
 path_to_workbench = os.path.join(current_dir, "workbench")
@@ -55,7 +56,7 @@ for filepath in list(glob.iglob(f"{local_tests_dir}/**/*.yml", recursive=True))
     config = f.read()
     config = config.replace("https://islandora.dev", args.host)
     config = config.replace("admin", args.username)
-    config = config.replace("password", args.password)
+    config = re.sub("password$", args.password, config, 0, re.MULTILINE)
     config = config.replace("tests/assets/", "tests_local/assets/")
     f = open(filepath, "w")
     f.write(config)
diff --git a/setup.py b/setup.py
index 5bb533e..c5aaf11 100644
--- a/setup.py
+++ b/setup.py
@@ -21,6 +21,6 @@ setup(
         "typing-extensions",
         "rich",
     ],
-    python_requires=">=3.7",
+    python_requires=">=3.9",
     py_modules=[],
 )
diff --git a/workbench_export.py b/workbench_export.py
new file mode 100644
index 0000000..0c76513
--- /dev/null
+++ b/workbench_export.py
@@ -0,0 +1,547 @@
+import os
+import sys
+import json
+import logging
+import datetime
+import collections
+import requests_cache
+from workbench_utils import *
+from progress_bar import InitBar
+
+
+class WorkbenchExportBase:
+    def __init__(self, config, args=None):
+        self.config = config
+        self.args = args
+        self.field_definitions = get_field_definitions(config, "node")
+        self.required_fields = ["created", "uid", "langcode", "title", "node_id"]
+        self.seen_nids = set()
+        self.pbar = InitBar() if config.get("progress_bar") else None
+
+    @staticmethod
+    def deduplicate_list(input_list):
+        """Remove duplicates while preserving order."""
+        seen = set()
+        return [x for x in input_list if not (x in seen or seen.add(x))]
+
+    def initialize_csv_writer(self, csv_file, field_names, export_mode=False):
+        """Initialize CSV writer with optional metadata rows for export mode."""
+        writer = csv.DictWriter(csv_file, fieldnames=field_names, lineterminator="\n")
+        writer.writeheader()
+
+        if export_mode and self.field_definitions:
+            self._write_metadata_rows(writer, field_names)
+
+        return writer
+
+    def _write_metadata_rows(self, writer, field_names):
+        """Write field labels and cardinality rows for export mode."""
+        field_labels = collections.OrderedDict()
+        for field_name in field_names:
+            if (
+                field_name in self.field_definitions
+                and self.field_definitions[field_name]["label"] != ""
+            ):
+                field_labels[field_name] = self.field_definitions[field_name]["label"]
+            elif field_name == "REMOVE THIS COLUMN (KEEP THIS ROW)":
+                field_labels[field_name] = "LABEL (REMOVE THIS ROW)"
+            else:
+                field_labels[field_name] = ""
+        writer.writerow(field_labels)
+
+        cardinality = collections.OrderedDict()
+        cardinality["REMOVE THIS COLUMN (KEEP THIS ROW)"] = (
+            "NUMBER OF VALUES ALLOWED (REMOVE THIS ROW)"
+        )
+        cardinality["node_id"] = "1"
+        cardinality["uid"] = "1"
+        cardinality["langcode"] = "1"
+        cardinality["created"] = "1"
+        cardinality["title"] = "1"
+
+        for field_name in self.field_definitions:
+            if field_name in field_names:
+                cardinality[field_name] = (
+                    "unlimited"
+                    if self.field_definitions[field_name]["cardinality"] == -1
+                    else str(self.field_definitions[field_name]["cardinality"])
+                )
+        writer.writerow(cardinality)
+
+    def process_file_data(self, node_id, media_use_term_id=None, media_list=None):
+        """Handle file processing (download or URL) based on config."""
+        if self.config.get("export_file_url_instead_of_download", False):
+            result = get_media_file_url(
+                self.config,
+                node_id,
+                media_use_term_id=media_use_term_id,
+                media_list=media_list,
+            )
+        else:
+            result = download_file_from_drupal(
+                self.config,
+                node_id,
+                media_use_term_id=media_use_term_id,
+                media_list=media_list,
+            )
+        return result if result else ""  # Avoid 'False' values in file columns.
+
+    def log_progress(
+        self, message, row_count=None, total_rows=None, level=logging.INFO
+    ):
+        """Standardized progress logging with optional progress bar and log levels.
+
+        Args:
+            message: The message to log
+            row_count: Current row count (for progress bar)
+            total_rows: Total rows (for progress bar)
+            level: Logging level (e.g., logging.INFO, logging.WARNING, logging.ERROR)
+        """
+        if (
+            self.config.get("progress_bar")
+            and self.pbar
+            and row_count is not None
+            and total_rows is not None
+        ):
+            self.pbar(get_percentage(row_count, total_rows))
+        else:
+            print(f"{logging.getLevelName(level)}: {message}")
+
+        # Log to file with appropriate level
+        if level == logging.INFO:
+            logging.info(message)
+        elif level == logging.WARNING:
+            logging.warning(message)
+        elif level == logging.ERROR:
+            logging.error(message)
+        elif level == logging.DEBUG:
+            logging.debug(message)
+        else:
+            logging.info(message)  # default to info if unknown level
+
+    def row_log_suffix(self, row):
+        and_files = ""
+        if self.needs_file_column:
+            if self.config.get("export_file_url_instead_of_download", False):
+                and_files = " and file URL(s)"
+            else:
+                and_files = " and file(s)"
+
+        return and_files
+
+    def extract_node_id(self, node):
+        """Extract and validate node ID."""
+        try:
+            return node["nid"][0]["value"]
+        except (KeyError, IndexError):
+            message = "Skipping node with missing/invalid NID"
+            self.log_progress(message, level=logging.WARNING)
+            return None
+
+    def validate_content_type(self, node, nid):
+        """Verify node matches configured content type."""
+        try:
+            node_type = node["type"][0]["target_id"]
+        except (KeyError, IndexError):
+            node_type = "unknown"
+
+        if node_type != self.config["content_type"]:
+            message = (
+                f"Node {nid} not written to output CSV because its content type {node_type}"
+                + f' does not match the "content_type" configuration setting.'
+            )
+            self.log_progress(message, level=logging.ERROR)
+            return False
+        return True
+
+    def parse_json_response(self, response):
+        """Safely parse JSON from response."""
+        try:
+            return json.loads(response.text)
+        except json.decoder.JSONDecodeError as e:
+            message = f"Failed to decode JSON: {str(e)}"
+            self.log_progress(message, level=logging.WARNING)
+            return []
+
+    def needs_file_column(self):
+        """Check if we need file column in CSV."""
+        return self.config.get(
+            "export_file_url_instead_of_download", False
+        ) or self.config.get("export_file_directory")
+
+    def get_additional_files_config(self):
+        """Get additional files configuration as OrderedDict."""
+        if "additional_files" not in self.config:
+            return collections.OrderedDict()
+
+        additional_files = collections.OrderedDict()
+        for entry in self.config["additional_files"]:
+            for col_name, media_use_uri in entry.items():
+                additional_files[col_name] = media_use_uri
+        return additional_files
+
+    def execute_post_export_script(self, response, node_json):
+        """Execute node-specific post-export scripts, if any are configured."""
+        if len(self.config.get("node_post_export", [])) > 0:
+            for command in self.config.get("node_post_export", []):
+                (
+                    post_task_output,
+                    post_task_return_code,
+                ) = execute_entity_post_task_script(
+                    command,
+                    self.args.config,
+                    response.status_code,
+                    node_json,
+                )
+                if post_task_return_code == 0:
+                    logging.info(
+                        "Post node export script " + command + " executed successfully."
+                    )
+                else:
+                    logging.error("Post node export script " + command + " failed.")
+
+
+class CSVExporter(WorkbenchExportBase):
+    def __init__(self, config, args=None):
+        super().__init__(config, args)
+        self.csv_data = get_csv_data(config)
+        self.required_fields = [
+            "created",
+            "uid",
+            "langcode",
+            "title",
+            "node_id",
+            "REMOVE THIS COLUMN (KEEP THIS ROW)",
+        ]
+
+    def setup_csv_output_path(self):
+        """Set up CSV output path for CSV export."""
+        if self.config["export_csv_file_path"]:
+            csv_path = self.config["export_csv_file_path"]
+        else:
+            csv_path = os.path.join(
+                self.config["input_dir"],
+                self.config["input_csv"] + ".csv_file_with_field_values",
+            )
+
+        if os.path.exists(csv_path):
+            os.remove(csv_path)
+        return csv_path
+
+    def prepare_headers(self):
+        """Generate deduplicated list of CSV column headers for export_csv."""
+        field_names = list(self.field_definitions.keys())
+
+        if len(self.config["export_csv_field_list"]) > 0:
+            field_names = self.config["export_csv_field_list"]
+
+        # Add required fields at beginning
+        for field_name in self.required_fields:
+            field_names.insert(0, field_name)
+
+        deduped_field_names = self.deduplicate_list(field_names)
+
+        # We always include 'node_id and 'REMOVE THIS COLUMN (KEEP THIS ROW)'.
+        if "node_id" not in deduped_field_names:
+            deduped_field_names.insert(0, "node_id")
+            deduped_field_names.insert(0, "REMOVE THIS COLUMN (KEEP THIS ROW)")
+
+        # Handle file columns
+        if self.needs_file_column() and "file" not in deduped_field_names:
+            deduped_field_names.append("file")
+
+        # Add additional files columns
+        additional_files_entries = self.get_additional_files_config()
+        if additional_files_entries:
+            for column in additional_files_entries.keys():
+                if column not in deduped_field_names:
+                    deduped_field_names.append(column)
+
+        return deduped_field_names
+
+    def validate_and_get_node_id(self, csv_row):
+        """Validate and get node ID from CSV row."""
+        node_id = csv_row["node_id"]
+        if not value_is_numeric(node_id):
+            node_id = get_nid_from_url_alias(self.config, node_id)
+
+        if not ping_node(self.config, node_id):
+            self.log_progress(
+                f"Node {node_id} not found/accessible, skipping export.",
+                level=logging.WARNING,
+            )
+            return None
+
+        return node_id
+
+    def fetch_node_json(self, node_id):
+        """Fetch node JSON from Drupal."""
+        url = f"{self.config['host']}/node/{node_id}?_format=json"
+        response = issue_request(self.config, "GET", url)
+
+        if response.status_code != 200:
+            self.log_progress(
+                f"Error retrieving node {node_id}: HTTP {response.status_code}",
+                level=logging.WARNING,
+            )
+            return None
+
+        return json.loads(response.text)
+
+    def process_node_fields_to_row(self, node_json, field_names):
+        """Process node fields into a dictionary for CSV output."""
+        row = collections.OrderedDict()
+        nid = self.extract_node_id(node_json)
+
+        if not nid:
+            return None
+
+        row["node_id"] = nid
+        row["title"] = node_json.get("title", [{}])[0].get("value", "No title")
+
+        # Process fields
+        for field in field_names:
+            if field.startswith("field_") and field in node_json:
+                try:
+                    row[field] = serialize_field_json(
+                        self.config, self.field_definitions, field, node_json[field]
+                    )
+                except Exception as e:
+                    self.log_progress(
+                        f"Error serializing {field} for node {nid}: {str(e)}",
+                        level=logging.ERROR,
+                    )
+                    row[field] = "SERIALIZATION_ERROR"
+
+        return row
+
+    def process_node_row(self, node_json, field_names):
+        """Common processing for node rows."""
+        row = self.process_node_fields_to_row(node_json, field_names)
+        if not row:
+            return None
+
+        media_list = get_media_list(self.config, row["node_id"])
+        additional_files_entries = self.get_additional_files_config()
+
+        if self.needs_file_column():
+            row["file"] = self.process_file_data(row["node_id"], media_list=media_list)
+
+        if additional_files_entries:
+            for col_name, media_use_uri in additional_files_entries.items():
+                row[col_name] = self.process_file_data(
+                    row["node_id"],
+                    media_use_term_id=media_use_uri,
+                    media_list=media_list,
+                )
+
+        return row
+
+    def export(self):
+        """Main export method for CSV export."""
+        csv_file_path = self.setup_csv_output_path()
+        field_names = self.prepare_headers()
+
+        with open(csv_file_path, "a+", encoding="utf-8") as csv_file:
+            writer = self.initialize_csv_writer(csv_file, field_names, True)
+            self._process_nodes(writer, field_names)
+
+        return csv_file_path
+
+    def _process_nodes(self, writer, field_names):
+        """Process all nodes for CSV export."""
+        csv_data_list = list(self.csv_data)
+        row_count = 0
+
+        for row in csv_data_list:
+            # Delete expired items from request_cache before processing a row.
+            if self.config["enable_http_cache"]:
+                requests_cache.delete(expired=True)
+
+            node_id = self.validate_and_get_node_id(row)
+            if not node_id:
+                continue
+
+            node_json = self.fetch_node_json(node_id)
+            if not node_json or not self.validate_content_type(node_json, node_id):
+                continue
+
+            output_row = self.process_node_row(node_json, field_names)
+            if not output_row:
+                continue
+
+            writer.writerow(output_row)
+            suffix = self.row_log_suffix(row)
+
+            self.log_progress(
+                f'Exporting data{suffix} for node {node_id} "{output_row["title"]}."',
+                row_count,
+                len(csv_data_list),
+            )
+            row_count += 1
+
+
+class ViewExporter(WorkbenchExportBase):
+    def __init__(self, config, args):
+        super().__init__(config, args)
+        self.view_config = self.initialize_view_config()
+
+    def setup_csv_output_path(self):
+        """Set up CSV output path with timestamp for view exports."""
+        if self.config["export_csv_file_path"]:
+            return self.config["export_csv_file_path"]
+
+        config_base = os.path.basename(self.args.config).split(".")[0]
+        csv_path = os.path.join(
+            self.config["input_dir"],
+            f"{config_base}_view_export_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}.csv",
+        )
+
+        if os.path.exists(csv_path):
+            os.remove(csv_path)
+        return csv_path
+
+    def initialize_view_config(self):
+        """Initialize configuration for View export."""
+        view_parameters = (
+            "&".join(self.config["view_parameters"])
+            if "view_parameters" in self.config
+            else ""
+        )
+        return {
+            "base_url": f"{self.config['host']}/{self.config['view_path'].lstrip('/')}",
+            "parameters": view_parameters,
+            "initial_url": f"{self.config['host']}/{self.config['view_path'].lstrip('/')}?page=0&{view_parameters}",
+        }
+
+    def verify_view_accessibility(self):
+        """Verify that the View endpoint is accessible."""
+        status_code = ping_view_endpoint(self.config, self.view_config["initial_url"])
+        if status_code != 200:
+            message = f"Cannot access View at {self.view_config['initial_url']} (HTTP {status_code})."
+            self.log_progress(message, level=logging.ERROR)
+            sys.exit("Error: " + message + " See log for more information.")
+
+    def prepare_headers(self):
+        """Generate deduplicated list of CSV column headers."""
+        fields = ["node_id", "title"]
+
+        if self.config.get("export_csv_field_list"):
+            fields += [
+                f for f in self.config["export_csv_field_list"] if f not in fields
+            ]
+        else:
+            fields += [
+                f for f in self.field_definitions.keys() if f.startswith("field_")
+            ]
+
+        if self.needs_file_column():
+            fields.append("file")
+
+        if "additional_files" in self.config:
+            fields += self.get_additional_files_config().keys()
+
+        return self.deduplicate_list(fields)
+
+    def process_node_fields_to_row(self, node_json, field_names):
+        """Process node fields into a dictionary for CSV output."""
+        row = collections.OrderedDict()
+        nid = self.extract_node_id(node_json)
+
+        if not nid:
+            return None
+
+        row["node_id"] = nid
+        row["title"] = node_json.get("title", [{}])[0].get("value", "No title")
+
+        # Process fields
+        for field in field_names:
+            if field.startswith("field_") and field in node_json:
+                try:
+                    row[field] = serialize_field_json(
+                        self.config, self.field_definitions, field, node_json[field]
+                    )
+                except Exception as e:
+                    self.log_progress(
+                        f"Error serializing {field} for node {nid}: {str(e)}",
+                        level=logging.ERROR,
+                    )
+                    row[field] = "SERIALIZATION_ERROR"
+
+        return row
+
+    def process_node_row(self, node_json, field_names):
+        """Process node data into CSV row."""
+        row = self.process_node_fields_to_row(node_json, field_names)
+        if not row:
+            return None
+
+        media_list = get_media_list(self.config, row["node_id"])
+        additional_files_entries = self.get_additional_files_config()
+
+        if self.needs_file_column():
+            row["file"] = self.process_file_data(row["node_id"], media_list=media_list)
+
+        if additional_files_entries:
+            for col_name, media_use_uri in additional_files_entries.items():
+                row[col_name] = self.process_file_data(
+                    row["node_id"],
+                    media_use_term_id=media_use_uri,
+                    media_list=media_list,
+                )
+
+        return row
+
+    def export(self):
+        """Main export method for View export."""
+        self.verify_view_accessibility()
+        csv_file_path = self.setup_csv_output_path()
+        field_names = self.prepare_headers()
+
+        with open(csv_file_path, "a+", encoding="utf-8") as csv_file:
+            writer = self.initialize_csv_writer(csv_file, field_names)
+            self._process_view_pages(writer, field_names)
+
+        return csv_file_path
+
+    def _process_view_pages(self, writer, field_names):
+        """Process paginated View results."""
+        page = 0
+
+        while True:
+            current_url = f"{self.view_config['base_url']}?page={page}&{self.view_config['parameters']}"
+            response = issue_request(self.config, "GET", current_url)
+
+            if response.status_code != 200:
+                self.log_progress(
+                    f"Skipping page {page} due to HTTP {response.status_code}",
+                    level=logging.WARNING,
+                )
+                page += 1
+                continue
+
+            nodes = self.parse_json_response(response)
+            if not nodes:
+                break
+
+            for node in nodes:
+                if self.config.get("enable_http_cache", False):
+                    requests_cache.delete(expired=True)
+
+                nid = self.extract_node_id(node)
+                if not nid or nid in self.seen_nids:
+                    continue
+
+                self.seen_nids.add(nid)
+                if not self.validate_content_type(node, nid):
+                    continue
+
+                row = self.process_node_row(node, field_names)
+                if row:
+                    writer.writerow(row)
+
+                    suffix = self.row_log_suffix(row)
+                    self.log_progress(f"Exported node{suffix} {nid}: {row['title']}")
+                    self.execute_post_export_script(response, json.dumps(node))
+
+            page += 1
diff --git a/workbench_fields.py b/workbench_fields.py
index 4ef4837..7369b49 100644
--- a/workbench_fields.py
+++ b/workbench_fields.py
@@ -3,28 +3,185 @@
 Support for additional field types should be added as new classes here,
 with accompanying tests in field_tests.py and field_tests_values.py.
 
-Note: If new field types are added to workbench_fields.py, corresponding logic must
-be added to functions in other Workbench modules (e.g. workbench_utils, workbench)
-that create, update, or export Drupal entities. Those places are commented in the
-code with either:
+Note: If new field types are added to workbench_fields.py, you must also update the WorkbenchFieldFactory
+class at the bottom of this file to include the new field type and its corresponding class. This is used
+by workbench and workbench_utils.py to instantiate the correct class for each field type.
 
-# Assemble Drupal field structures from CSV data. If new field types are added to
-# workbench_fields.py, they need to be registered in the following if/elif/else block.
-
-or
-
-# Assemble CSV output Drupal field data. If new field types are added to
-# workbench_fields.py, they need to be registered in the following if/elif/else block.
 """
 
-import json
+from collections import OrderedDict
+
 from workbench_utils import *
 
 
-class SimpleField:
+class WorkbenchField:
+    """Base class for field operations. All subclasses must implement the create(),
+    update(), remove_invalid_values(), and serialize() methods. Subclasses may also
+    implement the split_string() method, which is called in the common create()."""
+
+    def create(
+        self,
+        config: dict,
+        field_definitions: dict,
+        entity: dict,
+        row: OrderedDict,
+        field_name: str,
+    ) -> dict:
+        """Parameters
+        ----------
+         config : dict
+             The configuration settings defined by workbench_config.get_config().
+         field_definitions : dict
+             The field definitions object defined by get_field_definitions().
+         entity : dict
+             The dict that will be POSTed to Drupal as JSON.
+         row : OrderedDict.
+             The current CSV record.
+         field_name : string
+             The Drupal fieldname/CSV column header.
+         Returns
+         -------
+         dictionary
+             A dictionary representing the entity that is POSTed to Drupal as JSON.
+        """
+        if not row[field_name]:
+            return entity
+
+        id_field = row.get(config.get("id_field", "not_applicable"), "not_applicable")
+        subvalues = self.split_string(
+            config, row[field_name]
+        )  # Split string into list of values
+        subvalues = self.dedupe_values(subvalues)
+
+        cardinality = int(field_definitions[field_name].get("cardinality", -1))
+        if -1 < cardinality < len(subvalues):
+            subvalues = subvalues[:cardinality]
+            log_field_cardinality_violation(
+                field_name,
+                id_field,
+                str(cardinality),
+            )
+        entity[field_name] = subvalues
+
+        return entity
+
+    def update(
+        self,
+        config: dict,
+        field_definitions: dict,
+        entity: dict,
+        row: OrderedDict,
+        field_name: string,
+        entity_field_values: list,
+    ) -> dict:
+        """Note: this method appends incoming CSV values to existing values, replaces existing field
+        values with incoming values, or deletes all values from fields, depending on whether
+        config['update_mode'] is 'append', 'replace', or 'delete'. It does not replace individual
+        values within fields.
+        """
+        """Parameters
+           ----------
+            config : dict
+                The configuration settings defined by workbench_config.get_config().
+            field_definitions : dict
+                The field definitions object defined by get_field_definitions().
+            entity : dict
+                The dict that will be POSTed to Drupal as JSON.
+            row : OrderedDict.
+                The current CSV record.
+            field_name : string
+                The Drupal fieldname/CSV column header.
+            entity_field_values : list
+                List of dictionaries containing existing value(s) for field_name in the entity being updated.
+            Returns
+            -------
+            dictionary
+                A dictionary representing the entity that is PATCHed to Drupal as JSON.
+        """
+        raise Exception("The update method must be implemented in all subclasses.")
+
+    @staticmethod
+    def dedupe_values(values: list) -> list:
+        """Removes duplicate entries from 'values'."""
+        """Parameters
+           ----------
+            values : list
+                List containing value(s) to dedupe. Members could be strings
+                from CSV or dictionairies.
+            Returns
+            -------
+            list
+                A list of unique field values.
+        """
+        return deduplicate_field_values(values)
+
+    def remove_invalid_values(
+        self, config: dict, field_definitions: dict, field_name: str, values: list
+    ) -> list:
+        """Removes invalid entries from 'values'."""
+        """Parameters
+           ----------
+            config : dict
+                The configuration settings defined by workbench_config.get_config().
+            field_definitions : dict
+                The field definitions object defined by get_field_definitions().
+            field_name : string
+                The Drupal fieldname/CSV column header.
+            values : list
+                List containing strings split from CSV values.
+            Returns
+            -------
+            list
+                A list of valid field values.
+        """
+        raise Exception(
+            "The remove_invalid_values method must be implemented in all subclasses."
+        )
+
+    def serialize(
+        self, config: dict, field_definitions: dict, field_name: str, field_data: str
+    ) -> str:
+        """Serialized values into a format consistent with Workbench's CSV-field input format."""
+        """Parameters
+           ----------
+            config : dict
+                The configuration settings defined by workbench_config.get_config().
+            field_definitions : dict
+                The field definitions object defined by get_field_definitions().
+            field_name : string
+                The Drupal fieldname/CSV column header.
+            field_data : string
+                Raw JSON from the field named 'field_name'.
+            Returns
+            -------
+            string
+                A string structured same as the Workbench CSV field data for this field type,
+                or None if there is nothing to return.
+        """
+        raise Exception("The serialize method must be implemented in all subclasses.")
+
+    def split_string(self, config: dict, value: str):
+        """Splits a string into a list of values. Called in common create() and update() methods."""
+        """Parameters
+           ----------
+            config : dict
+                The configuration settings defined by workbench_config.get_config().
+            value : string
+                A string to split.
+            Returns
+            -------
+            list
+                A list of values.
+        """
+        raise Exception(
+            "The split_string method must be implemented in any subclass that wishes to use it."
+        )
+
+
+class SimpleField(WorkbenchField):
     """Functions for handling fields with text and other "simple" Drupal field data types,
     e.g. fields that have a "{'value': 'xxx'}" structure such as plain text fields, ETDF
-    fields. All functions return a "entity" dictionary that is passed to Requests' "json"
+    fields. All functions return an "entity" dictionary that is passed to Requests' "json"
     parameter.
 
     Note that text fields that are "formatted" (i.e., use text formats/output filters)
@@ -125,7 +282,7 @@ class SimpleField:
             Returns
             -------
             dictionary
-                A dictionary represeting the entity that is PATCHed to Drupal as JSON.
+                A dictionary representing the entity that is PATCHed to Drupal as JSON.
         """
         if config["update_mode"] == "delete":
             entity[field_name] = []
@@ -224,20 +381,6 @@ class SimpleField:
 
         return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def remove_invalid_values(self, config, field_definitions, field_name, values):
         """Removes invalid entries from 'values'."""
         """Parameters
@@ -367,9 +510,9 @@ class SimpleField:
             return subvalues[0]
 
 
-class GeolocationField:
+class GeolocationField(WorkbenchField):
     """Functions for handling fields with 'geolocation' Drupal field data type.
-    All functions return a "entity" dictionary that is passed to Requests'
+    All functions return an "entity" dictionary that is passed to Requests'
     "json" parameter.
 
     Note: this class assumes that the entity has the field identified in 'field_name'.
@@ -378,41 +521,10 @@ class GeolocationField:
     """
 
     def create(self, config, field_definitions, entity, row, field_name):
-        """Parameters
-        ----------
-         config : dict
-             The configuration settings defined by workbench_config.get_config().
-         field_definitions : dict
-             The field definitions object defined by get_field_definitions().
-         entity : dict
-             The dict that will be POSTed to Drupal as JSON.
-         row : OrderedDict.
-             The current CSV record.
-         field_name : string
-             The Drupal fieldname/CSV column header.
-         Returns
-         -------
-         dictionary
-             A dictionary represeting the entity that is POSTed to Drupal as JSON.
-        """
-        if not row[field_name]:
-            return entity
-
-        id_field = row.get(config.get("id_field", "not_applicable"), "not_applicable")
-        subvalues = split_geolocation_string(config, row[field_name])
-        subvalues = self.dedupe_values(subvalues)
-
-        cardinality = int(field_definitions[field_name].get("cardinality", -1))
-        if -1 < cardinality < len(subvalues):
-            subvalues = subvalues[:cardinality]
-            log_field_cardinality_violation(
-                field_name,
-                id_field,
-                cardinality,
-            )
-        entity[field_name] = subvalues
+        return super().create(config, field_definitions, entity, row, field_name)
 
-        return entity
+    def split_string(self, config: dict, value: str):
+        return split_geolocation_string(config, value)
 
     def update(
         self, config, field_definitions, entity, row, field_name, entity_field_values
@@ -439,7 +551,7 @@ class GeolocationField:
             Returns
             -------
             dictionary
-                A dictionary represeting the entity that is PATCHed to Drupal as JSON.
+                A dictionary representing the entity that is PATCHed to Drupal as JSON.
         """
         if config["update_mode"] == "delete":
             entity[field_name] = []
@@ -460,7 +572,7 @@ class GeolocationField:
 
         cardinality = int(field_definitions[field_name].get("cardinality", -1))
         if config["update_mode"] == "replace":
-            subvalues = split_geolocation_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             subvalues = self.dedupe_values(subvalues)
             field_values = []
             for subvalue in subvalues:
@@ -473,7 +585,7 @@ class GeolocationField:
             entity[field_name] = field_values
 
         if config["update_mode"] == "append":
-            subvalues = split_geolocation_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             subvalues = self.dedupe_values(subvalues)
             for subvalue in subvalues:
                 entity_field_values.append(subvalue)
@@ -487,20 +599,6 @@ class GeolocationField:
         entity[field_name] = self.dedupe_values(entity[field_name])
         return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def remove_invalid_values(self, config, field_definitions, field_name, values):
         """Removes invalid entries from 'values'."""
         """Parameters
@@ -566,7 +664,7 @@ class GeolocationField:
             return subvalues[0]
 
 
-class LinkField:
+class LinkField(WorkbenchField):
     """Functions for handling fields with 'link' Drupal field data type.
     All functions return a "entity" dictionary that is passed to Requests'
     "json" parameter.
@@ -577,42 +675,10 @@ class LinkField:
     """
 
     def create(self, config, field_definitions, entity, row, field_name):
-        """Parameters
-        ----------
-         config : dict
-             The configuration settings defined by workbench_config.get_config().
-         field_definitions : dict
-             The field definitions object defined by get_field_definitions().
-         entity : dict
-             The dict that will be POSTed to Drupal as JSON.
-         row : OrderedDict.
-             The current CSV record.
-         field_name : string
-             The Drupal fieldname/CSV column header.
-         Returns
-         -------
-         dictionary
-             A dictionary represeting the entity that is POSTed to Drupal as JSON.
-        """
-        if not row[field_name]:
-            return entity
+        return super().create(config, field_definitions, entity, row, field_name)
 
-        id_field = row.get(config.get("id_field", "not_applicable"), "not_applicable")
-
-        subvalues = split_link_string(config, row[field_name])
-        subvalues = self.dedupe_values(subvalues)
-
-        cardinality = int(field_definitions[field_name].get("cardinality", -1))
-        if -1 < cardinality < len(subvalues):
-            subvalues = subvalues[:cardinality]
-            log_field_cardinality_violation(
-                field_name,
-                id_field,
-                cardinality,
-            )
-        entity[field_name] = subvalues
-
-        return entity
+    def split_string(self, config: dict, value: str):
+        return split_link_string(config, value)
 
     def update(
         self, config, field_definitions, entity, row, field_name, entity_field_values
@@ -639,7 +705,7 @@ class LinkField:
             Returns
             -------
             dictionary
-                A dictionary represeting the entity that is PATCHed to Drupal as JSON.
+                A dictionary representing the entity that is PATCHed to Drupal as JSON.
         """
         if config["update_mode"] == "delete":
             entity[field_name] = []
@@ -661,7 +727,7 @@ class LinkField:
         cardinality = int(field_definitions[field_name].get("cardinality", -1))
         if config["update_mode"] == "replace":
             field_values = []
-            subvalues = split_link_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             subvalues = self.dedupe_values(subvalues)
             if -1 < cardinality < len(subvalues):
                 log_field_cardinality_violation(
@@ -672,7 +738,7 @@ class LinkField:
                 field_values.append(subvalue)
             entity[field_name] = field_values
         if config["update_mode"] == "append":
-            subvalues = split_link_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             subvalues = self.dedupe_values(subvalues)
             for subvalue in subvalues:
                 entity_field_values.append(subvalue)
@@ -685,20 +751,6 @@ class LinkField:
 
         return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def remove_invalid_values(self, config, field_definitions, field_name, values):
         """Removes invalid entries from 'values'."""
         """Parameters
@@ -771,7 +823,7 @@ class LinkField:
             return subvalues[0]
 
 
-class EntityReferenceField:
+class EntityReferenceField(WorkbenchField):
     """Functions for handling fields with 'entity_reference' Drupal field data type.
     All functions return a "entity" dictionary that is passed to Requests' "json"
     parameter.
@@ -797,7 +849,7 @@ class EntityReferenceField:
          Returns
          -------
          dictionary
-             A dictionary represeting the entity that is POSTed to Drupal as JSON.
+             A dictionary representing the entity that is POSTed to Drupal as JSON.
         """
         if not row[field_name]:
             return entity
@@ -943,20 +995,6 @@ class EntityReferenceField:
 
         return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def remove_invalid_values(self, config, field_definitions, field_name, values):
         """Removes invalid entries from 'values'."""
         """Parameters
@@ -1018,10 +1056,17 @@ class EntityReferenceField:
                 term_name = get_term_name(config, subvalue["target_id"])
                 if vocab_id is not False and term_name is not False:
                     subvalues.append(vocab_id + ":" + term_name)
-            else:
+            elif subvalue["target_type"] == "taxonomy_term":
                 # Output term IDs.
                 if ping_term(config, subvalue["target_id"]) is True:
                     subvalues.append(str(subvalue["target_id"]))
+            elif subvalue["target_type"] == "node":
+                # Output node IDs.
+                if ping_node(config, subvalue["target_id"]) is True:
+                    subvalues.append(str(subvalue["target_id"]))
+            else:
+                # Fallback for any other entity type.
+                subvalues.append(str(subvalue["target_id"]))
 
         if len(subvalues) > 1:
             return config["subdelimiter"].join(subvalues)
@@ -1031,7 +1076,7 @@ class EntityReferenceField:
             return subvalues[0]
 
 
-class TypedRelationField:
+class TypedRelationField(WorkbenchField):
     """Functions for handling fields with 'typed_relation' Drupal field data type.
     All functions return a "entity" dictionary that is passed to Requests' "json"
     parameter.
@@ -1060,7 +1105,7 @@ class TypedRelationField:
          Returns
          -------
          dictionary
-             A dictionary represeting the entity that is POSTed to Drupal as JSON.
+             A dictionary representing the entity that is POSTed to Drupal as JSON.
         """
         if not row[field_name]:
             return entity
@@ -1113,7 +1158,7 @@ class TypedRelationField:
             Returns
             -------
             dictionary
-                A dictionary represeting the entity that is PATCHed to Drupal as JSON.
+                A dictionary representing the entity that is PATCHed to Drupal as JSON.
         """
         if config["update_mode"] == "delete":
             entity[field_name] = []
@@ -1176,20 +1221,6 @@ class TypedRelationField:
 
         return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def remove_invalid_values(self, config, field_definitions, field_name, values):
         """Removes invalid entries from 'values'."""
         """Parameters
@@ -1262,7 +1293,7 @@ class TypedRelationField:
             return subvalues[0]
 
 
-class AuthorityLinkField:
+class AuthorityLinkField(WorkbenchField):
     """Functions for handling fields with 'authority_link' Drupal field data type.
     All functions return a "entity" dictionary that is passed to Requests' "json"
     parameter.
@@ -1273,37 +1304,10 @@ class AuthorityLinkField:
     """
 
     def create(self, config, field_definitions, entity, row, field_name):
-        """Parameters
-        ----------
-         config : dict
-             The configuration settings defined by workbench_config.get_config().
-         field_definitions : dict
-             The field definitions object defined by get_field_definitions().
-         entity : dict
-             The dict that will be POSTed to Drupal as JSON.
-         row : OrderedDict.
-             The current CSV record.
-         field_name : string
-             The Drupal fieldname/CSV column header.
-         Returns
-         -------
-         dictionary
-             A dictionary represeting the entity that is POSTed to Drupal as JSON.
-        """
-        if not row[field_name]:
-            return entity
-
-        id_field = row.get(config.get("id_field", "not_applicable"), "not_applicable")
-        subvalues = split_authority_link_string(config, row[field_name])
-        subvalues = self.dedupe_values(subvalues)
-
-        cardinality = int(field_definitions[field_name].get("cardinality", -1))
-        if -1 < cardinality < len(subvalues):
-            subvalues = subvalues[:cardinality]
-            log_field_cardinality_violation(field_name, id_field, str(cardinality))
-        entity[field_name] = subvalues
+        return super().create(config, field_definitions, entity, row, field_name)
 
-        return entity
+    def split_string(self, config: dict, value: str):
+        return split_authority_link_string(config, value)
 
     def update(
         self, config, field_definitions, entity, row, field_name, entity_field_values
@@ -1352,7 +1356,7 @@ class AuthorityLinkField:
         cardinality = int(field_definitions[field_name].get("cardinality", -1))
         if config["update_mode"] == "replace":
             field_values = []
-            subvalues = split_authority_link_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             subvalues = self.dedupe_values(subvalues)
             if -1 < cardinality < len(subvalues):
                 log_field_cardinality_violation(
@@ -1363,7 +1367,7 @@ class AuthorityLinkField:
                 field_values.append(subvalue)
             entity[field_name] = field_values
         if config["update_mode"] == "append":
-            subvalues = split_authority_link_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             for subvalue in subvalues:
                 entity_field_values.append(subvalue)
             if -1 < cardinality < len(entity_field_values):
@@ -1376,20 +1380,6 @@ class AuthorityLinkField:
 
         return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def remove_invalid_values(self, config, field_definitions, field_name, values):
         """Removes invalid entries from 'values'."""
         """Parameters
@@ -1469,7 +1459,7 @@ class AuthorityLinkField:
             return subvalues[0]
 
 
-class MediaTrackField:
+class MediaTrackField(WorkbenchField):
     """Functions for handling fields with "media_track" Drupal (Islandora) field data type.
     All functions return a "entity" dictionary that is passed to Requests' "json"
     parameter.
@@ -1480,68 +1470,14 @@ class MediaTrackField:
     """
 
     def create(self, config, field_definitions, entity, row, field_name):
-        """Parameters
-        ----------
-         config : dict
-             The configuration settings defined by workbench_config.get_config().
-         field_definitions : dict
-             The field definitions object defined by get_field_definitions().
-         entity : dict
-             The dict that will be POSTed to Drupal as JSON.
-         row : OrderedDict.
-             The current CSV record.
-         field_name : string
-             The Drupal fieldname/CSV column header.
-         Returns
-         -------
-         dictionary
-             A dictionary represeting the entity that is POSTed to Drupal as JSON.
-        """
-        if not row[field_name]:
-            return entity
-
-        id_field = row.get(config.get("id_field", "not_applicable"), "not_applicable")
-        subvalues = split_media_track_string(config, row[field_name])
-        subvalues = self.dedupe_values(subvalues)
-        cardinality = int(field_definitions[field_name].get("cardinality", -1))
-        if -1 < cardinality < len(subvalues):
-            subvalues = subvalues[:cardinality]
-            log_field_cardinality_violation(
-                field_name,
-                id_field,
-                cardinality,
-            )
-        entity[field_name] = subvalues
+        return super().create(config, field_definitions, entity, row, field_name)
 
-        return entity
+    def split_string(self, config: dict, value: str):
+        return split_media_track_string(config, value)
 
     def update(
         self, config, field_definitions, entity, row, field_name, entity_field_values
     ):
-        """Note: this method appends incoming CSV values to existing values, replaces existing field
-        values with incoming values, or deletes all values from fields, depending on whether
-        config['update_mode'] is 'append', 'replace', or 'delete'. It doesn not replace individual
-        values within fields.
-        """
-        """Parameters
-           ----------
-            config : dict
-                The configuration settings defined by workbench_config.get_config().
-            field_definitions : dict
-                The field definitions object defined by get_field_definitions().
-            entity : dict
-                The dict that will be POSTed to Drupal as JSON.
-            row : OrderedDict.
-                The current CSV record.
-            field_name : string
-                The Drupal fieldname/CSV column header.
-            entity_field_values : list
-                List of dictionaries containing existing value(s) for field_name in the entity being updated.
-            Returns
-            -------
-            dictionary
-                A dictionary represeting the entity that is PATCHed to Drupal as JSON.
-        """
         if config["update_mode"] == "delete":
             entity[field_name] = []
             return entity
@@ -1555,7 +1491,7 @@ class MediaTrackField:
         cardinality = int(field_definitions[field_name].get("cardinality", -1))
         if config["update_mode"] == "replace":
             field_values = []
-            subvalues = split_media_track_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             subvalues = self.dedupe_values(subvalues)
             if -1 < cardinality < len(subvalues):
                 log_field_cardinality_violation(
@@ -1566,7 +1502,7 @@ class MediaTrackField:
                 field_values.append(subvalue)
             entity[field_name] = field_values
         if config["update_mode"] == "append":
-            subvalues = split_media_track_string(config, row[field_name])
+            subvalues = self.split_string(config, row[field_name])
             for subvalue in subvalues:
                 entity_field_values.append(subvalue)
             if -1 < cardinality < len(entity_field_values):
@@ -1578,20 +1514,6 @@ class MediaTrackField:
                 entity[field_name] = entity_field_values
         return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def remove_invalid_values(self, config, field_definitions, field_name, values):
         """Removes invalid entries from 'values'."""
         """Parameters
@@ -1677,7 +1599,7 @@ class MediaTrackField:
             return subvalues[0]
 
 
-class EntityReferenceRevisionsField:
+class EntityReferenceRevisionsField(WorkbenchField):
     """Functions for handling fields with 'entity_reference_revisions' Drupal field
     data type. This field *can* reference nodes, taxonomy terms, and media, but
     workbench only supports paragraphs for now.
@@ -1969,20 +1891,6 @@ class EntityReferenceRevisionsField:
                 entity[field_name] = entity[field_name][slice(0, cardinality)]
             return entity
 
-    def dedupe_values(self, values):
-        """Removes duplicate entries from 'values'."""
-        """Parameters
-           ----------
-            values : list
-                List containing value(s) to dedupe. Members could be strings
-                from CSV or dictionairies.
-            Returns
-            -------
-            list
-                A list of unique field values.
-        """
-        return deduplicate_field_values(values)
-
     def serialize(self, config, field_definitions, field_name, field_data):
         """Serialized values into a format consistent with Workbench's CSV-field input format."""
         """Parameters
@@ -2164,3 +2072,32 @@ class EntityReferenceRevisionsField:
         elif len(subvalues) == 0:
             return None
         return subvalues[0]
+
+
+class WorkbenchFieldFactory:
+
+    @staticmethod
+    def get_field_handler(field_type: str) -> WorkbenchField:
+        """Returns a WorkbenchField subclass based on the field type."""
+        """Parameters
+        field_type : string
+            The Drupal field type.
+        Returns
+        -------
+        WorkbenchField
+            A WorkbenchField subclass.
+        """
+        field_to_class_map = {
+            "entity_reference": "EntityReferenceField",
+            "entity_reference_revisions": "EntityReferenceRevisionsField",
+            "typed_relation": "TypedRelationField",
+            "authority_link": "AuthorityLinkField",
+            "media_track": "MediaTrackField",
+            "geolocation": "GeolocationField",
+            "link": "LinkField",
+        }
+        """Returns a WorkbenchField subclass based on the field type."""
+        if field_type in field_to_class_map:
+            return globals()[field_to_class_map[field_type]]()
+        else:
+            return SimpleField()
diff --git a/workbench_utils.py b/workbench_utils.py
index 0324d0f..546f4ea 100644
--- a/workbench_utils.py
+++ b/workbench_utils.py
@@ -1155,7 +1155,7 @@ def get_mid_from_media_url_alias(config, url_alias):
 
 
 def get_nid_from_url_without_config(url):
-    """Gets a node ID from a raw URL, with no accompanying config data. Useful
+    """Gets a node ID from a raw Drupal URL, with no accompanying config data. Useful
        within integration tests where the config is not directly accessible.
 
     Parameters
@@ -1168,7 +1168,7 @@ def get_nid_from_url_without_config(url):
         The node ID, or False if the URL cannot be found.
     """
     url = url + "?_format=json"
-    response = requests.get(url)
+    response = requests.get(url, verify=False)
     if response.status_code != 200:
         return False
     else:
@@ -1901,11 +1901,13 @@ def check_input(config, args):
         "get_media_report_from_view",
         "update_terms",
         "create_redirects",
+        "add_alt_text",
+        "update_alt_text",
     ]
     joiner = ", "
     if config["task"] not in tasks:
         message = (
-            '"task" in your configuration file must be one of "create", "update", "delete", '
+            '"task" in your configuration file must be one of "create", "update", "delete", "add_alt_text", "update_alt_text", '
             + '"add_media", "update_media", "delete_media", "delete_media_by_node", "create_from_files", "create_terms", "export_csv", "get_data_from_view", "update_terms", or "create_redirects".'
         )
         logging.error(message)
@@ -1921,7 +1923,7 @@ def check_input(config, args):
             logging.warning(message)
         if (
             config["recovery_mode_starting_from_node_id"] is not False
-            and config["recovery_mode_starting_from_node_id"].isnumeric() is True
+            and value_is_numeric(config["recovery_mode_starting_from_node_id"]) is True
         ):
             message = f'"recovery_mode" option in effect. Items that have already been ingested with node IDs starting at {config["recovery_mode_starting_from_node_id"]} will be skipped.'
             print(message)
@@ -2601,6 +2603,17 @@ def check_input(config, args):
         print(message)
         logging.info(message)
 
+        if (
+            "field_viewer_override_extensions" in config
+            or "field_viewer_override_models" in config
+        ):
+            preprocessed_input_csv_file_path = get_preprocessed_input_csv_file_path(
+                config
+            )
+            message = f'You should review "{preprocessed_input_csv_file_path}" to ensure that values in the "field_viewer_override" column have been correctly assigned based on your configuration settings.'
+            print("Warning: " + message)
+            logging.warning(message)
+
     # Check that Drupal fields that are required are in the 'create' task CSV file.
     if config["task"] == "create":
         required_drupal_fields_node = get_required_bundle_fields(
@@ -2892,6 +2905,75 @@ def check_input(config, args):
                 logging.error(message)
                 sys.exit("Error: " + message)
 
+    if config["task"] in ["add_alt_text", "update_alt_text"]:
+        _alt_text_required_options = ["task", "host", "username", "password"]
+        for _alt_text_required_options in _alt_text_required_options:
+            if _alt_text_required_options not in config_keys:
+                message = (
+                    "Please check your config file for required values: "
+                    + joiner.join(delete_media_required_options)
+                    + "."
+                )
+                logging.error(message)
+                sys.exit("Error: " + message)
+        update_mode_options = ["replace", "append", "delete"]
+        if config["update_mode"] not in update_mode_options:
+            message = (
+                'Your "update_mode" config option must be one of the following: '
+                + joiner.join(update_mode_options)
+                + "."
+            )
+            logging.error(message)
+            sys.exit("Error: " + message)
+
+        validate_alt_text_csv_data = get_csv_data(config)
+        row_counter = 0
+        for count, row in enumerate(validate_alt_text_csv_data, start=1):
+            row_counter += 1
+            if len(row["node_id"]) > 0:
+                node_id = row["node_id"]
+                parent_node_exists = ping_node(config, row["node_id"], warn=False)
+                if parent_node_exists is False:
+                    message = f'Node identified in "node_id" ({node_id}) in row "{row_counter}" of your input CSV cannot be found or accessed.'
+                    logging.error(message)
+                    sys.exit(
+                        "Error: " + message + " See Workbench log for more information."
+                    )
+            else:
+                message = f"Row {row_counter} in your input CSV file is empty."
+                logging.error(message)
+                sys.exit(
+                    "Error: " + message + " See Workbench log for more information."
+                )
+
+            if len(row["image_alt_text"]) > config["max_image_alt_text_length"]:
+                image_alt_text = row["image_alt_text"]
+                max_alt_text_length = config["max_image_alt_text_length"]
+                node_id = row["node_id"]
+                message = f"Alt text in input CSV row with node ID {node_id} is longer than the maximum configured alt text length ({max_alt_text_length})"
+                logging.warning(
+                    message
+                    + f" (length is {len(image_alt_text)} characters). This row will be skipped."
+                )
+                print("Warning: " + message + ". See log for more information.")
+
+    if config["task"] == "create":
+        validate_alt_text_csv_data = get_csv_data(config)
+        row_counter = 0
+        for count, row in enumerate(validate_alt_text_csv_data, start=1):
+            row_counter += 1
+            if "image_alt_text" in row:
+                if len(row["image_alt_text"]) > config["max_image_alt_text_length"]:
+                    image_alt_text = row["image_alt_text"]
+                    max_alt_text_length = config["max_image_alt_text_length"]
+                    node_id = row[config["id_field"]]
+                    message = f"Alt text in input CSV row with node ID {node_id} is longer than the maximum configured alt text length ({max_alt_text_length})"
+                    logging.warning(
+                        message
+                        + f" (length is {len(image_alt_text)} characters). Adding the alt text in this row will be skipped."
+                    )
+                    print("Warning: " + message + ". See log for more information.")
+
     if config["task"] == "create_terms" or config["task"] == "update_terms":
         # Check that all required fields are present in the CSV.
         field_definitions = get_field_definitions(
@@ -3027,7 +3109,9 @@ def check_input(config, args):
                     parent_nids = row["field_member_of"].split(config["subdelimiter"])
                     for parent_nid in parent_nids:
                         if len(parent_nid) > 0:
-                            parent_node_exists = ping_node(config, parent_nid)
+                            parent_node_exists = ping_node(
+                                config, parent_nid, warn=False
+                            )
                             if parent_node_exists is False:
                                 message = (
                                     "The 'field_member_of' field in row with ID '"
@@ -3143,6 +3227,16 @@ def check_input(config, args):
             message = 'For "create_redirects" tasks, your CSV file must contain a "redirect_target" column.'
             logging.error(message)
             sys.exit("Error: " + message)
+    if config["task"] in ["add_alt_text", "update_alt_text"]:
+        if "node_id" not in csv_column_headers:
+            t = config["task"]
+            message = f'For "{t}" tasks, your CSV file must contain a "node_id" column.'
+            logging.error(message)
+            sys.exit("Error: " + message)
+        if "image_alt_text" not in csv_column_headers:
+            message = f'For "{t}" tasks, your CSV file must contain a "image_alt_text" column.'
+            logging.error(message)
+            sys.exit("Error: " + message)
 
     warnings_about_redirect_input_csv = False
     if config["task"] == "create_redirects":
@@ -5693,25 +5787,47 @@ def clean_image_alt_text(input_string):
     return cleaned_string
 
 
-def patch_image_alt_text(config, media_id, node_csv_row):
+def patch_image_alt_text(config, media_id, csv_row):
     """Patch the alt text value for an image media. Use the parent node's title
     unless the CSV record contains an image_alt_text field with something in it.
     """
     if config["standalone_media_url"] is True:
-        get_endpoint = config["host"] + "/media/" + str(media_id) + "?_format=json"
+        get_media_endpoint = (
+            config["host"] + "/media/" + str(media_id) + "?_format=json"
+        )
+    else:
+        get_media_endpoint = (
+            config["host"] + "/media/" + str(media_id) + "/edit?_format=json"
+        )
+    get_media_headers = {"Content-Type": "application/json"}
+    get_media_response = issue_request(
+        config, "GET", get_media_endpoint, get_media_headers
+    )
+    if get_media_response.status_code == 200:
+        get_media_response_body = json.loads(get_media_response.text)
+        field_media_image_target_id = get_media_response_body["field_media_image"][0][
+            "target_id"
+        ]
     else:
-        get_endpoint = config["host"] + "/media/" + str(media_id) + "/edit?_format=json"
-    get_headers = {"Content-Type": "application/json"}
-    get_response = issue_request(config, "GET", get_endpoint, get_headers)
-    get_response_body = json.loads(get_response.text)
-    field_media_image_target_id = get_response_body["field_media_image"][0]["target_id"]
+        logging.error(
+            f"Media {get_media_endpoint} returned an HTTP status code of {get_media_response.status_code}."
+        )
+        return False
 
-    for field_name, field_value in node_csv_row.items():
+    for field_name, field_value in csv_row.items():
         if field_name == "title":
             alt_text = clean_image_alt_text(field_value)
-        if field_name == "image_alt_text" and len(field_value) > 0:
+        # "image_alt_text" can be in "create", "add_alt_text", or "update_alt_text" input CSV.
+        if field_name == "image_alt_text":
             alt_text = clean_image_alt_text(field_value)
 
+    max_image_alt_text_length = config["max_image_alt_text_length"]
+    if len(alt_text) > max_image_alt_text_length:
+        logging.warning(
+            f'Alt text "{alt_text}" is longer than the configured maximum length ({max_image_alt_text_length}), skipping adding it to image.'
+        )
+        return False
+
     media_json = {
         "bundle": [{"target_id": "image"}],
         "field_media_image": [
@@ -5720,18 +5836,22 @@ def patch_image_alt_text(config, media_id, node_csv_row):
     }
 
     if config["standalone_media_url"] is True:
-        patch_endpoint = config["host"] + "/media/" + str(media_id) + "?_format=json"
+        patch_media_endpoint = (
+            config["host"] + "/media/" + str(media_id) + "?_format=json"
+        )
     else:
-        patch_endpoint = (
+        patch_media_endpoint = (
             config["host"] + "/media/" + str(media_id) + "/edit?_format=json"
         )
-    patch_headers = {"Content-Type": "application/json"}
-    patch_response = issue_request(
-        config, "PATCH", patch_endpoint, patch_headers, media_json
+    patch_media_headers = {"Content-Type": "application/json"}
+    patch_media_response = issue_request(
+        config, "PATCH", patch_media_endpoint, patch_media_headers, media_json
     )
 
-    if patch_response.status_code != 200:
-        logging.warning("Alt text for image media %s not updated.", patch_endpoint)
+    if patch_media_response.status_code != 200:
+        logging.error("Alt text for image media %s not updated.", patch_media_endpoint)
+
+    return patch_media_response.status_code
 
 
 def remove_media_and_file(config, media_id):
@@ -5796,7 +5916,10 @@ def remove_media_and_file(config, media_id):
     media_bundle_name = get_media_response_body["bundle"][0]["target_id"]
     if media_bundle_name in config["media_track_file_fields"]:
         track_file_field = config["media_track_file_fields"][media_bundle_name]
-        if track_file_field in get_media_response_body[track_file_field]:
+        if (
+            track_file_field in get_media_response_body
+            and len(get_media_response_body[track_file_field]) > 0
+        ):
             for track_file in get_media_response_body[track_file_field]:
                 track_file_id = track_file["target_id"]
                 track_file_endpoint = (
@@ -5847,8 +5970,15 @@ def remove_media_and_file(config, media_id):
     return False
 
 
+def get_preprocessed_input_csv_file_path(config):
+    return (
+        os.path.join(config["temp_dir"], os.path.basename(config["input_csv"]))
+        + ".preprocessed"
+    )
+
+
 def get_csv_data(config, csv_file_target="node_fields", file_path=None):
-    """Read the input CSV data and prepare it for use in create, update, etc. tasks.
+    """Read the input CSV data and prepare it for use in all tasks that use an input CSV file.
 
     This function reads the source CSV file (or the CSV dump from Google Sheets or Excel),
     applies some prepocessing to each CSV record (specifically, it adds any CSV field
@@ -5907,10 +6037,8 @@ def get_csv_data(config, csv_file_target="node_fields", file_path=None):
         logging.error(message)
         sys.exit(message)
 
-    preprocessed_csv_path = (
-        os.path.join(config["temp_dir"], os.path.basename(input_csv_path))
-        + ".preprocessed"
-    )
+    preprocessed_csv_path = get_preprocessed_input_csv_file_path(config)
+
     csv_writer_file_handle = open(
         preprocessed_csv_path, "w+", newline="", encoding="utf-8"
     )
@@ -5978,8 +6106,16 @@ def get_csv_data(config, csv_file_target="node_fields", file_path=None):
         x for x in csv_reader_fieldnames if x not in config["ignore_csv_columns"]
     ]
 
+    #  If configured to do so, add "field_viewer_override" to output CSV so we can autopopulate the field_viewer_override column.
+    if config["task"] == "create" and (
+        "field_viewer_override_extensions" in config
+        or "field_viewer_override_models" in config
+    ):
+        if "field_viewer_override" not in csv_reader_fieldnames:
+            csv_reader_fieldnames.append("field_viewer_override")
+
     # CSV field templates and CSV value templates currently apply only to node CSV files, not vocabulary CSV files.
-    tasks = ["create", "update"]
+    tasks = ["create", "update", "add_media"]
     if config["task"] in tasks and csv_file_target == "node_fields":
         # If the config file contains CSV field templates, append them to the CSV data.
         # Make a copy of the column headers so we can skip adding templates to the new CSV
@@ -6045,8 +6181,8 @@ def get_csv_data(config, csv_file_target="node_fields", file_path=None):
         for row in itertools.islice(csv_reader, csv_start_row, config["csv_stop_row"]):
             row_num += 1
 
-            csv_rows_to_process_allowed_tasks = ["create", "update"]
-            # Skip rows specified not in config['csv_rows_to_process'] if its value is a path to a file.
+            csv_rows_to_process_allowed_tasks = ["create", "update", "add_media"]
+            # If the value in config['csv_rows_to_process'] is a path to a file, skip rows not identified in the file.
             if (
                 "csv_rows_to_process" in config
                 and config["task"] in csv_rows_to_process_allowed_tasks
@@ -6065,21 +6201,23 @@ def get_csv_data(config, csv_file_target="node_fields", file_path=None):
                 if row[config["id_field"]] not in ids_to_process:
                     continue
 
-            # Skip rows specified not in config['csv_rows_to_process'] if its value is a list.
+            # If the value of in config['csv_rows_to_process'] is a list, skip the rows not in the list.
             if (
                 "csv_rows_to_process" in config
                 and config["task"] in csv_rows_to_process_allowed_tasks
                 and len(config["csv_rows_to_process"]) > 0
                 and isinstance(config["csv_rows_to_process"], list)
             ):
+                config["csv_rows_to_process"] = [
+                    str(x) for x in config["csv_rows_to_process"]
+                ]
                 if row[config["id_field"]] not in config["csv_rows_to_process"]:
                     continue
 
-            # Apply the "is" and "isnot" csv_row_filters defined defined above.
-            # If the field/value combo is in the 'isnot' list, skip this row.
+            # Apply the "is" and "isnot" csv_row_filters defined defined above. If the field/value
+            # combo is in the 'isnot' list, skip this row.
             filter_out_this_csv_row = False
             if "csv_row_filters" in config and len(config["csv_row_filters"]) > 0:
-                # filter_out_this_csv_row = False
                 if len(row_filters_isnot) > 0:
                     for filter_field, filter_values in row_filters_isnot.items():
                         if len(filter_values) > 0 and filter_field in row:
@@ -6098,7 +6236,6 @@ def get_csv_data(config, csv_file_target="node_fields", file_path=None):
 
                 # If the field/value combo is not in the 'is' list, skip this row.
                 if len(row_filters_is) > 0:
-                    # filter_out_this_csv_row = False
                     for filter_field, filter_values in row_filters_is.items():
                         if len(filter_values) > 0 and filter_field in row:
                             # Split out multiple field values to test each one.
@@ -6139,6 +6276,15 @@ def get_csv_data(config, csv_file_target="node_fields", file_path=None):
                             config, "csv_value_templates", row
                         )
 
+                    #  If configured to do so, populate field_viewer_override column.
+                    if config["task"] == "create" and (
+                        "field_viewer_override_extensions" in config
+                        or "field_viewer_override_models" in config
+                    ):
+                        row["field_viewer_override"] = (
+                            get_field_viewer_override_from_condition(config, row)
+                        )
+
                     # Convert node URLs into node IDs.
                     if config["task"] in [
                         "update",
@@ -6824,87 +6970,16 @@ def get_term_field_data(config, vocab_id, term_name, term_csv_row):
             # Assemble Drupal field structures from CSV data. If new field types are added to
             # workbench_fields.py, they need to be registered in the following if/elif/else block.
 
-            # Entity reference fields (taxonomy_term and node)
-            if vocab_field_definitions[field_name]["field_type"] == "entity_reference":
-                entity_reference_field = workbench_fields.EntityReferenceField()
-                term_field_data = entity_reference_field.create(
-                    config,
-                    vocab_field_definitions,
-                    term_field_data,
-                    term_csv_row,
-                    field_name,
-                )
-
-            # Entity reference revision fields (paragraphs).
-            elif (
+            field = workbench_fields.WorkbenchFieldFactory.get_field_handler(
                 vocab_field_definitions[field_name]["field_type"]
-                == "entity_reference_revisions"
-            ):
-                entity_reference_revisions_field = (
-                    workbench_fields.EntityReferenceRevisionsField()
-                )
-                term_field_data = entity_reference_revisions_field.create(
-                    config,
-                    vocab_field_definitions,
-                    term_field_data,
-                    term_csv_row,
-                    field_name,
-                )
-
-            # Typed relation fields.
-            elif vocab_field_definitions[field_name]["field_type"] == "typed_relation":
-                typed_relation_field = workbench_fields.TypedRelationField()
-                term_field_data = typed_relation_field.create(
-                    config,
-                    vocab_field_definitions,
-                    term_field_data,
-                    term_csv_row,
-                    field_name,
-                )
-
-            # Geolocation fields.
-            elif vocab_field_definitions[field_name]["field_type"] == "geolocation":
-                geolocation_field = workbench_fields.GeolocationField()
-                term_field_data = geolocation_field.create(
-                    config,
-                    vocab_field_definitions,
-                    term_field_data,
-                    term_csv_row,
-                    field_name,
-                )
-
-            # Link fields.
-            elif vocab_field_definitions[field_name]["field_type"] == "link":
-                link_field = workbench_fields.LinkField()
-                term_field_data = link_field.create(
-                    config,
-                    vocab_field_definitions,
-                    term_field_data,
-                    term_csv_row,
-                    field_name,
-                )
-
-            # Authority Link fields.
-            elif vocab_field_definitions[field_name]["field_type"] == "authority_link":
-                authority_link_field = workbench_fields.AuthorityLinkField()
-                term_field_data = authority_link_field.create(
-                    config,
-                    vocab_field_definitions,
-                    term_field_data,
-                    term_csv_row,
-                    field_name,
-                )
-
-            # For non-entity reference and non-typed relation fields (text, integer, boolean etc.).
-            else:
-                simple_field = workbench_fields.SimpleField()
-                term_field_data = simple_field.create(
-                    config,
-                    vocab_field_definitions,
-                    term_field_data,
-                    term_csv_row,
-                    field_name,
-                )
+            )
+            term_field_data = field.create(
+                config,
+                vocab_field_definitions,
+                term_field_data,
+                term_csv_row,
+                field_name,
+            )
 
         return term_field_data
 
@@ -8871,7 +8946,7 @@ def create_children_from_directory(config, parent_csv_record, parent_node_id):
 
         if (
             config["recovery_mode_starting_from_node_id"] is not False
-            and config["recovery_mode_starting_from_node_id"].isnumeric() is True
+            and value_is_numeric(config["recovery_mode_starting_from_node_id"]) is True
             and parent_id is not None
         ):
             nid_in_map = recovery_mode_id_in_csv_id_to_node_id_map(
@@ -8959,10 +9034,61 @@ def create_children_from_directory(config, parent_csv_record, parent_node_id):
                 {"target_id": paged_content_model_tid, "target_type": "taxonomy_term"}
             ]
 
-        if "field_display_hints" in parent_csv_record:
-            node_json["field_display_hints"] = [
+        # Add viewer override if defined in CSV or config.
+        if "paged_content_page_viewer_override" in config:
+            if (
+                value_is_numeric(config["paged_content_page_viewer_override"]) is False
+                and config["paged_content_page_viewer_override"].startswith("http")
+                is True
+            ):
+                page_viewer_override_tid_info = get_all_representations_of_term(
+                    config,
+                    vocab_id="islandora_display",
+                    uri=config["paged_content_page_viewer_override"],
+                )
+                page_viewer_override_tid = page_viewer_override_tid_info["term_id"]
+            elif (
+                value_is_numeric(config["paged_content_page_viewer_override"]) is False
+                and config["paged_content_page_viewer_override"].startswith("http")
+                is False
+            ):
+                page_viewer_override_tid_info = get_all_representations_of_term(
+                    config,
+                    vocab_id="islandora_display",
+                    name=config["paged_content_page_viewer_override"],
+                )
+                page_viewer_override_tid = page_viewer_override_tid_info["term_id"]
+            else:
+                page_viewer_override_tid = config["paged_content_page_viewer_override"]
+
+        viewer_override_fieldname = config["viewer_override_fieldname"]
+        if (
+            viewer_override_fieldname in parent_csv_record
+            and "paged_content_page_viewer_override" in config
+        ):
+            node_json[viewer_override_fieldname] = [
+                {
+                    "target_id": page_viewer_override_tid,
+                    "target_type": "taxonomy_term",
+                }
+            ]
+        if (
+            viewer_override_fieldname in parent_csv_record
+            and "paged_content_page_viewer_override" not in config
+        ):
+            node_json[viewer_override_fieldname] = [
+                {
+                    "target_id": parent_csv_record["field_viewer_override"],
+                    "target_type": "taxonomy_term",
+                }
+            ]
+        if (
+            viewer_override_fieldname not in parent_csv_record
+            and "paged_content_page_viewer_override" in config
+        ):
+            node_json[viewer_override_fieldname] = [
                 {
-                    "target_id": parent_csv_record["field_display_hints"],
+                    "target_id": page_viewer_override_tid,
                     "target_type": "taxonomy_term",
                 }
             ]
@@ -8987,87 +9113,25 @@ def create_children_from_directory(config, parent_csv_record, parent_node_id):
                 if inherited_field in [
                     "title",
                     "field_model",
-                    "field_display_hints",
                     "uid",
                     "created",
+                    config["viewer_override_fieldname"],
                 ]:
                     continue
 
                 # Assemble Drupal field structures from CSV data. If new field types are added to
                 # workbench_fields.py, they need to be registered in the following if/elif/else block.
 
-                # Entity reference fields (taxonomy_term and node).
-                if (
+                field = workbench_fields.WorkbenchFieldFactory.get_field_handler(
                     field_definitions[inherited_field]["field_type"]
-                    == "entity_reference"
-                ):
-                    entity_reference_field = workbench_fields.EntityReferenceField()
-                    node_json = entity_reference_field.create(
-                        config,
-                        field_definitions,
-                        node_json,
-                        csv_row_to_apply_to_paged_children,
-                        inherited_field,
-                    )
-
-                # Typed relation fields.
-                elif (
-                    field_definitions[inherited_field]["field_type"] == "typed_relation"
-                ):
-                    typed_relation_field = workbench_fields.TypedRelationField()
-                    node_json = typed_relation_field.create(
-                        config,
-                        field_definitions,
-                        node_json,
-                        csv_row_to_apply_to_paged_children,
-                        inherited_field,
-                    )
-
-                # Geolocation fields.
-                elif field_definitions[inherited_field]["field_type"] == "geolocation":
-                    geolocation_field = workbench_fields.GeolocationField()
-                    node_json = geolocation_field.create(
-                        config,
-                        field_definitions,
-                        node_json,
-                        csv_row_to_apply_to_paged_children,
-                        inherited_field,
-                    )
-
-                # Link fields.
-                elif field_definitions[inherited_field]["field_type"] == "link":
-                    link_field = workbench_fields.LinkField()
-                    node_json = link_field.create(
-                        config,
-                        field_definitions,
-                        node_json,
-                        csv_row_to_apply_to_paged_children,
-                        inherited_field,
-                    )
-
-                # Authority Link fields.
-                elif (
-                    field_definitions[inherited_field]["field_type"] == "authority_link"
-                ):
-                    link_field = workbench_fields.AuthorityLinkField()
-                    node_json = link_field.create(
-                        config,
-                        field_definitions,
-                        node_json,
-                        csv_row_to_apply_to_paged_children,
-                        inherited_field,
-                    )
-
-                # For non-entity reference and non-typed relation fields (text, integer, boolean etc.).
-                else:
-                    simple_field = workbench_fields.SimpleField()
-                    node_json = simple_field.create(
-                        config,
-                        field_definitions,
-                        node_json,
-                        csv_row_to_apply_to_paged_children,
-                        inherited_field,
-                    )
+                )
+                node_json = field.create(
+                    config,
+                    field_definitions,
+                    node_json,
+                    csv_row_to_apply_to_paged_children,
+                    inherited_field,
+                )
 
         node_headers = {"Content-Type": "application/json"}
         node_endpoint = "/node?_format=json"
@@ -9269,7 +9333,7 @@ def get_rollback_csv_filepath(config):
 
     if config["timestamp_rollback"] is True or (
         config["recovery_mode_starting_from_node_id"] is not False
-        and config["recovery_mode_starting_from_node_id"].isnumeric() is True
+        and value_is_numeric(config["recovery_mode_starting_from_node_id"]) is True
     ):
         now_string = EXECUTION_START_TIME.strftime("%Y_%m_%d_%H_%M_%S")
 
@@ -9277,7 +9341,7 @@ def get_rollback_csv_filepath(config):
         rollback_csv_filename = f"{rollback_csv_filename_basename}.{now_string}.csv"
     elif (
         config["recovery_mode_starting_from_node_id"] is not False
-        and config["recovery_mode_starting_from_node_id"].isnumeric() is True
+        and value_is_numeric(config["recovery_mode_starting_from_node_id"]) is True
     ):
         rollback_csv_filename = (
             f"{rollback_csv_filename_basename}.{now_string}.recovery_mode.csv"
@@ -9365,7 +9429,7 @@ def get_rollback_config_filepath(config):
 
     if config["timestamp_rollback"] is True or (
         config["recovery_mode_starting_from_node_id"] is not False
-        and config["recovery_mode_starting_from_node_id"].isnumeric() is True
+        and value_is_numeric(config["recovery_mode_starting_from_node_id"]) is True
     ):
         now_string = EXECUTION_START_TIME.strftime("%Y_%m_%d_%H_%M_%S")
 
@@ -9376,7 +9440,7 @@ def get_rollback_config_filepath(config):
         )
     elif (
         config["recovery_mode_starting_from_node_id"] is not False
-        and config["recovery_mode_starting_from_node_id"].isnumeric() is True
+        and value_is_numeric(config["recovery_mode_starting_from_node_id"]) is True
     ):
         rollback_config_filepath = os.path.join(
             f"{rb_config_file_dir}",
@@ -9432,6 +9496,7 @@ def write_rollback_config(config, path_to_rollback_csv_file):
             "password": password,
             "input_dir": config["input_dir"],
             "standalone_media_url": config["standalone_media_url"],
+            "secure_ssl_only": config["secure_ssl_only"],
             "input_csv": path_to_rollback_csv_file,
         },
         rollback_config_file,
@@ -9823,8 +9888,13 @@ def check_file_exists(config, filename):
     # It's a remote file.
     if filename.startswith("http"):
         try:
+            headers = {"User-Agent": config["user_agent"]}
+
             head_response = requests.head(
-                filename, allow_redirects=True, verify=config["secure_ssl_only"]
+                filename,
+                allow_redirects=True,
+                verify=config["secure_ssl_only"],
+                headers=headers,
             )
             if head_response.status_code == 200:
                 return True
@@ -10145,123 +10215,216 @@ def get_remote_file_extension(config, file_url):
     return extension_with_dot
 
 
-def download_file_from_drupal(config, node_id):
-    """Download a media file from Drupal."""
+def get_field_viewer_override_from_condition(config, row):
+    """Derive value for the field_viewer_override CSV column based on conditions defined in configuration."""
     """Parameters
         ----------
         config : dict
             The configuration settings defined by workbench_config.get_config().
-        node_id : string
-            The ID of the node to delete media from.
+        row: OrderedDict
+            A CSV row. For pages/children created from subdirectories, this
+            is a version of the parent's row so we can get $csv_value values for non-required fields.
         Returns
         -------
-        file_name
-            The downloaded file's name, or False if unable to download the file.
+        The term ID, term name, or ther URI for the term from the Islandora Display vocabulary (whatever was
+        used in the configuration setting).
     """
-    if config["export_file_directory"] is None:
-        return False
+    # If the field_viewer_override column is populated, don't change it.
+    if "field_viewer_override" in row and len(row["field_viewer_override"].strip()) > 0:
+        return row["field_viewer_override"]
 
-    if not os.path.exists(config["export_file_directory"]):
-        try:
-            os.mkdir(config["export_file_directory"])
-        except Exception as e:
-            message = (
-                'Path in configuration option "export_file_directory" ("'
-                + config["export_file_directory"]
-                + '") is not writable.'
-            )
-            logging.error(message + " " + str(e))
-            sys.exit("Error: " + message + " See log for more detail.")
-    else:
-        message = (
-            'Path in configuration option "export_file_directory" ("'
-            + config["export_file_directory"]
-            + '") already exists.'
-        )
-        logging.info(message)
+    return_value = ""
+    # Get the field_viewer_override value from the row's field_model value first.
+    if (
+        "field_viewer_override_models" in config
+        and config["field_viewer_override_models"] is not None
+        and len(config["field_viewer_override_models"]) > 0
+    ):
+        for override in config["field_viewer_override_models"]:
+            for islandora_display_term_id, conditions in override.items():
+                conditions = [x.lower() for x in conditions]
+                if row["field_model"].lower() in conditions:
+                    # If multiple matches, last match pertains.
+                    return_value = islandora_display_term_id
+
+    # Then get field_viewer_override value from the extension in the row's "file" field, replacing the earlier assigned term ID/name/URI if necessary.
+    if (
+        "field_viewer_override_extensions" in config
+        and config["field_viewer_override_extensions"] is not None
+        and len(config["field_viewer_override_extensions"]) > 0
+    ):
+        if len(row["file"].strip()) == 0:
+            return row["field_viewer_override"]
+
+        if row["file"].strip().startswith("http"):
+            extension = get_remote_file_extension(config, row["file"])
+            extension = extension.lstrip(".").lower()
+        else:
+            _filename, extension = os.path.splitext(row["file"])
+            extension = extension.lstrip(".").lower()
 
+        for override in config["field_viewer_override_extensions"]:
+            for islandora_display_term_id, conditions in override.items():
+                conditions = [x.lower() for x in conditions]
+                conditions = [x.lstrip(".") for x in conditions]
+                if extension in conditions:
+                    # If multiple matches, last match pertains.
+                    return_value = islandora_display_term_id
+
+    return return_value
+
+
+def get_media_list(config, node_id, media_list=None):
+    """Retrieve media list for a node if not provided."""
+    if media_list is not None:
+        return media_list
     media_list_url = f"{config['host']}/node/{node_id}/media?_format=json"
-    media_list_response = issue_request(config, "GET", media_list_url)
-    if media_list_response.status_code == 200:
-        try:
-            media_list = json.loads(media_list_response.text)
-        except json.decoder.JSONDecodeError as e:
+    response = issue_request(config, "GET", media_list_url)
+    if response.status_code != 200:
+        logging.error(
+            f"Media list request failed for node {node_id}: {response.status_code}"
+        )
+        return None
+    try:
+        return json.loads(response.text)
+    except json.decoder.JSONDecodeError as e:
+        logging.error(f"Media query for node {node_id} failed: {e}")
+        return None
+
+
+def resolve_media_use_term_id(config, media_use_term_id, node_id):
+    """Resolve media use term ID from URI or configuration."""
+    if media_use_term_id is None:
+        media_use_term_id = config.get("export_file_media_use_term_id")
+        if media_use_term_id is None:
             logging.error(
-                f"Media query for node {node_id} produced the following error: {e}"
+                f"No media use term ID provided or configured for node {node_id}."
             )
-            return False
-        if len(media_list) == 0:
-            logging.warning(f"Node {node_id} has no media.")
-            return False
-
-        if str(config["export_file_media_use_term_id"]).startswith("http"):
-            config["export_file_media_use_term_id"] = get_term_id_from_uri(
-                config, config["export_file_media_use_term_id"]
+            return None
+    if isinstance(media_use_term_id, str) and media_use_term_id.startswith("http"):
+        term_id = get_term_id_from_uri(config, media_use_term_id)
+        if term_id is None:
+            logging.error(
+                f"Failed to convert URI {media_use_term_id} to term ID for node {node_id}."
             )
+            return None
+        return term_id
+    return media_use_term_id
+
+
+def find_file_url_in_media(config, media_list, media_use_term_id, node_id):
+    """Find the file URL in media entries matching the use term."""
+    for media in media_list:
+        for file_field in file_fields:
+            if file_field in media:
+                media_use_terms = media.get("field_media_use", [])
+                media_use_ids = [term.get("target_id") for term in media_use_terms]
+                if media_use_term_id in media_use_ids:
+                    file_info = media[file_field]
+                    if len(file_info) > 0:
+                        file_url = file_info[0].get("url")
+                        if file_url:
+                            return file_url
+    logging.debug(
+        f"No valid media found for node {node_id} with use term {media_use_term_id}"
+    )
+    return None
 
-        if config["export_file_media_use_term_id"] is False:
+
+def get_media_file_url(config, node_id, media_use_term_id=None, media_list=None):
+    """Retrieve and validate the URL of a media file from Drupal."""
+    media_list = get_media_list(config, node_id, media_list)
+    if media_list is None:
+        return False
+
+    resolved_term_id = resolve_media_use_term_id(config, media_use_term_id, node_id)
+    if resolved_term_id is None:
+        return False
+
+    file_url = find_file_url_in_media(config, media_list, resolved_term_id, node_id)
+    if not file_url:
+        return False
+
+    try:
+        head_response = requests.head(
+            file_url,
+            allow_redirects=True,
+            verify=config["secure_ssl_only"],
+            timeout=10,
+        )
+        if head_response.status_code != 200:
             logging.error(
-                f'Unknown value for configuration setting "export_file_media_use_term_id": {config["export_file_media_use_term_id"]}.'
+                f"URL validation failed for node {node_id}: {file_url} (HTTP {head_response.status_code})"
             )
             return False
+    except Exception as e:
+        logging.error(f"HEAD request failed for {file_url}: {str(e)}")
+        return False
 
-        for media in media_list:
-            for file_field_name in file_fields:
-                if file_field_name in media:
-                    if (
-                        len(media[file_field_name])
-                        and media["field_media_use"][0]["target_id"]
-                        == config["export_file_media_use_term_id"]
-                    ):
-                        url_filename = os.path.basename(
-                            media[file_field_name][0]["url"]
-                        )
-                        downloaded_file_path = os.path.join(
-                            config["export_file_directory"], url_filename
-                        )
-                        if os.path.exists(downloaded_file_path):
-                            downloaded_file_path = get_deduped_file_path(
-                                downloaded_file_path
-                            )
-                        f = open(downloaded_file_path, "wb+")
-                        # User needs to be anonymous since authenticated users are getting 403 responses. Probably something in
-                        # Drupal's FileAccessControlHandler code is doing this.
-                        file_download_response = requests.get(
-                            media[file_field_name][0]["url"],
-                            allow_redirects=True,
-                            verify=config["secure_ssl_only"],
-                        )
-                        if file_download_response.status_code == 200:
-                            f.write(file_download_response.content)
-                            f.close()
-                            filename_for_logging = os.path.basename(
-                                downloaded_file_path
-                            )
-                            logging.info(
-                                f'File "{filename_for_logging}" downloaded for node {node_id}.'
-                            )
-                            if os.path.isabs(config["export_file_directory"]):
-                                return downloaded_file_path
-                            else:
-                                return filename_for_logging
-                        else:
-                            message = (
-                                f"File at {media[file_field_name][0]['url']} (part of media for node {node_id}) could "
-                                + f"not be downloaded (HTTP response code {file_download_response.status_code})."
-                            )
-                            logging.error(message)
-                            return False
-                    else:
-                        logging.warning(
-                            f'Node {node_id} has no files in "{file_field_name}".'
-                        )
-                        return False
-                else:
-                    continue
+    logging.info(f"URL validated for node {node_id}: {file_url}")
+    return file_url
+
+
+def download_file_from_drupal(config, node_id, media_use_term_id=None, media_list=None):
+    """Download a media file from Drupal."""
+    if config.get("export_file_directory") is None:
+        logging.error("export_file_directory is not configured")
+        return False
+
+    if not os.path.exists(config["export_file_directory"]):
+        try:
+            os.mkdir(config["export_file_directory"])
+        except Exception as e:
+            message = f'Path "export_file_directory" ("{config["export_file_directory"]}") is not writable: {str(e)}'
+            logging.error(message)
+            sys.exit("Error: " + message + " See log for more detail.")
     else:
-        logging.error(
-            f"Attempt to fetch media list {media_list_url} returned an {media_list_response.status_code} HTTP response."
+        logging.info(
+            f'Path "export_file_directory" ("{config["export_file_directory"]}") already exists.'
         )
+
+    media_list = get_media_list(config, node_id, media_list)
+    if media_list is None:
+        return False
+
+    resolved_term_id = resolve_media_use_term_id(config, media_use_term_id, node_id)
+    if resolved_term_id is None:
+        return False
+
+    file_url = find_file_url_in_media(config, media_list, resolved_term_id, node_id)
+    if not file_url:
+        return False
+
+    url_filename = os.path.basename(file_url)
+    downloaded_file_path = os.path.join(config["export_file_directory"], url_filename)
+    if os.path.exists(downloaded_file_path):
+        downloaded_file_path = get_deduped_file_path(downloaded_file_path)
+
+    try:
+        with open(downloaded_file_path, "wb+") as f:
+            file_download_response = requests.get(
+                file_url,
+                allow_redirects=True,
+                verify=config["secure_ssl_only"],
+            )
+            if file_download_response.status_code == 200:
+                f.write(file_download_response.content)
+                filename_for_logging = os.path.basename(downloaded_file_path)
+                logging.info(
+                    f'File "{filename_for_logging}" downloaded for node {node_id}.'
+                )
+                return (
+                    downloaded_file_path
+                    if os.path.isabs(config["export_file_directory"])
+                    else filename_for_logging
+                )
+            else:
+                logging.error(
+                    f"File download failed for node {node_id}: {file_url} (HTTP {file_download_response.status_code})"
+                )
+                return False
+    except Exception as e:
+        logging.error(f"File download failed for node {node_id}: {str(e)}")
         return False
 
 
@@ -10725,61 +10888,61 @@ def serialize_field_json(config, field_definitions, field_name, field_data):
     # Assemble CSV output Drupal field data. If new field types are added to
     # workbench_fields.py, they need to be registered in the following if/elif/else block.
 
-    # Entity reference fields (taxonomy term and node).
-    if field_definitions[field_name]["field_type"] == "entity_reference":
-        serialized_field = workbench_fields.EntityReferenceField()
-        csv_field_data = serialized_field.serialize(
-            config, field_definitions, field_name, field_data
-        )
-    # Entity reference revision fields (mostly paragraphs).
-    elif field_definitions[field_name]["field_type"] == "entity_reference_revisions":
-        serialized_field = workbench_fields.EntityReferenceRevisionsField()
-        csv_field_data = serialized_field.serialize(
-            config, field_definitions, field_name, field_data
-        )
-    # Typed relation fields (currently, only taxonomy term)
-    elif field_definitions[field_name]["field_type"] == "typed_relation":
-        serialized_field = workbench_fields.TypedRelationField()
-        csv_field_data = serialized_field.serialize(
-            config, field_definitions, field_name, field_data
-        )
-    # Geolocation fields.
-    elif field_definitions[field_name]["field_type"] == "geolocation":
-        serialized_field = workbench_fields.GeolocationField()
-        csv_field_data = serialized_field.serialize(
-            config, field_definitions, field_name, field_data
-        )
-    # Link fields.
-    elif field_definitions[field_name]["field_type"] == "link":
-        serialized_field = workbench_fields.LinkField()
-        csv_field_data = serialized_field.serialize(
-            config, field_definitions, field_name, field_data
-        )
-    # Authority Link fields.
-    elif field_definitions[field_name]["field_type"] == "authority_link":
-        serialized_field = workbench_fields.AuthorityLinkField()
-        csv_field_data = serialized_field.serialize(
-            config, field_definitions, field_name, field_data
-        )
-    # Simple fields.
-    else:
-        serialized_field = workbench_fields.SimpleField()
-        csv_field_data = serialized_field.serialize(
-            config, field_definitions, field_name, field_data
-        )
+    serialized_field = workbench_fields.WorkbenchFieldFactory.get_field_handler(
+        field_definitions[field_name]["field_type"]
+    )
+    csv_field_data = serialized_field.serialize(
+        config, field_definitions, field_name, field_data
+    )
 
     return csv_field_data
 
 
 def csv_subset_warning(config):
-    """Create a message indicating that the csv_start_row and csv_stop_row config
-    options are present and that a subset of the input CSV will be used.
-    """
-    csv_data = list(get_csv_data(config))
-    start_row_id = csv_data[0][config["id_field"]]
-    stop_row_id = csv_data[-1][config["id_field"]]
+    """Create a message indicating that a subset of the input CSV data will be used,
+    specific to the current configuration settings."""
+
+    preprocessed_input_csv_file_path = get_preprocessed_input_csv_file_path(config)
+    review_message = f'You should review "{preprocessed_input_csv_file_path}" to ensure the correct rows will be processed.'
+
+    if "csv_row_filters" in config and len(config["csv_row_filters"]) > 0:
+        message = f'Using a subset of input CSV defined in your "csv_row_filters" setting. {review_message}'
+        logging.info(message)
+        return
+
+    if "csv_rows_to_process" in config:
+        if len(config["csv_rows_to_process"]) > 0 and isinstance(
+            config["csv_rows_to_process"], str
+        ):
+            path_to_ids_file = os.path.abspath(config["csv_rows_to_process"])
+            if os.path.exists(path_to_ids_file):
+                with open(path_to_ids_file) as fh:
+                    ids_to_process = fh.read().splitlines()
+                    ids_to_process = [x for x in ids_to_process if x]
+                message = (
+                    f'Using a subset of input CSV rows listed in "{path_to_ids_file}".'
+                )
+                print(message)
+                logging.info(message)
+                return
+            else:
+                message = f'File identified in the "csv_rows_to_process" config setting ({path_to_ids_file}) cannot be found.'
+                logging.error(message)
+                sys.exit("Error: " + message)
+
+        if len(config["csv_rows_to_process"]) > 0 and isinstance(
+            config["csv_rows_to_process"], list
+        ):
+            message = f'Using a subset of input CSV defined in your "csv_rows_to_process" setting. {review_message}'
+            print(message)
+            logging.info(message)
+            return
 
     if config["csv_start_row"] != 0 or config["csv_stop_row"] is not None:
+        csv_data = list(get_csv_data(config))
+        start_row_id = csv_data[0][config["id_field"]]
+        stop_row_id = csv_data[-1][config["id_field"]]
+
         message = f"Using a subset of the input CSV (will start at row {config['csv_start_row']} / row ID \"{start_row_id}\", stop at row {config['csv_stop_row']} / row ID \"{stop_row_id}\")."
         if config["csv_start_row"] != 0 and config["csv_stop_row"] is None:
             message = f"Using a subset of the input CSV (will start at row {config['csv_start_row']} / row ID {start_row_id})."
@@ -10787,6 +10950,7 @@ def csv_subset_warning(config):
             message = f"Using a subset of the input CSV (will stop at row {config['csv_stop_row']} / row ID {stop_row_id})."
         print(message)
         logging.info(message)
+        return
 
 
 def get_entity_reference_view_endpoints(config):


Test patch:
diff --git a/tests/assets/additional_files_test/create.csv b/tests/assets/additional_files_test/create.csv
index 92229de..02844f1 100644
--- a/tests/assets/additional_files_test/create.csv
+++ b/tests/assets/additional_files_test/create.csv
@@ -1,2 +1,2 @@
-file,id,title,field_model,preservation,transcript
-metropolis_maria_robot.png,001,Maria,http://purl.org/coar/resource_type/c_c513,metropolis_maria_robot_preservation.png,transcript.txt
+file,id,title,field_model,field_abstract,preservation,transcript
+metropolis_maria_robot.png,1,Maria,http://purl.org/coar/resource_type/c_c513,Test to exclude fields from export.,metropolis_maria_robot_preservation.png,transcript.txt
\ No newline at end of file
diff --git a/tests/assets/additional_files_test/export_files.yml b/tests/assets/additional_files_test/export_files.yml
new file mode 100644
index 0000000..ff0d34d
--- /dev/null
+++ b/tests/assets/additional_files_test/export_files.yml
@@ -0,0 +1,12 @@
+task: export_csv
+host: https://islandora.dev
+username: admin
+password: password
+input_csv: rollback.csv
+input_dir: "tests/assets/additional_files_test"
+export_csv_file_path: tests/assets/additional_files_test/exported_files/export_files.csv
+export_file_directory: tests/assets/additional_files_test/exported_files/
+additional_files:
+  - preservation: "http://pcdm.org/use#PreservationMasterFile"
+  - transcript: "http://pcdm.org/use#Transcript"
+secure_ssl_only: false
diff --git a/tests/assets/additional_files_test/export_urls.yml b/tests/assets/additional_files_test/export_urls.yml
new file mode 100644
index 0000000..1cfad72
--- /dev/null
+++ b/tests/assets/additional_files_test/export_urls.yml
@@ -0,0 +1,13 @@
+task: export_csv
+host: https://islandora.dev
+username: admin
+password: password
+input_csv: rollback.csv
+input_dir: "tests/assets/additional_files_test"
+export_csv_field_list: ['field_model']
+export_csv_file_path: tests/assets/additional_files_test/exported_files/export_urls.csv
+export_file_url_instead_of_download: true
+additional_files:
+  - preservation: "http://pcdm.org/use#PreservationMasterFile"
+  - transcript: "http://pcdm.org/use#Transcript"
+secure_ssl_only: false
diff --git a/tests/assets/alt_text_test/alt_text_test_create.csv b/tests/assets/alt_text_test/alt_text_test_create.csv
new file mode 100644
index 0000000..63aa070
--- /dev/null
+++ b/tests/assets/alt_text_test/alt_text_test_create.csv
@@ -0,0 +1,2 @@
+id,title,field_model,file
+at_01,Alt text test node,Image,alt_text_test_image.jpg
diff --git a/tests/assets/alt_text_test/create.yml b/tests/assets/alt_text_test/create.yml
new file mode 100644
index 0000000..2621b67
--- /dev/null
+++ b/tests/assets/alt_text_test/create.yml
@@ -0,0 +1,8 @@
+task: create
+host: https://islandora.dev
+username: admin
+password: password
+input_dir: "tests/assets/alt_text_test"
+input_csv: alt_text_test_create.csv
+secure_ssl_only: false
+standalone_media_url: true
diff --git a/tests/assets/alt_text_test/update_alt_text.yml b/tests/assets/alt_text_test/update_alt_text.yml
new file mode 100644
index 0000000..97a2d35
--- /dev/null
+++ b/tests/assets/alt_text_test/update_alt_text.yml
@@ -0,0 +1,8 @@
+task: update_alt_text
+host: https://islandora.dev
+username: admin
+password: password
+input_dir: "tests/assets/alt_text_test"
+input_csv: alt_text_test_update_replace.csv
+secure_ssl_only: false
+standalone_media_url: true
diff --git a/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_both.yml b/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_both.yml
new file mode 100644
index 0000000..5f72c78
--- /dev/null
+++ b/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_both.yml
@@ -0,0 +1,9 @@
+field_viewer_override_models:
+ - http://mozilla.github.io/pdf.js: ["Digital document"]
+ - http://openseadragon.github.io: ["Image"]
+ - MyCustomDisplayTermName: ["MyModel term name", "My other custom model term name"]
+field_viewer_override_extensions:
+ - http://mozilla.github.io/pdf.js: ["pdf"]
+ - http://openseadragon.github.io: ["tiff", "tif"]
+ - made_up_term_name: ["foo", "bar"]
+
diff --git a/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_extensions.yml b/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_extensions.yml
new file mode 100644
index 0000000..2a9c4f4
--- /dev/null
+++ b/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_extensions.yml
@@ -0,0 +1,4 @@
+field_viewer_override_extensions:
+ - http://mozilla.github.io/pdf.js: ["pdf"]
+ - http://openseadragon.github.io: ["tiff", ".Tif"]
+ - made_up_term_name: ["foo", ".Bar"]
diff --git a/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_models.yml b/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_models.yml
new file mode 100644
index 0000000..c99c3f7
--- /dev/null
+++ b/tests/assets/get_field_viewer_override_from_condition_test/field_viewer_override_models.yml
@@ -0,0 +1,4 @@
+field_viewer_override_models:
+ - http://mozilla.github.io/pdf.js: ["Digital document"]
+ - http://openseadragon.github.io: ["image"]
+ - MyCustomDisplayTermName: ["MyModel term name", "My other custom model term name"]
diff --git a/tests/field_tests.py b/tests/field_tests.py
index e1868a7..a489a42 100644
--- a/tests/field_tests.py
+++ b/tests/field_tests.py
@@ -5263,5 +5263,181 @@ class TestTypedRelationField(unittest.TestCase):
         self.assertEqual(output, expected)
 
 
+class TestWorkbenchFields(unittest.TestCase):
+    """"""
+
+    def setUp(self):
+        self.config = {
+            "task": "create",
+            "subdelimiter": "|",
+            "id_field": "id",
+            "columns_with_term_names": [],
+        }
+
+    def test_create(self):
+        existing_node = {
+            "type": [{"target_id": "islandora_object", "target_type": "node_type"}],
+            "title": [{"value": "Test node"}],
+            "status": [{"value": 1}],
+        }
+
+        # Create a node with a simple field of cardinality 1, no subdelimiters.
+        self.field_definitions = {
+            "field_foo": {
+                "cardinality": 1,
+                "formatted_text": False,
+                "field_type": "text",
+            }
+        }
+
+        field = workbench_fields.WorkbenchField()
+        csv_record = collections.OrderedDict()
+        csv_record["id"] = "simple_001"
+        csv_record["field_foo"] = "Field foo value"
+        with self.assertRaises(Exception):
+            field.create(
+                self.config,
+                self.field_definitions,
+                existing_node,
+                csv_record,
+                "field_foo",
+            )
+
+    def test_update(self):
+        # Update the node title, first with an 'update_mode' of replace.
+        existing_node = {
+            "type": [{"target_id": "islandora_object", "target_type": "node_type"}],
+            "title": [{"value": "Old title - replace."}],
+            "status": [{"value": 1}],
+        }
+
+        self.field_definitions = {
+            "title": {"cardinality": 1, "formatted_text": False, "field_type": "text"}
+        }
+
+        self.config["task"] = "update"
+        self.config["update_mode"] = "replace"
+
+        field = workbench_fields.WorkbenchField()
+        csv_record = collections.OrderedDict()
+        csv_record["title"] = "New title - replace."
+        csv_record["node_id"] = 1
+        with self.assertRaises(Exception):
+            field.update(
+                self.config,
+                self.field_definitions,
+                existing_node,
+                csv_record,
+                "title",
+                existing_node["title"],
+            )
+
+    def test_dudupe_values(self):
+        # First, split values from CSV.
+        input = [
+            "first value",
+            "first value",
+            "second value",
+            "second value",
+            "third value",
+        ]
+        field = workbench_fields.WorkbenchField()
+        output = field.dedupe_values(input)
+        self.assertEqual(output, ["first value", "second value", "third value"])
+
+        # Then fully formed dictionaries.
+        input = [
+            {"value": "First string"},
+            {"value": "Second string"},
+            {"value": "First string"},
+            {"value": "Second string"},
+            {"value": "Third string"},
+        ]
+        field = workbench_fields.SimpleField()
+        output = field.dedupe_values(input)
+        self.assertEqual(
+            output,
+            [
+                {"value": "First string"},
+                {"value": "Second string"},
+                {"value": "Third string"},
+            ],
+        )
+
+    def test_remove_invalid_values(self):
+
+        field = workbench_fields.WorkbenchField()
+
+        with self.assertRaises(Exception):
+            field.remove_invalid_values(
+                self.config,
+                self.field_definitions,
+                "field_foo",
+                ["Value1", "Value2", "Value3", "Value1"],
+            )
+
+    def test_serialize(self):
+
+        field = workbench_fields.WorkbenchField()
+
+        with self.assertRaises(Exception):
+            field.serialize(
+                self.config,
+                self.field_definitions,
+                "field_foo",
+                '{ "value": "Field foo value" }',
+            )
+
+
+class TestWorkbenchFieldFactory(unittest.TestCase):
+    def test_generate_expected_fields(self):
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler("link"),
+            workbench_fields.LinkField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler("typed_relation"),
+            workbench_fields.TypedRelationField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler(
+                "entity_reference"
+            ),
+            workbench_fields.EntityReferenceField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler(
+                "entity_reference_revisions"
+            ),
+            workbench_fields.EntityReferenceRevisionsField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler("geolocation"),
+            workbench_fields.GeolocationField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler("authority_link"),
+            workbench_fields.AuthorityLinkField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler("media_track"),
+            workbench_fields.MediaTrackField,
+        )
+
+    def test_generate_default_field(self):
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler("default"),
+            workbench_fields.SimpleField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler(""),
+            workbench_fields.SimpleField,
+        )
+        self.assertIsInstance(
+            workbench_fields.WorkbenchFieldFactory.get_field_handler("anything"),
+            workbench_fields.SimpleField,
+        )
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/islandora_tests.py b/tests/islandora_tests.py
index ca3faa2..73997e4 100644
--- a/tests/islandora_tests.py
+++ b/tests/islandora_tests.py
@@ -7,6 +7,7 @@ contain tests that interact with an Islandora instance.
 
 import sys
 import os
+import csv
 import glob
 from ruamel.yaml import YAML
 import subprocess
@@ -16,6 +17,7 @@ import json
 import unittest
 import time
 import copy
+import shutil
 
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 import workbench_utils
@@ -487,6 +489,117 @@ class TestUpdate(unittest.TestCase):
             os.remove(create_csv_preprocessed_file)
 
 
+class TestImageAltText(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.nids = list()
+        cls.islandora_host = ""
+        cls.islandora_username = ""
+        cls.islandora_password = ""
+        cls.create_config_file_path = ""
+        cls.current_dir = ""
+
+    def setUp(self):
+        self.current_dir = os.path.dirname(os.path.abspath(__file__))
+        self.create_config_file_path = os.path.join(
+            self.current_dir, "assets", "alt_text_test", "create.yml"
+        )
+
+        self.update_config_file_path = os.path.join(
+            self.current_dir, "assets", "alt_text_test", "update_alt_text.yml"
+        )
+
+        yaml = YAML()
+        with open(self.create_config_file_path, "r") as f:
+            config_file_contents = f.read()
+        config_data = yaml.load(config_file_contents)
+        config = {}
+        for k, v in config_data.items():
+            config[k] = v
+        self.islandora_host = config["host"]
+        self.islandora_username = config["username"]
+        self.islandora_password = config["password"]
+
+        TestImageAltText.islandora_host = config["host"]
+        TestImageAltText.islandora_username = config["username"]
+        TestImageAltText.islandora_password = config["password"]
+
+        TestImageAltText.current_dir = self.current_dir
+        TestImageAltText.create_config_file_path = self.create_config_file_path
+
+        self.create_cmd = ["./workbench", "--config", self.create_config_file_path]
+        self.update_cmd = ["./workbench", "--config", self.update_config_file_path]
+        self.temp_dir = "/tmp"
+
+        self.update_csv_path = os.path.join(
+            self.current_dir,
+            "assets",
+            "alt_text_test",
+            "alt_text_test_update_replace.csv",
+        )
+
+    def test_create_with_alt_text(self):
+        requests.packages.urllib3.disable_warnings()
+        create_output = subprocess.check_output(self.create_cmd)
+        create_output = create_output.decode().strip()
+        create_lines = create_output.splitlines()
+        with open(self.update_csv_path, "w") as fh:
+            fh.write("node_id,image_alt_text\n")
+            for line in create_lines:
+                if "created at" in line:
+                    nid = line.rsplit("/", 1)[-1]
+                    nid = nid.strip(".")
+                    if workbench_utils.value_is_numeric(nid):
+                        TestImageAltText.nids.append(nid)
+                        fh.write(nid + ",A medieval cat")
+
+    @unittest.skipIf(
+        "GITHUB_ACTIONS" in os.environ,
+        "Passes when tests locally run but not in Github workflows.",
+    )
+    def test_update_alt_text(self):
+        requests.packages.urllib3.disable_warnings()
+        create_output = subprocess.check_output(self.update_cmd)
+
+        media_list_url = (
+            f"{self.islandora_host}/node/{TestImageAltText.nids[0]}/media?_format=json"
+        )
+        media_list_response = requests.get(
+            media_list_url,
+            auth=(self.islandora_username, self.islandora_password),
+            verify=False,
+        )
+        media_list = json.loads(media_list_response.text)
+
+        for media in media_list:
+            if media["bundle"][0]["target_id"] == "image":
+                if "field_media_image" in media:
+                    alt_text = media["field_media_image"][0]["alt"]
+                    self.assertEqual(alt_text, "A medieval cat")
+
+    @classmethod
+    def tearDownClass(cls):
+        for nid in TestImageAltText.nids:
+            quick_delete_cmd = [
+                "./workbench",
+                "--config",
+                TestImageAltText.create_config_file_path,
+                "--quick_delete_node",
+                TestImageAltText.islandora_host + "/node/" + nid,
+            ]
+            quick_delete_output = subprocess.check_output(quick_delete_cmd)
+
+        rollback_file_path = os.path.join(
+            TestImageAltText.current_dir,
+            "assets",
+            "alt_text_test",
+            "rollback.csv",
+        )
+        if os.path.exists(rollback_file_path):
+            os.remove(rollback_file_path)
+
+
 class TestCreateWithNonLatinText(unittest.TestCase):
 
     def setUp(self):
@@ -761,8 +874,8 @@ class TestSecondaryTaskWithGoogleSheets(unittest.TestCase):
             os.remove(google_sheet_csv_path)
 
         secondary_task_google_sheets_csv_paths = glob.glob(
-            "*secondary_task_with_google_sheets_and_excel_test_google_sheets_secondary*",
-            root_dir=self.temp_dir,
+            self.temp_dir
+            + "/**secondary_task_with_google_sheets_and_excel_test_google_sheets_secondary*"
         )
         for secondary_csv_file_path in secondary_task_google_sheets_csv_paths:
             if os.path.exists(os.path.join(self.temp_dir, secondary_csv_file_path)):
@@ -861,8 +974,7 @@ class TestSecondaryTaskWithExcel(unittest.TestCase):
             os.remove(excel_csv_path)
 
         secondary_task_excel_csv_paths = glob.glob(
-            "*secondary_task_with_google_sheets_and_excel_test_excel_secondary*",
-            root_dir=self.temp_dir,
+            "/**secondary_task_with_google_sheets_and_excel_test_google_sheets_secondary*"
         )
         for secondary_csv_file_path in secondary_task_excel_csv_paths:
             if os.path.exists(os.path.join(self.temp_dir, secondary_csv_file_path)):
@@ -1514,6 +1626,106 @@ class TestAdditionalFilesAddMediaAllowMissingFilesTrue(unittest.TestCase):
             os.remove(self.true_with_additional_files_log_file_path)
 
 
+class TestExportCSVWithAdditionalFiles(unittest.TestCase):
+    def setUp(self):
+        """Create nodes with additional files for testing export."""
+        self.current_dir = os.path.dirname(os.path.abspath(__file__))
+        self.test_assets_dir = os.path.join(
+            self.current_dir, "assets", "additional_files_test"
+        )
+
+        # Create nodes using the existing additional files test infrastructure
+        self.create_config = os.path.join(self.test_assets_dir, "create.yml")
+        create_cmd = ["./workbench", "--config", self.create_config]
+        subprocess.check_output(create_cmd)
+
+        # Get node IDs from rollback.csv generated by create task
+        self.rollback_csv = os.path.join(self.test_assets_dir, "rollback.csv")
+        with open(self.rollback_csv, "r") as f:
+            self.nids = [
+                line.strip() for line in f.readlines()[3:]
+            ]  # Skip header and comments
+
+        # Setup export directory relative to test assets
+        self.export_dir = os.path.join(self.test_assets_dir, "exported_files")
+        os.makedirs(self.export_dir, exist_ok=True)
+
+    def test_export_additional_files_as_urls(self):
+        """Test exporting additional files as URLs."""
+        requests.packages.urllib3.disable_warnings()
+        export_config = os.path.join(self.test_assets_dir, "export_urls.yml")
+        export_cmd = ["./workbench", "--config", export_config]
+        subprocess.check_output(export_cmd)
+
+        exported_csv = os.path.join(self.export_dir, "export_urls.csv")
+        self.assertTrue(os.path.exists(exported_csv))
+
+        with open(exported_csv, "r") as f:
+            reader = csv.DictReader(f)
+            for row in reader:
+                # Verify main fields, file and additional files columns\
+                self.assertIn("node_id", row)
+                self.assertIn("title", row)
+                self.assertIn("field_model", row)
+                self.assertIn("file", row)
+                self.assertIn("preservation", row)
+                self.assertIn("transcript", row)
+
+                # Verify that columns not in the field list are not included.
+                self.assertNotIn("field_abstract", row)
+
+                # Check URLs are valid and accessible
+                for url_field in ["file", "preservation", "transcript"]:
+                    if row[url_field]:  # Handle empty fields
+                        response = requests.head(row[url_field], verify=False)
+                        self.assertEqual(
+                            response.status_code,
+                            200,
+                            f"Failed to access {url_field} URL: {row[url_field]}",
+                        )
+
+    def test_export_additional_files_as_files(self):
+        """Test exporting additional files as downloaded files."""
+        export_config = os.path.join(self.test_assets_dir, "export_files.yml")
+        export_cmd = ["./workbench", "--config", export_config]
+        subprocess.check_output(export_cmd)
+
+        exported_csv = os.path.join(self.export_dir, "export_files.csv")
+        self.assertTrue(os.path.exists(exported_csv))
+
+        with open(exported_csv, "r") as f:
+            reader = csv.DictReader(f)
+            for row in reader:
+                # Check all file fields exist in export directory
+                for field in ["file", "preservation", "transcript"]:
+                    if row[field]:  # Handle empty fields
+                        file_path = os.path.join(self.export_dir, row[field])
+                        self.assertTrue(
+                            os.path.exists(file_path),
+                            f"File {row[field]} not found in export directory",
+                        )
+
+    def tearDown(self):
+        """Cleanup created nodes and exported files."""
+        for nid in self.nids:
+            quick_delete_cmd = [
+                "./workbench",
+                "--config",
+                self.create_config,
+                "--quick_delete_node",
+                "https://islandora.dev/node/" + nid,
+            ]
+            quick_delete_output = subprocess.check_output(quick_delete_cmd)
+
+        # Cleanup export directory
+        if os.path.exists(self.export_dir):
+            shutil.rmtree(self.export_dir)
+
+        # Remove generated files
+        if os.path.exists(self.rollback_csv):
+            os.remove(self.rollback_csv)
+
+
 class TestUpdateMediaFields(unittest.TestCase):
     """Create a couple nodes plus image media, update the media's field_original_name
     and field_width fields, then confirm they were updated by GETting the media's JSON.
diff --git a/tests/islandora_tests_recovery_mode.py b/tests/islandora_tests_recovery_mode.py
index c709577..19fd5a2 100644
--- a/tests/islandora_tests_recovery_mode.py
+++ b/tests/islandora_tests_recovery_mode.py
@@ -394,6 +394,10 @@ class TestCreateRecoveryModeSingleItemsWithParentId(unittest.TestCase):
             os.remove(self.rollback_full_config_file_path)
 
 
+@unittest.skipIf(
+    "GITHUB_ACTIONS" in os.environ,
+    "Passes when tests locally run but not in Github workflows.",
+)
 class TestCreateRecoveryModePagedItemsFromDirectories(unittest.TestCase):
     current_dir = os.path.dirname(os.path.abspath(__file__))
 
diff --git a/tests/unit_tests.py b/tests/unit_tests.py
index 737bba8..2dcfd86 100644
--- a/tests/unit_tests.py
+++ b/tests/unit_tests.py
@@ -656,6 +656,123 @@ class TestSetMediaType(unittest.TestCase):
         self.assertEqual(res, "barmediatype")
 
 
+class TestGetFieldViewerOverrideFromConditio(unittest.TestCase):
+
+    def setUp(self):
+        yaml = YAML()
+        dir_path = os.path.dirname(os.path.realpath(__file__))
+
+        # Overrides by extension.
+        extensions_config_file_path = os.path.join(
+            dir_path,
+            "assets",
+            "get_field_viewer_override_from_condition_test",
+            "field_viewer_override_extensions.yml",
+        )
+        with open(extensions_config_file_path, "r") as f:
+            extensions_config_file_contents = f.read()
+        self.extensions_config_yaml = yaml.load(extensions_config_file_contents)
+
+        # Overrides by model.
+        models_config_file_path = os.path.join(
+            dir_path,
+            "assets",
+            "get_field_viewer_override_from_condition_test",
+            "field_viewer_override_models.yml",
+        )
+        with open(models_config_file_path, "r") as f:
+            models_config_file_contents = f.read()
+        self.models_config_yaml = yaml.load(models_config_file_contents)
+
+        # Overrides when both are present in config.
+        both_config_file_path = os.path.join(
+            dir_path,
+            "assets",
+            "get_field_viewer_override_from_condition_test",
+            "field_viewer_override_both.yml",
+        )
+        with open(both_config_file_path, "r") as f:
+            both_config_file_contents = f.read()
+        self.both_config_yaml = yaml.load(both_config_file_contents)
+
+    def test_overrides_by_extension(self):
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["file"] = "/tmp/foo.Tif"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.extensions_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "http://openseadragon.github.io")
+
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["file"] = "/tmp/foo.BAR"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.extensions_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "made_up_term_name")
+
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_viewer_override"] = "teststring"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.extensions_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "teststring")
+
+    def test_overrides_by_model(self):
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_model"] = "Digital document"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.models_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "http://mozilla.github.io/pdf.js")
+
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_model"] = "DIGITAL DOCUMENT"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.models_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "http://mozilla.github.io/pdf.js")
+
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_model"] = "My other custom model term name"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.models_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "MyCustomDisplayTermName")
+
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_viewer_override"] = "MyCustomDisplayTermName"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.extensions_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "MyCustomDisplayTermName")
+
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_viewer_override"] = "MyCUSTOMDisplayTermName"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.extensions_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "MyCUSTOMDisplayTermName")
+
+    def test_overrides_both_present(self):
+        # If both configurations for overrides are present, the one for overriding by extension pertains.
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_model"] = "Digital document"
+        fake_csv_record["file"] = "/tmp/foo.tiff"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.both_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "http://openseadragon.github.io")
+
+        fake_csv_record = collections.OrderedDict()
+        fake_csv_record["field_model"] = "Digital document"
+        fake_csv_record["file"] = "/tmp/foo.tiff"
+        fake_csv_record["field_viewer_override"] = "MyModel term name"
+        res = workbench_utils.get_field_viewer_override_from_condition(
+            self.both_config_yaml, fake_csv_record
+        )
+        self.assertEqual(res, "MyModel term name")
+
+
 class TestGetCsvFromExcel(unittest.TestCase):
     """Note: this tests the extraction of CSV data from Excel only,
     not using an Excel file as an input CSV file. That is tested
@@ -1525,11 +1642,11 @@ class TestMimeTypeFunctions(unittest.TestCase):
         config = dict({"input_dir": "."})
         fixtures = [
             {
-                "file_path": "tests/assets/mime_type_test/test.tXt",
+                "file_path": "tests/assets/mime_type_test/test2.tXt",
                 "mime_type": "text/plain",
             },
             {
-                "file_path": "tests/assets/mime_type_test/test.hocR",
+                "file_path": "tests/assets/mime_type_test/test2.hocR",
                 "mime_type": "text/vnd.hocr+html",
             },
             {