Upload graphic into Qualtrics with Python

qualtrics_graphic_uploader.py

"""
Qualtrics Graphic Uploader

This gist provides classes for uploading graphics to Qualtrics surveys.
It supports uploading from both local files and remote URLs, with built-in
caching to avoid re-uploading the same graphics.

The module implements a singleton pattern for each uploader type to ensure
a single instance per application lifetime.
"""

import csv
from os import getenv
from pathlib import Path
from typing import Optional, Dict

import requests


class QualtricsGraphicUploader:
    """Base class for uploading graphics to Qualtrics libraries.

    This class implements a singleton pattern to ensure only one instance
    per subclass exists. It handles API authentication, caching, and provides
    the foundation for graphic uploads to Qualtrics survey libraries.

    Attributes:
        api_token (str): Qualtrics API token for authentication.
        data_center (str): Qualtrics data center code (e.g., 'fra1', 'iad1').
        library_id (str): Qualtrics library ID where graphics are stored.
        target_folder (str): Target folder in the Qualtrics library.
        cache_file (Path): Path to the CSV file storing upload mappings.
        upload_mapping (Dict[str, str]): In-memory cache of path/URL to graphic ID mappings.

    Raises:
        ValueError: If QUALTRICS_API_TOKEN environment variable is not set.
    """
    _instances: Dict[type, 'QualtricsGraphicUploader'] = {}

    def __new__(cls, *args, **kwargs):
        if cls not in cls._instances:
            instance = super(QualtricsGraphicUploader, cls).__new__(cls)
            cls._instances[cls] = instance
            instance._initialized = False
        return cls._instances[cls]

    def __init__(
            self,
            target_folder: Optional[str] = None,
            library_id: Optional[str] = None,
            api_token: Optional[str] = None,
            data_center: Optional[str] = None
    ):
        """Initialize the Qualtrics Graphic Uploader.

        Configuration is loaded from provided parameters or environment variables
        in the following order of precedence:
        1. Explicit parameter values
        2. Environment variables (QUALTRICS_* prefixed)
        3. Default values

        Parameters:
            target_folder (str, optional): Target folder in Qualtrics library.
                Defaults to QUALTRICS_TARGET_FOLDER environment variable.
            library_id (str, optional): Qualtrics library ID for storing graphics.
                Defaults to QUALTRICS_LIBRARY_ID environment variable.
            api_token (str, optional): Qualtrics API token for authentication.
                Defaults to QUALTRICS_API_TOKEN environment variable.
                Required - will raise ValueError if not provided.
            data_center (str, optional): Qualtrics data center code (e.g., 'fra1', 'iad1').
                Defaults to QUALTRICS_DATA_CENTER environment variable or 'fra1'.

        Raises:
            ValueError: If QUALTRICS_API_TOKEN is not set via parameter or environment variable.

        Note:
            Due to singleton pattern, initialization only occurs once per class.
            Subsequent instantiations return the cached instance.
        """
        # Only initialize once per instance
        if self._initialized:
            return

        self.api_token = api_token or getenv('QUALTRICS_API_TOKEN')

        if not self.api_token:
            raise ValueError(
                "QUALTRICS_API_TOKEN not set. "
                "Please set this environment variable or create a .env file."
            )

        self.data_center = data_center or getenv('QUALTRICS_DATA_CENTER', 'fra1')
        self.library_id = library_id or getenv('QUALTRICS_LIBRARY_ID')
        self.target_folder = target_folder or getenv('QUALTRICS_TARGET_FOLDER')

        self.cache_file = Path("output/upload_mapping.csv")
        self.upload_mapping: Dict[str, str] = self._load_cache()

        self._initialized = True

    def _load_cache(self) -> Dict[str, str]:
        """Load the upload mapping cache from CSV file.

        Creates a new cache file if it doesn't exist. The cache maps
        file paths or URLs to their corresponding Qualtrics graphic IDs.
        """
        mapping = {}
        if not self.cache_file.exists():
            with self.cache_file.open('w', encoding='utf-8', newline='') as f:
                writer = csv.DictWriter(f, fieldnames=['path', 'graphic_id'])
                writer.writeheader()

            return mapping

        with self.cache_file.open('r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                if 'path' in row and 'graphic_id' in row:
                    mapping[row['path']] = row['graphic_id']

        return mapping

    def _store_in_cache(self, path: str, graphic_id: str):
        """Store a path-to-graphic-ID mapping in the cache file."""
        with self.cache_file.open('a', encoding='utf-8', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=['path', 'graphic_id'])
            writer.writerow({'path': path, 'graphic_id': graphic_id})

    def get_cached_graphic_id(self, path: str) -> Optional[str]:
        """Retrieve a cached graphic ID for a given path or URL.

        Parameters:
            path (str): File path or URL to look up in the cache.

        Returns:
            Optional[str]: The cached graphic ID if found, None otherwise.

        Examples:
            >>> uploader = QualtricsGraphicUploader(api_token='test_token')
            >>> uploader.get_cached_graphic_id('nonexistent.png')
            >>> uploader._store_in_cache('example.png', 'id_456')
            >>> uploader.get_cached_graphic_id('example.png')
            'id_456'
        """
        if path in self.upload_mapping:
            return self.upload_mapping[path]
        return None


class QualtricsGraphicUrlUploader(QualtricsGraphicUploader):
    """Upload graphics to Qualtrics from remote URLs.

    This subclass handles uploading graphics that are available at remote URLs.
    It fetches the image from the URL and uploads it to the Qualtrics library.

    Inherits singleton pattern from base class - only one instance will exist.
    """

    def get_cached_graphic_id(self, image_url: str) -> Optional[str]:
        """Retrieve cached graphic ID for a URL.

        Parameters:
            image_url (str): URL of the image to look up.

        Returns:
            Optional[str]: The cached graphic ID if found, None otherwise.

        Examples:
            >>> url_uploader = QualtricsGraphicUrlUploader(api_token='test_token')
            >>> url_uploader.get_cached_graphic_id('https://example.com/image.png')
            >>> url_uploader._store_in_cache('https://example.com/image.png', 'url_id_789')
            >>> url_uploader.get_cached_graphic_id('https://example.com/image.png')
            'url_id_789'
        """
        if image_url in self.upload_mapping:
            graphic_id = self.upload_mapping[image_url]
            return graphic_id

        return None

    def get_graphic_id(self, image_url: str) -> str:
        """Get graphic ID for an image URL, uploading if necessary.

        Checks cache first for previously uploaded graphics. If not found,
        uploads the image from the URL to Qualtrics and caches the result.

        Parameters:
            image_url (str): URL of the image to upload.

        Returns:
            str: Qualtrics graphic ID for the image.

        Raises:
            RuntimeError: If upload to Qualtrics fails.

        Examples:
            Cache hit example:
            >>> url_uploader = QualtricsGraphicUrlUploader(api_token='test_token')
            >>> url_uploader._store_in_cache('https://example.com/image.png', 'cached_id_123')
            >>> result = url_uploader.get_graphic_id('https://example.com/image.png')  # doctest: +SKIP
            ✓ Using cached graphic ID for https://example.com/image.png: cached_id_123
            >>> result == 'cached_id_123'  # doctest: +SKIP
            True

            Upload failure example:
            >>> url_uploader = QualtricsGraphicUrlUploader(api_token='test_token')
            >>> try:
            ...     url_uploader.get_graphic_id('https://invalid-url.com/missing.png')
            ... except RuntimeError as e:
            ...     'Failed to upload' in str(e)  # doctest: +SKIP
            True
        """
        graphic_id = self.get_cached_graphic_id(image_url)
        if graphic_id:
            print(f"✓ Using cached graphic ID for {image_url}: {graphic_id}")
            return graphic_id

        # Upload to Qualtrics
        print(f"↑ Uploading {image_url}...")
        graphic_id = self._upload_to_qualtrics(image_url)

        if graphic_id:
            print(f"✓ Using new graphic ID for {image_url}: {graphic_id}")
            return graphic_id
        else:
            raise RuntimeError(f"Failed to upload {image_url}")

    def _upload_to_qualtrics(self, image_url: str) -> Optional[str]:
        """Upload an image from a URL to Qualtrics.

        Makes an API request to upload the image and stores the returned
        graphic ID in the cache.
        """
        try:
            image_name = image_url.split('/')[-1]
            response = requests.post(
                f"https://{self.data_center}.qualtrics.com/API/v3/libraries/{self.library_id}/graphics",
                headers={'X-API-TOKEN': self.api_token, 'Accept': 'application/json'},
                files={
                    'name': (None, image_name),
                    'fileUrl': (None, image_url),
                    'contentType': (None, self._content_type(image_url)),
                    'folder': (None, self.target_folder),
                },
            )

            if response.status_code in [200, 201]:
                data = response.json()
                if 'result' in data and 'id' in data['result']:
                    graphic_id = data['result']['id']
                    self._store_in_cache(image_url, graphic_id)
                    return data['result']['id']
                else:
                    print(f"Error: Unexpected response format for {image_url}")
                    return None
            else:
                print(f"Error uploading {image_url}: {response.status_code} - {response.text}")
                return None

        except Exception as e:
            print(f"Error uploading {image_url}: {e}")
            return None

    def _content_type(self, image_url: str) -> str:
        """Determine the MIME type of an image from its URL."""
        extension = image_url.split('.')[-1]
        if extension == 'png':
            return 'image/png'
        elif extension == 'gif':
            return 'image/gif'
        else:
            return 'image/jpeg'


class QualtricsGraphicLocalUploader(QualtricsGraphicUploader):
    """Upload graphics to Qualtrics from local file system.

    This subclass handles uploading graphics that exist as local files
    on the file system. It reads the file and uploads it to the Qualtrics library.

    Inherits singleton pattern from base class - only one instance will exist.
    """

    def get_cached_graphic_id(self, file_path: Path) -> Optional[str]:
        """Retrieve cached graphic ID for a local file path.

        Parameters:
            file_path (Path): Local file path to look up in the cache.

        Returns:
            Optional[str]: The cached graphic ID if found, None otherwise.

        Examples:
            >>> from pathlib import Path
            >>> local_uploader = QualtricsGraphicLocalUploader(api_token='test_token')
            >>> test_path = Path('input/images/test.png')
            >>> local_uploader.get_cached_graphic_id(test_path)
            >>> local_uploader._store_in_cache(str(test_path), 'local_id_456')
            >>> local_uploader.get_cached_graphic_id(test_path)
            'local_id_456'
        """
        relative_path = str(file_path)
        return super().get_cached_graphic_id(relative_path)

    def get_graphic_id(self, file_path: Path) -> str:
        """Get graphic ID for a local file, uploading if necessary.

        Checks cache first for previously uploaded graphics. If not found,
        uploads the file to Qualtrics and caches the result.

        Parameters:
            file_path (Path): Path to the local file to upload.

        Returns:
            str: Qualtrics graphic ID for the file.

        Raises:
            RuntimeError: If upload to Qualtrics fails.

        Examples:
            Cache hit example:
            >>> from pathlib import Path
            >>> local_uploader = QualtricsGraphicLocalUploader(api_token='test_token')
            >>> test_path = Path('input/images/cached.png')
            >>> local_uploader._store_in_cache(str(test_path), 'cached_local_id_789')
            >>> result = local_uploader.get_graphic_id(test_path)  # doctest: +SKIP
            ✓ Using cached graphic ID for cached.png: cached_local_id_789
            >>> result == 'cached_local_id_789'  # doctest: +SKIP
            True

            Upload failure example:
            >>> local_uploader = QualtricsGraphicLocalUploader(api_token='test_token')
            >>> try:
            ...     local_uploader.get_graphic_id(Path('nonexistent/file.png'))
            ... except RuntimeError as e:
            ...     'Failed to upload' in str(e)  # doctest: +SKIP
            True
        """
        graphic_id = self.get_cached_graphic_id(file_path)
        if graphic_id:
            print(f"✓ Using cached graphic ID for {file_path.name}: {graphic_id}")
            return graphic_id

        # Upload to Qualtrics
        print(f"↑ Uploading {file_path.name}...")
        graphic_id = self._upload_to_qualtrics(file_path)

        if graphic_id:
            print(f"✓ Using new graphic ID for {file_path.name}: {graphic_id}")
            return graphic_id
        else:
            raise RuntimeError(f"Failed to upload {file_path.name}")

    def _upload_to_qualtrics(self, file_path: Path) -> Optional[str]:
        """Upload a local file to Qualtrics.

        Reads the file from disk and makes an API request to upload it.
        Stores the returned graphic ID in the cache.
        """
        try:
            response = requests.post(
                f"https://{self.data_center}.qualtrics.com/API/v3/libraries/{self.library_id}/graphics",
                headers={'X-API-TOKEN': self.api_token},
                data={'folder': self.target_folder},
                files={'file': (file_path.name, file_path.open('rb'), self._content_type(file_path))},
            )

            if response.status_code in [200, 201]:
                data = response.json()
                if 'result' in data and 'id' in data['result']:
                    graphic_id = data['result']['id']
                    self._store_in_cache(str(file_path), graphic_id)
                    return data['result']['id']
                else:
                    print(f"Error: Unexpected response format for {file_path.name}")
                    return None
            else:
                print(f"Error uploading {file_path.name}: {response.status_code} - {response.text}")
                return None

        except Exception as e:
            print(f"Error uploading {file_path.name}: {e}")
            return None

    def _content_type(self, file_path: Path) -> str:
        """Determine the MIME type of file from its extension."""
        extension = file_path.suffix.lstrip('.')
        if extension == '.gif':
            return 'image/gif'
        if extension == 'png':
            return 'image/png'
        elif extension == 'gif':
            return 'image/gif'
        else:
            return 'image/jpeg'