mirror of
https://github.com/SickGear/SickGear.git
synced 2024-12-27 13:13:37 +00:00
328 lines
12 KiB
Python
328 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
# BSD 3-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.
|
|
#
|
|
# 3. Neither the name of the copyright holder nor the names of its
|
|
# contributors may be used to endorse or promote products derived from
|
|
# this software without specific prior written permission.
|
|
#
|
|
# 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)
|