mirror
/
bazarr
mirror of https://github.com/morpheus65535/bazarr
1
0
Fork 0
Code Issues Releases Wiki Activity
bazarr/libs/apprise/AppriseConfig.py

290 lines
9.4 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import six
from . import config
from . import ConfigBase
from . import URLBase
from .AppriseAsset import AppriseAsset
from .utils import GET_SCHEMA_RE
from .utils import parse_list
from .utils import is_exclusive_match
from .logger import logger
class AppriseConfig(object):
"""
Our Apprise Configuration File Manager
- Supports a list of URLs defined one after another (text format)
- Supports a destinct YAML configuration format
"""
def __init__(self, paths=None, asset=None, cache=True, **kwargs):
"""
Loads all of the paths 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.
If no path is specified then a default list is used.
If cache is set to True, then after the data is loaded, it's cached
within this object so it isn't retrieved again later.
"""
# Initialize a server list of URLs
self.configs = list()
# Prepare our Asset Object
self.asset = \
asset if isinstance(asset, AppriseAsset) else AppriseAsset()
if paths is not None:
# Store our path(s)
self.add(paths)
return
def add(self, configs, asset=None, tag=None):
"""
Adds one or more config URLs into our list.
You can override the global asset if you wish by including it with the
config(s) that you add.
"""
# Initialize our return status
return_status = True
if isinstance(asset, AppriseAsset):
# prepare default asset
asset = self.asset
if isinstance(configs, ConfigBase):
# Go ahead and just add our configuration into our list
self.configs.append(configs)
return True
elif isinstance(configs, six.string_types):
# Save our path
configs = (configs, )
elif not isinstance(configs, (tuple, set, list)):
logger.error(
'An invalid configuration path (type={}) was '
'specified.'.format(type(configs)))
return False
# Iterate over our
for _config in configs:
if isinstance(_config, ConfigBase):
# Go ahead and just add our configuration into our list
self.configs.append(_config)
continue
elif not isinstance(_config, six.string_types):
logger.warning(
"An invalid configuration (type={}) was specified.".format(
type(_config)))
return_status = False
continue
logger.debug("Loading configuration: {}".format(_config))
# Instantiate ourselves an object, this function throws or
# returns None if it fails
instance = AppriseConfig.instantiate(_config, asset=asset, tag=tag)
if not isinstance(instance, ConfigBase):
return_status = False
continue
# Add our initialized plugin to our server listings
self.configs.append(instance)
# Return our status
return return_status
def servers(self, tag=None, cache=True):
"""
Returns all of our servers dynamically build based on parsed
configuration.
If a tag is specified, it applies to the configuration sources
themselves and not the notification services inside them.
This is for filtering the configuration files polled for
results.
"""
# Build our tag setup
# - top level entries are treated as an 'or'
# - second level (or more) entries are treated as 'and'
#
# examples:
# tag="tagA, tagB" = tagA or tagB
# tag=['tagA', 'tagB'] = tagA or tagB
# tag=[('tagA', 'tagC'), 'tagB'] = (tagA and tagC) or tagB
# tag=[('tagB', 'tagC')] = tagB and tagC
response = list()
for entry in self.configs:
# Apply our tag matching based on our defined logic
if tag is not None and not is_exclusive_match(
logic=tag, data=entry.tags):
continue
# Build ourselves a list of services dynamically and return the
# as a list
response.extend(entry.servers(cache=cache))
return response
@staticmethod
def instantiate(url, asset=None, tag=None, suppress_exceptions=True):
"""
Returns the instance of a instantiated configuration plugin based on
the provided Server URL. If the url fails to be parsed, then None
is returned.
"""
# Attempt to acquire the schema at the very least to allow our
# configuration based urls.
schema = GET_SCHEMA_RE.match(url)
if schema is None:
# Plan B is to assume we're dealing with a file
schema = config.ConfigFile.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 config.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 = config.SCHEMA_MAP[schema].parse_url(url)
if not results:
# Failed to parse the server URL
logger.warning('Unparseable URL {}.'.format(url))
return None
# Build a list of tags to associate with the newly added notifications
results['tag'] = set(parse_list(tag))
# Prepare our Asset Object
results['asset'] = \
asset if isinstance(asset, AppriseAsset) else AppriseAsset()
if suppress_exceptions:
try:
# Attempt to create an instance of our plugin using the parsed
# URL information
cfg_plugin = config.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
cfg_plugin = config.SCHEMA_MAP[results['schema']](**results)
return cfg_plugin
def clear(self):
"""
Empties our configuration list
"""
self.configs[:] = []
def server_pop(self, index):
"""
Removes an indexed Apprise Notification from the servers
"""
# Tracking variables
prev_offset = -1
offset = prev_offset
for entry in self.configs:
servers = entry.servers(cache=True)
if len(servers) > 0:
# Acquire a new maximum offset to work with
offset = prev_offset + len(servers)
if offset >= index:
# we can pop an notification from our config stack
return entry.pop(index if prev_offset == -1
else (index - prev_offset - 1))
# Update our old offset
prev_offset = offset
# If we reach here, then we indexed out of range
raise IndexError('list index out of range')
def pop(self, index):
"""
Removes an indexed Apprise Configuration from the stack and
returns it.
"""
# Remove our entry
return self.configs.pop(index)
def __getitem__(self, index):
"""
Returns the indexed config entry of a loaded apprise configuration
"""
return self.configs[index]
def __iter__(self):
"""
Returns an iterator to our config list
"""
return iter(self.configs)
def __len__(self):
"""
Returns the number of config entries loaded
"""
return len(self.configs)
Reference in New Issue View Git Blame Copy Permalink