Turn large ntfy messages into a attachments

apprise/apprise_attachment.py

@@ -142,13 +142,8 @@ def add(self, attachments, asset=None, cache=None):
             # prepare default asset
             asset = self.asset
 
-        if isinstance(attachments, AttachBase):
-            # Go ahead and just add our attachments into our list
-            self.attachments.append(attachments)
-            return True
-
-        elif isinstance(attachments, str):
-            # Save our path
+        if isinstance(attachments, (AttachBase, str)):
+            # store our instance
             attachments = (attachments, )
 
         elif not isinstance(attachments, (tuple, set, list)):

apprise/attachment/base.py

@@ -148,6 +148,9 @@ def __init__(self, name=None, mimetype=None, cache=None, **kwargs):
         # Absolute path to attachment
         self.download_path = None
 
+        # Track open file pointers
+        self.__pointers = set()
+
         # Set our cache flag; it can be True, False, None, or a (positive)
         # integer... nothing else
         if cache is not None:
@@ -226,15 +229,14 @@ def mimetype(self):
         Content is cached once determied to prevent overhead of future
         calls.
         """
+        if not self.exists():
+            # we could not obtain our attachment
+            return None
 
         if self._mimetype:
             # return our pre-calculated cached content
             return self._mimetype
 
-        if not self.exists():
-            # we could not obtain our attachment
-            return None
-
         if not self.detected_mimetype:
             # guess_type() returns: (type, encoding) and sets type to None
             # if it can't otherwise determine it.
@@ -258,6 +260,9 @@ def exists(self, retrieve_if_missing=True):
         Simply returns true if the object has downloaded and stored the
         attachment AND the attachment has not expired.
         """
+        if self.location == ContentLocation.INACCESSIBLE:
+            # our content is inaccessible
+            return False
 
         cache = self.template_args['cache']['default'] \
             if self.cache is None else self.cache
