path: root/libs/apprise/AppriseAttachment.py

# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
#    this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

from . import attachment
from . import URLBase
from .AppriseAsset import AppriseAsset
from .logger import logger
from .common import ContentLocation
from .common import CONTENT_LOCATIONS
from .common import ATTACHMENT_SCHEMA_MAP
from .utils import GET_SCHEMA_RE


class AppriseAttachment:
    """
    Our Apprise Attachment File Manager

    """

    def __init__(self, paths=None, asset=None, cache=True, location=None,
                 **kwargs):
        """
        Loads all of the paths/urls specified (if any).

        The path can either be a single string identifying one explicit
        location, otherwise you can pass in a series of locations to scan
        via a list.

        By default we cache our responses so that subsiquent calls does not
        cause the content to be retrieved again.  For local file references
        this makes no difference at all.  But for remote content, this does
        mean more then one call can be made to retrieve the (same) data.  This
        method can be somewhat inefficient if disabled.  Only disable caching
        if you understand the consequences.

        You can alternatively set the cache value to an int identifying the
        number of seconds the previously retrieved can exist for before it
        should be considered expired.

        It's also worth nothing that the cache value is only set to elements
        that are not already of subclass AttachBase()

        Optionally set your current ContentLocation in the location argument.
        This is used to further handle attachments. The rules are as follows:
          - INACCESSIBLE: You simply have disabled use of the object; no
                          attachments will be retrieved/handled.
          - HOSTED:       You are hosting an attachment service for others.
                          In these circumstances all attachments that are LOCAL
                          based (such as file://) will not be allowed.
          - LOCAL:        The least restrictive mode as local files can be
                          referenced in addition to hosted.

        In all both HOSTED and LOCAL modes, INACCESSIBLE attachment types will
        continue to be inaccessible.  However if you set this field (location)
        to None (it's default value) the attachment location category will not
        be tested in any way (all attachment types will be allowed).

        The location field is also a global option that can be set when
        initializing the Apprise object.

        """

        # Initialize our attachment listings
        self.attachments = list()

        # Set our cache flag
        self.cache = cache

        # Prepare our Asset Object
        self.asset = \
            asset if isinstance(asset, AppriseAsset) else AppriseAsset()

        if location is not None and location not in CONTENT_LOCATIONS:
            msg = "An invalid Attachment location ({}) was specified." \
                  .format(location)
            logger.warning(msg)
            raise TypeError(msg)

        # Store our location
        self.location = location

        # Now parse any paths specified
        if paths is not None:
            # Store our path(s)
            if not self.add(paths):
                # Parse Source domain based on from_addr
                raise TypeError("One or more attachments could not be added.")

    def add(self, attachments, asset=None, cache=None):
        """
        Adds one or more attachments into our list.

        By default we cache our responses so that subsiquent calls does not
        cause the content to be retrieved again.  For local file references
        this makes no difference at all.  But for remote content, this does
        mean more then one call can be made to retrieve the (same) data.  This
        method can be somewhat inefficient if disabled.  Only disable caching
        if you understand the consequences.

        You can alternatively set the cache value to an int identifying the
        number of seconds the previously retrieved can exist for before it
        should be considered expired.

        It's also worth nothing that the cache value is only set to elements
        that are not already of subclass AttachBase()
        """
        # Initialize our return status
        return_status = True

        # Initialize our default cache value
        cache = cache if cache is not None else self.cache

        if asset is None:
            # prepare default asset
            asset = self.asset

        if isinstance(attachments, attachment.AttachBase):
            # Go ahead and just add our attachments into our list
            self.attachments.append(attachments)
            return True

        elif isinstance(attachments, str):
            # Save our path
            attachments = (attachments, )

        elif not isinstance(attachments, (tuple, set, list)):
            logger.error(
                'An invalid attachment url (type={}) was '
                'specified.'.format(type(attachments)))
            return False

        # Iterate over our attachments
        for _attachment in attachments:
            if self.location == ContentLocation.INACCESSIBLE:
                logger.warning(
                    "Attachments are disabled; ignoring {}"
                    .format(_attachment))
                return_status = False
                continue

            if isinstance(_attachment, str):
                logger.debug("Loading attachment: {}".format(_attachment))
                # Instantiate ourselves an object, this function throws or
                # returns None if it fails
                instance = AppriseAttachment.instantiate(
                    _attachment, asset=asset, cache=cache)
                if not isinstance(instance, attachment.AttachBase):
                    return_status = False
                    continue

            elif isinstance(_attachment, AppriseAttachment):
                # We were provided a list of Apprise Attachments
                # append our content together
                instance = _attachment.attachments

            elif not isinstance(_attachment, attachment.AttachBase):
                logger.warning(
                    "An invalid attachment (type={}) was specified.".format(
                        type(_attachment)))
                return_status = False
                continue

            else:
                # our entry is of type AttachBase, so just go ahead and point
                # our instance to it for some post processing below
                instance = _attachment

            # Apply some simple logic if our location flag is set
            if self.location and ((
                    self.location == ContentLocation.HOSTED
                    and instance.location != ContentLocation.HOSTED)
                    or instance.location == ContentLocation.INACCESSIBLE):
                logger.warning(
                    "Attachment was disallowed due to accessibility "
                    "restrictions ({}->{}): {}".format(
                        self.location, instance.location,
                        instance.url(privacy=True)))
                return_status = False
                continue

            # Add our initialized plugin to our server listings
            if isinstance(instance, list):
                self.attachments.extend(instance)

            else:
                self.attachments.append(instance)

        # Return our status
        return return_status

    @staticmethod
    def instantiate(url, asset=None, cache=None, suppress_exceptions=True):
        """
        Returns the instance of a instantiated attachment plugin based on
        the provided Attachment URL.  If the url fails to be parsed, then None
        is returned.

        A specified cache value will over-ride anything set

        """
        # Attempt to acquire the schema at the very least to allow our
        # attachment based urls.
        schema = GET_SCHEMA_RE.match(url)
        if schema is None:
            # Plan B is to assume we're dealing with a file
            schema = attachment.AttachFile.protocol
            url = '{}://{}'.format(schema, URLBase.quote(url))

        else:
            # Ensure our schema is always in lower case
            schema = schema.group('schema').lower()

            # Some basic validation
            if schema not in ATTACHMENT_SCHEMA_MAP:
                logger.warning('Unsupported schema {}.'.format(schema))
                return None

        # Parse our url details of the server object as dictionary containing
        # all of the information parsed from our URL
        results = ATTACHMENT_SCHEMA_MAP[schema].parse_url(url)

        if not results:
            # Failed to parse the server URL
            logger.warning('Unparseable URL {}.'.format(url))
            return None

        # Prepare our Asset Object
        results['asset'] = \
            asset if isinstance(asset, AppriseAsset) else AppriseAsset()

        if cache is not None:
            # Force an over-ride of the cache value to what we have specified
            results['cache'] = cache

        if suppress_exceptions:
            try:
                # Attempt to create an instance of our plugin using the parsed
                # URL information
                attach_plugin = \
                    ATTACHMENT_SCHEMA_MAP[results['schema']](**results)

            except Exception:
                # the arguments are invalid or can not be used.
                logger.warning('Could not load URL: %s' % url)
                return None

        else:
            # Attempt to create an instance of our plugin using the parsed
            # URL information but don't wrap it in a try catch
            attach_plugin = ATTACHMENT_SCHEMA_MAP[results['schema']](**results)

        return attach_plugin

    def clear(self):
        """
        Empties our attachment list

        """
        self.attachments[:] = []

    def size(self):
        """
        Returns the total size of accumulated attachments
        """
        return sum([len(a) for a in self.attachments if len(a) > 0])

    def pop(self, index=-1):
        """
        Removes an indexed Apprise Attachment from the stack and returns it.

        by default the last element is poped from the list
        """
        # Remove our entry
        return self.attachments.pop(index)

    def __getitem__(self, index):
        """
        Returns the indexed entry of a loaded apprise attachments
        """
        return self.attachments[index]

    def __bool__(self):
        """
        Allows the Apprise object to be wrapped in an 'if statement'.
        True is returned if at least one service has been loaded.
        """
        return True if self.attachments else False

    def __iter__(self):
        """
        Returns an iterator to our attachment list
        """
        return iter(self.attachments)

    def __len__(self):
        """
        Returns the number of attachment entries loaded
        """
        return len(self.attachments)