@@ -295,6 +300,11 @@ def invalidate(self):
           - download_path: Must contain a absolute path to content
           - detected_mimetype: Should identify mimetype of content
         """
+
+        # Remove all open pointers
+        while self.__pointers:
+            self.__pointers.pop().close()
+
         self.detected_name = None
         self.download_path = None
         self.detected_mimetype = None
@@ -314,6 +324,26 @@ def download(self):
         raise NotImplementedError(
             "download() is implimented by the child class.")
 
+    def open(self, mode='rb'):
+        """
+        return our file pointer and track it (we'll auto close later
+        """
+        pointer = open(self.path, mode=mode)
+        self.__pointers.add(pointer)
+        return pointer
+
+    def __enter__(self):
+        """
+        support with keyword
+        """
+        return self.open()
+
+    def __exit__(self, value_type, value, traceback):
+        """
+        stub to do nothing; but support exit of with statement gracefully
+        """
+        return
+
     @staticmethod
     def parse_url(url, verify_host=True, mimetype_db=None, sanitize=True):
         """Parses the URL and returns it broken apart into a dictionary.
@@ -376,3 +406,9 @@ def __bool__(self):
         True is returned if our content was downloaded correctly.
         """
         return True if self.path else False
+
+    def __del__(self):
+        """
+        Perform any house cleaning
+        """
+        self.invalidate()

apprise/attachment/file.py

@@ -78,7 +78,8 @@ def url(self, privacy=False, *args, **kwargs):
 
         return 'file://{path}{params}'.format(
             path=self.quote(self.dirty_path),
-            params='?{}'.format(self.urlencode(params)) if params else '',
+            params='?{}'.format(self.urlencode(params, safe='/'))
+            if params else '',
         )
 
     def download(self, **kwargs):

apprise/attachment/http.py

@@ -352,7 +352,7 @@ def url(self, privacy=False, *args, **kwargs):
             port='' if self.port is None or self.port == default_port
                  else ':{}'.format(self.port),
             fullpath=self.quote(self.fullpath, safe='/'),
-            params=self.urlencode(params),
+            params=self.urlencode(params, safe='/'),
         )
 
     @staticmethod

apprise/attachment/memory.py

@@ -0,0 +1,212 @@
+# -*- coding: utf-8 -*-
+# BSD 2-Clause License
+#
+# Apprise - Push Notification Library.
+# Copyright (c) 2024, 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.
+
+import re
+import os
+import io
+from .base import AttachBase
+from ..common import ContentLocation
+from ..locale import gettext_lazy as _
+import uuid
+
+
+class AttachMemory(AttachBase):
+    """
+    A wrapper for Memory based attachment sources
+    """
+
+    # The default descriptive name associated with the service
+    service_name = _('Memory')
+
+    # The default protocol
+    protocol = 'memory'
+
+    # Content is local to the same location as the apprise instance
+    # being called (server-side)
+    location = ContentLocation.LOCAL
+
+    def __init__(self, content=None, name=None, mimetype=None,
+                 encoding='utf-8', **kwargs):
+        """
+        Initialize Memory Based Attachment Object
+
+        """
+        # Create our BytesIO object
+        self._data = io.BytesIO()
+
+        if content is None:
+            # Empty; do nothing
+            pass
+
+        elif isinstance(content, str):
+            content = content.encode(encoding)
+            if mimetype is None:
+                mimetype = 'text/plain'
+
+            if not name:
+                # Generate a unique filename
+                name = str(uuid.uuid4()) + '.txt'
+
+        elif not isinstance(content, bytes):
+            raise TypeError(
+                'Provided content for memory attachment is invalid')
+
+        # Store our content
+        if content:
+            self._data.write(content)
+
+        if mimetype is None:
+            # Default mimetype
+            mimetype = 'application/octet-stream'
+
+        if not name:
+            # Generate a unique filename
+            name = str(uuid.uuid4()) + '.dat'
+
+        # Initialize our base object
+        super().__init__(name=name, mimetype=mimetype, **kwargs)
+
+        return
+
+    def url(self, privacy=False, *args, **kwargs):
+        """
+        Returns the URL built dynamically based on specified arguments.
+        """
+
+        # Define any URL parameters
+        params = {
+            'mime': self._mimetype,
+        }
+
+        return 'memory://{name}?{params}'.format(
+            name=self.quote(self._name),
+            params=self.urlencode(params, safe='/')
+        )
+
+    def open(self, *args, **kwargs):
+        """
+        return our memory object
+        """
+        # Return our object
+        self._data.seek(0, 0)
+        return self._data
+
+    def __enter__(self):
+        """
+        support with clause
+        """
+        # Return our object
+        self._data.seek(0, 0)
+        return self._data
+
+    def download(self, **kwargs):
+        """
+        Handle memory download() call
+        """
+
+        if self.location == ContentLocation.INACCESSIBLE:
+            # our content is inaccessible
+            return False
+
+        if self.max_file_size > 0 and len(self) > self.max_file_size:
+            # The content to attach is to large
+            self.logger.error(
+                'Content exceeds allowable maximum memory size '
+                '({}KB): {}'.format(
+                    int(self.max_file_size / 1024), self.url(privacy=True)))
+
+            # Return False (signifying a failure)
+            return False
+
+        return True
+
+    def invalidate(self):
+        """
+        Removes data
+        """
+        self._data.truncate(0)
+        return
+
+    def exists(self):
+        """
+        over-ride exists() call
+        """
+        size = len(self)
+        return True if self.location != ContentLocation.INACCESSIBLE \
+            and size > 0 and (
+                self.max_file_size <= 0 or
+                (self.max_file_size > 0 and size <= self.max_file_size)) \
+            else False
+
+    @staticmethod
+    def parse_url(url):
+        """
+        Parses the URL so that we can handle all different file paths
+        and return it as our path object
+
+        """
+
+        results = AttachBase.parse_url(url, verify_host=False)
+        if not results:
+            # We're done early; it's not a good URL
+            return results
+
+        if 'name' not in results:
+            # Allow fall-back to be from URL
+            match = re.match(r'memory://(?P<path>[^?]+)(\?.*)?', url, re.I)
+            if match:
+                # Store our filename only (ignore any defined paths)
+                results['name'] = \
+                    os.path.basename(AttachMemory.unquote(match.group('path')))
+        return results
+
+    @property
+    def path(self):
+        """
+        return the filename
+        """
+        if not self.exists():
+            # we could not obtain our path
+            return None
+
+        return self._name
+
+    def __len__(self):
+        """
+        Returns the size of he memory attachment
+
+        """
+        return self._data.getbuffer().nbytes
+
+    def __bool__(self):
+        """
+        Allows the Apprise object to be wrapped in an based 'if statement'.
+        True is returned if our content was downloaded correctly.
+        """
+
+        return self.exists()