extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2025-01-03 20:49:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

Refactored templating; introduced rto= and cto= URL parameters (#264)

Browse Source
This commit is contained in:
Chris Caron 2020-07-31 21:49:38 -04:00 committed by GitHub
parent 57cb5f38d2
commit a91064af8f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
64 changed files with 627 additions and 593 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

129
apprise/URLBase.py
View File

@ -42,6 +42,7 @@ except ImportError:
from urllib.parse import quote as _quote
from urllib.parse import urlencode as _urlencode
from .AppriseLocale import gettext_lazy as _
from .AppriseAsset import AppriseAsset
from .utils import parse_url
from .utils import parse_bool
@ -101,11 +102,11 @@ class URLBase(object):
# The connect timeout is the number of seconds Requests will wait for your
# client to establish a connection to a remote machine (corresponding to
# the connect()) call on the socket.
request_connect_timeout = 4.0
socket_connect_timeout = 4.0
# The read timeout is the number of seconds the client will wait for the
# server to send a response.
request_read_timeout = 2.5
socket_read_timeout = 4.0
# Handle
# Maintain a set of tags to associate with this specific notification
@ -117,6 +118,78 @@ class URLBase(object):
# Logging
logger = logging.getLogger(__name__)
# Define a default set of template arguments used for dynamically building
# details about our individual plugins for developers.
# Define object templates
templates = ()
# Provides a mapping of tokens, certain entries are fixed and automatically
# configured if found (such as schema, host, user, pass, and port)
template_tokens = {}
# Here is where we define all of the arguments we accept on the url
# such as: schema://whatever/?cto=5.0&rto=15
# These act the same way as tokens except they are optional and/or
# have default values set if mandatory. This rule must be followed
template_args = {
'verify': {
'name': _('Verify SSL'),
# SSL Certificate Authority Verification
'type': 'bool',
# Provide a default
'default': verify_certificate,
# look up default using the following parent class value at
# runtime.
'_lookup_default': 'verify_certificate',
},
'rto': {
'name': _('Socket Read Timeout'),
'type': 'float',
# Provide a default
'default': socket_read_timeout,
# look up default using the following parent class value at
# runtime. The variable name identified here (in this case
# socket_read_timeout) is checked and it's result is placed
# over-top of the 'default'. This is done because once a parent
# class inherits this one, the overflow_mode already set as a
# default 'could' be potentially over-ridden and changed to a
# different value.
'_lookup_default': 'socket_read_timeout',
},
'cto': {
'name': _('Socket Connect Timeout'),
'type': 'float',
# Provide a default
'default': socket_connect_timeout,
# look up default using the following parent class value at
# runtime. The variable name identified here (in this case
# socket_connect_timeout) is checked and it's result is placed
# over-top of the 'default'. This is done because once a parent
# class inherits this one, the overflow_mode already set as a
# default 'could' be potentially over-ridden and changed to a
# different value.
'_lookup_default': 'socket_connect_timeout',
},
}
# kwargs are dynamically built because a prefix causes us to parse the
# content slightly differently. The prefix is required and can be either
# a (+ or -). Below would handle the +key=value:
# {
# 'headers': {
# 'name': _('HTTP Header'),
# 'prefix': '+',
# 'type': 'string',
# },
# },
#
# In a kwarg situation, the 'key' is always presumed to be treated as
# a string. When the 'type' is defined, it is being defined to respect
# the 'value'.
template_kwargs = {}
def __init__(self, asset=None, **kwargs):
"""
Initialize some general logging and common server arguments that will
@ -141,6 +214,9 @@ class URLBase(object):
self.port = int(self.port)
except (TypeError, ValueError):
self.logger.warning(
'Invalid port number specified {}'
.format(self.port))
self.port = None
self.user = kwargs.get('user')
@ -153,6 +229,26 @@ class URLBase(object):
# Always unquote the password if it exists
self.password = URLBase.unquote(self.password)
# Store our Timeout Variables
if 'socket_read_timeout' in kwargs:
try:
self.socket_read_timeout = \
float(kwargs.get('socket_read_timeout'))
except (TypeError, ValueError):
self.logger.warning(
'Invalid socket read timeout (rto) was specified {}'
.format(kwargs.get('socket_read_timeout')))
if 'socket_connect_timeout' in kwargs:
try:
self.socket_connect_timeout = \
float(kwargs.get('socket_connect_timeout'))
except (TypeError, ValueError):
self.logger.warning(
'Invalid socket connect timeout (cto) was specified {}'
.format(kwargs.get('socket_connect_timeout')))
if 'tag' in kwargs:
# We want to associate some tags with our notification service.
# the code below gets the 'tag' argument if defined, otherwise
@ -481,7 +577,26 @@ class URLBase(object):
"""This is primarily used to fullfill the `timeout` keyword argument
that is used by requests.get() and requests.put() calls.
"""
return (self.request_connect_timeout, self.request_read_timeout)
return (self.socket_connect_timeout, self.socket_read_timeout)
def url_parameters(self, *args, **kwargs):
"""
Provides a default set of args to work with. This can greatly
simplify URL construction in the acommpanied url() function.
The following property returns a dictionary (of strings) containing
all of the parameters that can be set on a URL and managed through
this class.
"""
return {
# The socket read timeout
'rto': str(self.socket_read_timeout),
# The request/socket connect timeout
'cto': str(self.socket_connect_timeout),
# Certificate verification
'verify': 'yes' if self.verify_certificate else 'no',
}
@staticmethod
def parse_url(url, verify_host=True):
@ -528,6 +643,14 @@ class URLBase(object):
if 'user' in results['qsd']:
results['user'] = results['qsd']['user']
# Store our socket read timeout if specified
if 'rto' in results['qsd']:
results['socket_read_timeout'] = results['qsd']['rto']
# Store our socket connect timeout if specified
if 'cto' in results['qsd']:
results['socket_connect_timeout'] = results['qsd']['cto']
return results
@staticmethod

12
apprise/attachment/AttachFile.py
View File

@ -57,20 +57,20 @@ class AttachFile(AttachBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {}
# Define any URL parameters
params = {}
if self._mimetype:
# A mime-type was enforced
args['mime'] = self._mimetype
params['mime'] = self._mimetype
if self._name:
# A name was enforced
args['name'] = self._name
params['name'] = self._name
return 'file://{path}{args}'.format(
return 'file://{path}{params}'.format(
path=self.quote(self.dirty_path),
args='?{}'.format(self.urlencode(args)) if args else '',
params='?{}'.format(self.urlencode(params)) if params else '',
)
def download(self, **kwargs):

22
apprise/attachment/AttachHTTP.py
View File

@ -254,10 +254,8 @@ class AttachHTTP(AttachBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
# Prepare our cache value
if self.cache is not None:
@ -267,21 +265,21 @@ class AttachHTTP(AttachBase):
cache = int(self.cache)
# Set our cache value
args['cache'] = cache
params['cache'] = cache
if self._mimetype:
# A format was enforced
args['mime'] = self._mimetype
params['mime'] = self._mimetype
if self._name:
# A name was enforced
args['name'] = self._name
params['name'] = self._name
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Append our headers into our parameters
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Apply any remaining entries to our URL
args.update(self.qsd)
params.update(self.qsd)
# Determine Authentication
auth = ''
@ -298,14 +296,14 @@ class AttachHTTP(AttachBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}{fullpath}?{args}'.format(
return '{schema}://{auth}{hostname}{port}{fullpath}?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=self.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
fullpath=self.quote(self.fullpath, safe='/'),
args=self.urlencode(args),
params=self.urlencode(params),
)
@staticmethod

10
apprise/config/ConfigFile.py
View File

@ -69,19 +69,19 @@ class ConfigFile(ConfigBase):
else:
cache = int(self.cache)
# Define any arguments set
args = {
# Define any URL parameters
params = {
'encoding': self.encoding,
'cache': cache,
}
if self.config_format:
# A format was enforced; make sure it's passed back with the url
args['format'] = self.config_format
params['format'] = self.config_format
return 'file://{path}{args}'.format(
return 'file://{path}{params}'.format(
path=self.quote(self.path),
args='?{}'.format(self.urlencode(args)) if args else '',
params='?{}'.format(self.urlencode(params)) if params else '',
)
def read(self, **kwargs):

14
apprise/config/ConfigHTTP.py
View File

@ -100,18 +100,20 @@ class ConfigHTTP(ConfigBase):
cache = int(self.cache)
# Define any arguments set
args = {
'verify': 'yes' if self.verify_certificate else 'no',
params = {
'encoding': self.encoding,
'cache': cache,
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if self.config_format:
# A format was enforced; make sure it's passed back with the url
args['format'] = self.config_format
params['format'] = self.config_format
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Determine Authentication
auth = ''
@ -128,14 +130,14 @@ class ConfigHTTP(ConfigBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}{fullpath}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}{fullpath}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=self.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
fullpath=self.quote(self.fullpath, safe='/'),
args=self.urlencode(args),
params=self.urlencode(params),
)
def read(self, **kwargs):

58
apprise/plugins/NotifyBase.py
View File

@ -80,21 +80,11 @@ class NotifyBase(URLBase):
# use a <b> tag. The below causes the <b>title</b> to get generated:
default_html_tag_id = 'b'
# Define a default set of template arguments used for dynamically building
# details about our individual plugins for developers.
# Define object templates
templates = ()
# Provides a mapping of tokens, certain entries are fixed and automatically
# configured if found (such as schema, host, user, pass, and port)
template_tokens = {}
# Here is where we define all of the arguments we accept on the url
# such as: schema://whatever/?overflow=upstream&format=text
# These act the same way as tokens except they are optional and/or
# have default values set if mandatory. This rule must be followed
template_args = {
template_args = dict(URLBase.template_args, **{
'overflow': {
'name': _('Overflow Mode'),
'type': 'choice:string',
@ -119,34 +109,7 @@ class NotifyBase(URLBase):
# runtime.
'_lookup_default': 'notify_format',
},
'verify': {
'name': _('Verify SSL'),
# SSL Certificate Authority Verification
'type': 'bool',
# Provide a default
'default': URLBase.verify_certificate,
# look up default using the following parent class value at
# runtime.
'_lookup_default': 'verify_certificate',
},
}
# kwargs are dynamically built because a prefix causes us to parse the
# content slightly differently. The prefix is required and can be either
# a (+ or -). Below would handle the +key=value:
# {
# 'headers': {
# 'name': _('HTTP Header'),
# 'prefix': '+',
# 'type': 'string',
# },
# },
#
# In a kwarg situation, the 'key' is always presumed to be treated as
# a string. When the 'type' is defined, it is being defined to respect
# the 'value'.
template_kwargs = {}
})
def __init__(self, **kwargs):
"""
@ -368,6 +331,23 @@ class NotifyBase(URLBase):
raise NotImplementedError(
"send() is not implimented by the child class.")
def url_parameters(self, *args, **kwargs):
"""
Provides a default set of parameters to work with. This can greatly
simplify URL construction in the acommpanied url() function in all
defined plugin services.
"""
params = {
'format': self.notify_format,
'overflow': self.overflow_mode,
}
params.update(super(NotifyBase, self).url_parameters(*args, **kwargs))
# return default arguments
return kwargs
@staticmethod
def parse_url(url, verify_host=True):
"""Parses the URL and returns it broken apart into a dictionary.

14
apprise/plugins/NotifyBoxcar.py
View File

@ -320,15 +320,15 @@ class NotifyBoxcar(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{access}/{secret}/{targets}?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{access}/{secret}/{targets}?{params}'.format(
schema=self.secure_protocol,
access=self.pprint(self.access, privacy, safe=''),
secret=self.pprint(
@ -336,7 +336,7 @@ class NotifyBoxcar(NotifyBase):
targets='/'.join([
NotifyBoxcar.quote(x, safe='') for x in chain(
self.tags, self.device_tokens) if x != DEFAULT_TAG]),
args=NotifyBoxcar.urlencode(args),
params=NotifyBoxcar.urlencode(params),
)
@staticmethod

14
apprise/plugins/NotifyClickSend.py
View File

@ -272,14 +272,14 @@ class NotifyClickSend(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'batch': 'yes' if self.batch else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Setup Authentication
auth = '{user}:{password}@'.format(
user=NotifyClickSend.quote(self.user, safe=''),
@ -287,12 +287,12 @@ class NotifyClickSend(NotifyBase):
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
)
return '{schema}://{auth}{targets}?{args}'.format(
return '{schema}://{auth}{targets}?{params}'.format(
schema=self.secure_protocol,
auth=auth,
targets='/'.join(
[NotifyClickSend.quote(x, safe='') for x in self.targets]),
args=NotifyClickSend.urlencode(args),
params=NotifyClickSend.urlencode(params),
)
@staticmethod

18
apprise/plugins/NotifyD7Networks.py
View File

@ -395,28 +395,28 @@ class NotifyD7Networks(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'batch': 'yes' if self.batch else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if self.priority != self.template_args['priority']['default']:
args['priority'] = str(self.priority)
params['priority'] = str(self.priority)
if self.source:
args['from'] = self.source
params['from'] = self.source
return '{schema}://{user}:{password}@{targets}/?{args}'.format(
return '{schema}://{user}:{password}@{targets}/?{params}'.format(
schema=self.secure_protocol,
user=NotifyD7Networks.quote(self.user, safe=''),
password=self.pprint(
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
targets='/'.join(
[NotifyD7Networks.quote(x, safe='') for x in self.targets]),
args=NotifyD7Networks.urlencode(args))
params=NotifyD7Networks.urlencode(params))
@staticmethod
def parse_url(url):

18
apprise/plugins/NotifyDBus.py
View File

@ -355,27 +355,27 @@ class NotifyDBus(NotifyBase):
DBusUrgency.HIGH: 'high',
}
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'urgency': 'normal' if self.urgency not in _map
else _map[self.urgency],
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# x in (x,y) screen coordinates
if self.x_axis:
args['x'] = str(self.x_axis)
params['x'] = str(self.x_axis)
# y in (x,y) screen coordinates
if self.y_axis:
args['y'] = str(self.y_axis)
params['y'] = str(self.y_axis)
return '{schema}://_/?{args}'.format(
return '{schema}://_/?{params}'.format(
schema=self.schema,
args=NotifyDBus.urlencode(args),
params=NotifyDBus.urlencode(params),
)
@staticmethod

22
apprise/plugins/NotifyDiscord.py
View File

@ -406,23 +406,23 @@ class NotifyDiscord(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'tts': 'yes' if self.tts else 'no',
'avatar': 'yes' if self.avatar else 'no',
'footer': 'yes' if self.footer else 'no',
'footer_logo': 'yes' if self.footer_logo else 'no',
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{webhook_id}/{webhook_token}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{webhook_id}/{webhook_token}/?{params}'.format(
schema=self.secure_protocol,
webhook_id=self.pprint(self.webhook_id, privacy, safe=''),
webhook_token=self.pprint(self.webhook_token, privacy, safe=''),
args=NotifyDiscord.urlencode(args),
params=NotifyDiscord.urlencode(params),
)
@staticmethod
@ -504,16 +504,16 @@ class NotifyDiscord(NotifyBase):
r'^https?://discord(app)?\.com/api/webhooks/'
r'(?P<webhook_id>[0-9]+)/'
r'(?P<webhook_token>[A-Z0-9_-]+)/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
return NotifyDiscord.parse_url(
'{schema}://{webhook_id}/{webhook_token}/{args}'.format(
'{schema}://{webhook_id}/{webhook_token}/{params}'.format(
schema=NotifyDiscord.secure_protocol,
webhook_id=result.group('webhook_id'),
webhook_token=result.group('webhook_token'),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

18
apprise/plugins/NotifyEmail.py
View File

@ -687,26 +687,26 @@ class NotifyEmail(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define an URL parameters
params = {
'from': self.from_addr,
'name': self.from_name,
'mode': self.secure_mode,
'smtp': self.smtp_host,
'timeout': self.timeout,
'user': self.user,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if len(self.cc) > 0:
# Handle our Carbon Copy Addresses
args['cc'] = ','.join(self.cc)
params['cc'] = ','.join(self.cc)
if len(self.bcc) > 0:
# Handle our Blind Carbon Copy Addresses
args['bcc'] = ','.join(self.bcc)
params['bcc'] = ','.join(self.bcc)
# pull email suffix from username (if present)
user = None if not self.user else self.user.split('@')[0]
@ -734,7 +734,7 @@ class NotifyEmail(NotifyBase):
has_targets = \
not (len(self.targets) == 1 and self.targets[0] == self.from_addr)
return '{schema}://{auth}{hostname}{port}/{targets}?{args}'.format(
return '{schema}://{auth}{hostname}{port}/{targets}?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyEmail.quote(self.host, safe=''),
@ -742,7 +742,7 @@ class NotifyEmail(NotifyBase):
else ':{}'.format(self.port),
targets='' if not has_targets else '/'.join(
[NotifyEmail.quote(x, safe='') for x in self.targets]),
args=NotifyEmail.urlencode(args),
params=NotifyEmail.urlencode(params),
)
@staticmethod

14
apprise/plugins/NotifyEmby.py
View File

@ -596,14 +596,14 @@ class NotifyEmby(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'modal': 'yes' if self.modal else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine Authentication
auth = ''
if self.user and self.password:
@ -617,13 +617,13 @@ class NotifyEmby(NotifyBase):
user=NotifyEmby.quote(self.user, safe=''),
)
return '{schema}://{auth}{hostname}{port}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyEmby.quote(self.host, safe=''),
port='' if self.port is None or self.port == self.emby_default_port
else ':{}'.format(self.port),
args=NotifyEmby.urlencode(args),
params=NotifyEmby.urlencode(params),
)
@property

18
apprise/plugins/NotifyEnigma2.py
View File

@ -184,16 +184,16 @@ class NotifyEnigma2(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'timeout': str(self.timeout),
}
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Append our headers into our parameters
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine Authentication
auth = ''
@ -210,14 +210,14 @@ class NotifyEnigma2(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}{fullpath}?{args}'.format(
return '{schema}://{auth}{hostname}{port}{fullpath}?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyEnigma2.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
fullpath=NotifyEnigma2.quote(self.fullpath, safe='/'),
args=NotifyEnigma2.urlencode(args),
params=NotifyEnigma2.urlencode(params),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):

14
apprise/plugins/NotifyFaast.py
View File

@ -169,18 +169,18 @@ class NotifyFaast(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{authtoken}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{authtoken}/?{params}'.format(
schema=self.protocol,
authtoken=self.pprint(self.authtoken, privacy, safe=''),
args=NotifyFaast.urlencode(args),
params=NotifyFaast.urlencode(params),
)
@staticmethod

21
apprise/plugins/NotifyFlock.py
View File

@ -309,21 +309,22 @@ class NotifyFlock(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{token}/{targets}?{args}'\
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{token}/{targets}?{params}'\
.format(
schema=self.secure_protocol,
token=self.pprint(self.token, privacy, safe=''),
targets='/'.join(
[NotifyFlock.quote(target, safe='')
for target in self.targets]),
args=NotifyFlock.urlencode(args),
params=NotifyFlock.urlencode(params),
)
@staticmethod
@ -364,14 +365,14 @@ class NotifyFlock(NotifyBase):
result = re.match(
r'^https?://api\.flock\.com/hooks/sendMessage/'
r'(?P<token>[a-z0-9-]{24})/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
return NotifyFlock.parse_url(
'{schema}://{token}/{args}'.format(
'{schema}://{token}/{params}'.format(
schema=NotifyFlock.secure_protocol,
token=result.group('token'),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

14
apprise/plugins/NotifyGitter.py
View File

@ -367,20 +367,20 @@ class NotifyGitter(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{token}/{targets}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{token}/{targets}/?{params}'.format(
schema=self.secure_protocol,
token=self.pprint(self.token, privacy, safe=''),
targets='/'.join(
[NotifyGitter.quote(x, safe='') for x in self.targets]),
args=NotifyGitter.urlencode(args))
params=NotifyGitter.urlencode(params))
@staticmethod
def parse_url(url):

14
apprise/plugins/NotifyGnome.py
View File

@ -214,19 +214,19 @@ class NotifyGnome(NotifyBase):
GnomeUrgency.HIGH: 'high',
}
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'urgency': 'normal' if self.urgency not in _map
else _map[self.urgency],
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://_/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://_/?{params}'.format(
schema=self.protocol,
args=NotifyGnome.urlencode(args),
params=NotifyGnome.urlencode(params),
)
@staticmethod

20
apprise/plugins/NotifyGotify.py
View File

@ -80,11 +80,6 @@ class NotifyGotify(NotifyBase):
# Disable throttle rate
request_rate_per_sec = 0
# The connect timeout is the number of seconds Requests will wait for your
# client to establish a connection to a remote machine (corresponding to
# the connect()) call on the socket.
request_connect_timeout = 2.5
# Define object templates
templates = (
'{schema}://{host}/{token}',
@ -247,24 +242,25 @@ class NotifyGotify(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'priority': self.priority,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Our default port
default_port = 443 if self.secure else 80
return '{schema}://{hostname}{port}{fullpath}{token}/?{args}'.format(
return '{schema}://{hostname}{port}{fullpath}{token}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
hostname=NotifyGotify.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
fullpath=NotifyGotify.quote(self.fullpath, safe='/'),
token=self.pprint(self.token, privacy, safe=''),
args=NotifyGotify.urlencode(args),
params=NotifyGotify.urlencode(params),
)
@staticmethod

14
apprise/plugins/NotifyGrowl/__init__.py
View File

@ -282,18 +282,18 @@ class NotifyGrowl(NotifyBase):
GrowlPriority.EMERGENCY: 'emergency',
}
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'priority':
_map[GrowlPriority.NORMAL] if self.priority not in _map
else _map[self.priority],
'version': self.version,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
auth = ''
if self.user:
# The growl password is stored in the user field
@ -302,13 +302,13 @@ class NotifyGrowl(NotifyBase):
self.user, privacy, mode=PrivacyMode.Secret, safe=''),
)
return '{schema}://{auth}{hostname}{port}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyGrowl.quote(self.host, safe=''),
port='' if self.port is None or self.port == self.default_port
else ':{}'.format(self.port),
args=NotifyGrowl.urlencode(args),
params=NotifyGrowl.urlencode(params),
)
@staticmethod

24
apprise/plugins/NotifyIFTTT.py
View File

@ -291,23 +291,19 @@ class NotifyIFTTT(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
# Store any new key/value pairs added to our list
args.update({'+{}'.format(k): v for k, v in self.add_tokens})
args.update({'-{}'.format(k): '' for k in self.del_tokens})
params.update({'+{}'.format(k): v for k, v in self.add_tokens})
params.update({'-{}'.format(k): '' for k in self.del_tokens})
return '{schema}://{webhook_id}@{events}/?{args}'.format(
return '{schema}://{webhook_id}@{events}/?{params}'.format(
schema=self.secure_protocol,
webhook_id=self.pprint(self.webhook_id, privacy, safe=''),
events='/'.join([NotifyIFTTT.quote(x, safe='')
for x in self.events]),
args=NotifyIFTTT.urlencode(args),
params=NotifyIFTTT.urlencode(params),
)
@staticmethod
@ -357,16 +353,16 @@ class NotifyIFTTT(NotifyBase):
r'^https?://maker\.ifttt\.com/use/'
r'(?P<webhook_id>[A-Z0-9_-]+)'
r'/?(?P<events>([A-Z0-9_-]+/?)+)?'
r'/?(?P<args>\?.+)?$', url, re.I)
r'/?(?P<params>\?.+)?$', url, re.I)
if result:
return NotifyIFTTT.parse_url(
'{schema}://{webhook_id}{events}{args}'.format(
'{schema}://{webhook_id}{events}{params}'.format(
schema=NotifyIFTTT.secure_protocol,
webhook_id=result.group('webhook_id'),
events='' if not result.group('events')
else '@{}'.format(result.group('events')),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

16
apprise/plugins/NotifyJSON.py
View File

@ -128,15 +128,11 @@ class NotifyJSON(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Append our headers into our parameters
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Determine Authentication
auth = ''
@ -153,14 +149,14 @@ class NotifyJSON(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}{fullpath}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}{fullpath}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyJSON.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
fullpath=NotifyJSON.quote(self.fullpath, safe='/'),
args=NotifyJSON.urlencode(args),
params=NotifyJSON.urlencode(params),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):

14
apprise/plugins/NotifyJoin.py
View File

@ -332,23 +332,23 @@ class NotifyJoin(NotifyBase):
JoinPriority.EMERGENCY: 'emergency',
}
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'priority':
_map[self.template_args['priority']['default']]
if self.priority not in _map else _map[self.priority],
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{apikey}/{targets}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{apikey}/{targets}/?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
targets='/'.join([NotifyJoin.quote(x, safe='')
for x in self.targets]),
args=NotifyJoin.urlencode(args))
params=NotifyJoin.urlencode(params))
@staticmethod
def parse_url(url):

12
apprise/plugins/NotifyKavenegar.py
View File

@ -326,20 +326,16 @@ class NotifyKavenegar(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{source}{apikey}/{targets}?{args}'.format(
return '{schema}://{source}{apikey}/{targets}?{params}'.format(
schema=self.secure_protocol,
source='' if not self.source else '{}@'.format(self.source),
apikey=self.pprint(self.apikey, privacy, safe=''),
targets='/'.join(
[NotifyKavenegar.quote(x, safe='') for x in self.targets]),
args=NotifyKavenegar.urlencode(args))
params=NotifyKavenegar.urlencode(params))
@staticmethod
def parse_url(url):

12
apprise/plugins/NotifyKumulos.py
View File

@ -200,18 +200,14 @@ class NotifyKumulos(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{apikey}/{serverkey}/?{args}'.format(
return '{schema}://{apikey}/{serverkey}/?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
serverkey=self.pprint(self.serverkey, privacy, safe=''),
args=NotifyKumulos.urlencode(args),
params=NotifyKumulos.urlencode(params),
)
@staticmethod

18
apprise/plugins/NotifyMSG91.py
View File

@ -317,23 +317,23 @@ class NotifyMSG91(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'route': str(self.route),
}
if self.country:
args['country'] = str(self.country)
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{authkey}/{targets}/?{args}'.format(
if self.country:
params['country'] = str(self.country)
return '{schema}://{authkey}/{targets}/?{params}'.format(
schema=self.secure_protocol,
authkey=self.pprint(self.authkey, privacy, safe=''),
targets='/'.join(
[NotifyMSG91.quote(x, safe='') for x in self.targets]),
args=NotifyMSG91.urlencode(args))
params=NotifyMSG91.urlencode(params))
@staticmethod
def parse_url(url):

22
apprise/plugins/NotifyMSTeams.py
View File

@ -278,21 +278,21 @@ class NotifyMSTeams(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{token_a}/{token_b}/{token_c}/'\
'?{args}'.format(
'?{params}'.format(
schema=self.secure_protocol,
token_a=self.pprint(self.token_a, privacy, safe=''),
token_b=self.pprint(self.token_b, privacy, safe=''),
token_c=self.pprint(self.token_c, privacy, safe=''),
args=NotifyMSTeams.urlencode(args),
params=NotifyMSTeams.urlencode(params),
)
@staticmethod
@ -360,16 +360,16 @@ class NotifyMSTeams(NotifyBase):
r'IncomingWebhook/'
r'(?P<token_b>[A-Z0-9]+)/'
r'(?P<token_c>[A-Z0-9-]+)/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
return NotifyMSTeams.parse_url(
'{schema}://{token_a}/{token_b}/{token_c}/{args}'.format(
'{schema}://{token_a}/{token_b}/{token_c}/{params}'.format(
schema=NotifyMSTeams.secure_protocol,
token_a=result.group('token_a'),
token_b=result.group('token_b'),
token_c=result.group('token_c'),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

15
apprise/plugins/NotifyMacOSX.py
View File

@ -180,20 +180,21 @@ class NotifyMacOSX(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parametrs
params = {
'image': 'yes' if self.include_image else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if self.sound:
# Store our sound
args['sound'] = self.sound
params['sound'] = self.sound
return '{schema}://_/?{args}'.format(
return '{schema}://_/?{params}'.format(
schema=self.protocol,
args=NotifyMacOSX.urlencode(args),
params=NotifyMacOSX.urlencode(params),
)
@staticmethod

16
apprise/plugins/NotifyMailgun.py
View File

@ -315,26 +315,26 @@ class NotifyMailgun(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'region': self.region_name,
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if self.from_name is not None:
# from_name specified; pass it back on the url
args['name'] = self.from_name
params['name'] = self.from_name
return '{schema}://{user}@{host}/{apikey}/{targets}/?{args}'.format(
return '{schema}://{user}@{host}/{apikey}/{targets}/?{params}'.format(
schema=self.secure_protocol,
host=self.host,
user=NotifyMailgun.quote(self.user, safe=''),
apikey=self.pprint(self.apikey, privacy, safe=''),
targets='/'.join(
[NotifyMailgun.quote(x, safe='') for x in self.targets]),
args=NotifyMailgun.urlencode(args))
params=NotifyMailgun.urlencode(params))
@staticmethod
def parse_url(url):

22
apprise/plugins/NotifyMatrix.py
View File

@ -1011,15 +1011,15 @@ class NotifyMatrix(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
'mode': self.mode,
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine Authentication
auth = ''
if self.user and self.password:
@ -1036,14 +1036,14 @@ class NotifyMatrix(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}/{rooms}?{args}'.format(
return '{schema}://{auth}{hostname}{port}/{rooms}?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyMatrix.quote(self.host, safe=''),
port='' if self.port is None
or self.port == default_port else ':{}'.format(self.port),
rooms=NotifyMatrix.quote('/'.join(self.rooms)),
args=NotifyMatrix.urlencode(args),
params=NotifyMatrix.urlencode(params),
)
@staticmethod
@ -1119,16 +1119,16 @@ class NotifyMatrix(NotifyBase):
result = re.match(
r'^https?://webhooks\.t2bot\.io/api/v1/matrix/hook/'
r'(?P<webhook_token>[A-Z0-9_-]+)/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
mode = 'mode={}'.format(MatrixWebhookMode.T2BOT)
return NotifyMatrix.parse_url(
'{schema}://{webhook_token}/{args}'.format(
'{schema}://{webhook_token}/{params}'.format(
schema=NotifyMatrix.secure_protocol,
webhook_token=result.group('webhook_token'),
args='?{}'.format(mode) if not result.group('args')
else '{}&{}'.format(result.group('args'), mode)))
params='?{}'.format(mode) if not result.group('params')
else '{}&{}'.format(result.group('params'), mode)))
return None

16
apprise/plugins/NotifyMatterMost.py
View File

@ -278,19 +278,19 @@ class NotifyMatterMost(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if self.channels:
# historically the value only accepted one channel and is
# therefore identified as 'channel'. Channels have always been
# optional, so that is why this setting is nested in an if block
args['channel'] = ','.join(self.channels)
params['channel'] = ','.join(self.channels)
default_port = 443 if self.secure else self.default_port
default_schema = self.secure_protocol if self.secure else self.protocol
@ -304,7 +304,7 @@ class NotifyMatterMost(NotifyBase):
return \
'{schema}://{botname}{hostname}{port}{fullpath}{authtoken}' \
'/?{args}'.format(
'/?{params}'.format(
schema=default_schema,
botname=botname,
hostname=NotifyMatterMost.quote(self.host, safe=''),
@ -313,7 +313,7 @@ class NotifyMatterMost(NotifyBase):
fullpath='/' if not self.fullpath else '{}/'.format(
NotifyMatterMost.quote(self.fullpath, safe='/')),
authtoken=self.pprint(self.authtoken, privacy, safe=''),
args=NotifyMatterMost.urlencode(args),
params=NotifyMatterMost.urlencode(params),
)
@staticmethod

12
apprise/plugins/NotifyMessageBird.py
View File

@ -314,20 +314,16 @@ class NotifyMessageBird(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{apikey}/{source}/{targets}/?{args}'.format(
return '{schema}://{apikey}/{source}/{targets}/?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
source=self.source,
targets='/'.join(
[NotifyMessageBird.quote(x, safe='') for x in self.targets]),
args=NotifyMessageBird.urlencode(args))
params=NotifyMessageBird.urlencode(params))
@staticmethod
def parse_url(url):

14
apprise/plugins/NotifyNexmo.py
View File

@ -325,15 +325,15 @@ class NotifyNexmo(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'ttl': str(self.ttl),
}
return '{schema}://{key}:{secret}@{source}/{targets}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{key}:{secret}@{source}/{targets}/?{params}'.format(
schema=self.secure_protocol,
key=self.pprint(self.apikey, privacy, safe=''),
secret=self.pprint(
@ -341,7 +341,7 @@ class NotifyNexmo(NotifyBase):
source=NotifyNexmo.quote(self.source, safe=''),
targets='/'.join(
[NotifyNexmo.quote(x, safe='') for x in self.targets]),
args=NotifyNexmo.urlencode(args))
params=NotifyNexmo.urlencode(params))
@staticmethod
def parse_url(url):

16
apprise/plugins/NotifyNextcloud.py
View File

@ -227,15 +227,11 @@ class NotifyNextcloud(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Create URL parameters from our headers
params = {'+{}'.format(k): v for k, v in self.headers.items()}
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
# Determine Authentication
auth = ''
@ -252,7 +248,7 @@ class NotifyNextcloud(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}/{targets}?{args}' \
return '{schema}://{auth}{hostname}{port}/{targets}?{params}' \
.format(
schema=self.secure_protocol
if self.secure else self.protocol,
@ -262,7 +258,7 @@ class NotifyNextcloud(NotifyBase):
else ':{}'.format(self.port),
targets='/'.join([NotifyNextcloud.quote(x)
for x in self.targets]),
args=NotifyNextcloud.urlencode(args),
params=NotifyNextcloud.urlencode(params),
)
@staticmethod

28
apprise/plugins/NotifyNotica.py
View File

@ -280,25 +280,21 @@ class NotifyNotica(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
if self.mode == NoticaMode.OFFICIAL:
# Official URLs are easy to assemble
return '{schema}://{token}/?{args}'.format(
return '{schema}://{token}/?{params}'.format(
schema=self.protocol,
token=self.pprint(self.token, privacy, safe=''),
args=NotifyNotica.urlencode(args),
params=NotifyNotica.urlencode(params),
)
# If we reach here then we are assembling a self hosted URL
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Append URL parameters from our headers
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Authorization can be used for self-hosted sollutions
auth = ''
@ -317,7 +313,7 @@ class NotifyNotica(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}{fullpath}{token}/?{args}' \
return '{schema}://{auth}{hostname}{port}{fullpath}{token}/?{params}' \
.format(
schema=self.secure_protocol
if self.secure else self.protocol,
@ -328,7 +324,7 @@ class NotifyNotica(NotifyBase):
fullpath=NotifyNotica.quote(
self.fullpath, safe='/'),
token=self.pprint(self.token, privacy, safe=''),
args=NotifyNotica.urlencode(args),
params=NotifyNotica.urlencode(params),
)
@staticmethod
@ -382,14 +378,14 @@ class NotifyNotica(NotifyBase):
result = re.match(
r'^https?://notica\.us/?'
r'\??(?P<token>[^&]+)([&\s]*(?P<args>.+))?$', url, re.I)
r'\??(?P<token>[^&]+)([&\s]*(?P<params>.+))?$', url, re.I)
if result:
return NotifyNotica.parse_url(
'{schema}://{token}/{args}'.format(
'{schema}://{token}/{params}'.format(
schema=NotifyNotica.protocol,
token=result.group('token'),
args='' if not result.group('args')
else '?{}'.format(result.group('args'))))
params='' if not result.group('params')
else '?{}'.format(result.group('params'))))
return None

22
apprise/plugins/NotifyNotifico.py
View File

@ -199,20 +199,20 @@ class NotifyNotifico(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'color': 'yes' if self.color else 'no',
'prefix': 'yes' if self.prefix else 'no',
}
return '{schema}://{proj}/{hook}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{proj}/{hook}/?{params}'.format(
schema=self.secure_protocol,
proj=self.pprint(self.project_id, privacy, safe=''),
hook=self.pprint(self.msghook, privacy, safe=''),
args=NotifyNotifico.urlencode(args),
params=NotifyNotifico.urlencode(params),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
@ -365,15 +365,15 @@ class NotifyNotifico(NotifyBase):
r'^https?://n\.tkte\.ch/h/'
r'(?P<proj>[0-9]+)/'
r'(?P<hook>[A-Z0-9]+)/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
return NotifyNotifico.parse_url(
'{schema}://{proj}/{hook}/{args}'.format(
'{schema}://{proj}/{hook}/{params}'.format(
schema=NotifyNotifico.secure_protocol,
proj=result.group('proj'),
hook=result.group('hook'),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

12
apprise/plugins/NotifyOffice365.py
View File

@ -450,15 +450,11 @@ class NotifyOffice365(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{tenant}:{email}/{client_id}/{secret}' \
'/{targets}/?{args}'.format(
'/{targets}/?{params}'.format(
schema=self.secure_protocol,
tenant=self.pprint(self.tenant, privacy, safe=''),
# email does not need to be escaped because it should
@ -470,7 +466,7 @@ class NotifyOffice365(NotifyBase):
safe=''),
targets='/'.join(
[NotifyOffice365.quote(x, safe='') for x in self.targets]),
args=NotifyOffice365.urlencode(args))
params=NotifyOffice365.urlencode(params))
@staticmethod
def parse_url(url):

14
apprise/plugins/NotifyPopcornNotify.py
View File

@ -255,20 +255,20 @@ class NotifyPopcornNotify(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'batch': 'yes' if self.batch else 'no',
}
return '{schema}://{apikey}/{targets}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{apikey}/{targets}/?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
targets='/'.join(
[NotifyPopcornNotify.quote(x, safe='') for x in self.targets]),
args=NotifyPopcornNotify.urlencode(args))
params=NotifyPopcornNotify.urlencode(params))
@staticmethod
def parse_url(url):

14
apprise/plugins/NotifyProwl.py
View File

@ -237,20 +237,20 @@ class NotifyProwl(NotifyBase):
ProwlPriority.EMERGENCY: 'emergency',
}
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'priority': 'normal' if self.priority not in _map
else _map[self.priority],
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{apikey}/{providerkey}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{apikey}/{providerkey}/?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
providerkey=self.pprint(self.providerkey, privacy, safe=''),
args=NotifyProwl.urlencode(args),
params=NotifyProwl.urlencode(params),
)
@staticmethod

12
apprise/plugins/NotifyPushBullet.py
View File

@ -376,12 +376,8 @@ class NotifyPushBullet(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
targets = '/'.join([NotifyPushBullet.quote(x) for x in self.targets])
if targets == PUSHBULLET_SEND_TO_ALL:
@ -389,11 +385,11 @@ class NotifyPushBullet(NotifyBase):
# it from the recipients list
targets = ''
return '{schema}://{accesstoken}/{targets}/?{args}'.format(
return '{schema}://{accesstoken}/{targets}/?{params}'.format(
schema=self.secure_protocol,
accesstoken=self.pprint(self.accesstoken, privacy, safe=''),
targets=targets,
args=NotifyPushBullet.urlencode(args))
params=NotifyPushBullet.urlencode(params))
@staticmethod
def parse_url(url):

18
apprise/plugins/NotifyPushSafer.py
View File

@ -757,29 +757,25 @@ class NotifyPushSafer(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
if self.priority is not None:
# Store our priority; but only if it was specified
args['priority'] = \
params['priority'] = \
next((key for key, value in PUSHSAFER_PRIORITY_MAP.items()
if value == self.priority),
DEFAULT_PRIORITY) # pragma: no cover
if self.sound is not None:
# Store our sound; but only if it was specified
args['sound'] = \
params['sound'] = \
next((key for key, value in PUSHSAFER_SOUND_MAP.items()
if value == self.sound), '') # pragma: no cover
if self.vibration is not None:
# Store our vibration; but only if it was specified
args['vibration'] = str(self.vibration)
params['vibration'] = str(self.vibration)
targets = '/'.join([NotifyPushSafer.quote(x) for x in self.targets])
if targets == PUSHSAFER_SEND_TO_ALL:
@ -787,11 +783,11 @@ class NotifyPushSafer(NotifyBase):
# it from the recipients list
targets = ''
return '{schema}://{privatekey}/{targets}?{args}'.format(
return '{schema}://{privatekey}/{targets}?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
privatekey=self.pprint(self.privatekey, privacy, safe=''),
targets=targets,
args=NotifyPushSafer.urlencode(args))
params=NotifyPushSafer.urlencode(params))
@staticmethod
def parse_url(url):

12
apprise/plugins/NotifyPushed.py
View File

@ -305,14 +305,10 @@ class NotifyPushed(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{app_key}/{app_secret}/{targets}/?{args}'.format(
return '{schema}://{app_key}/{app_secret}/{targets}/?{params}'.format(
schema=self.secure_protocol,
app_key=self.pprint(self.app_key, privacy, safe=''),
app_secret=self.pprint(
@ -324,7 +320,7 @@ class NotifyPushed(NotifyBase):
# Users are prefixed with an @ symbol
['@{}'.format(x) for x in self.users],
)]),
args=NotifyPushed.urlencode(args))
params=NotifyPushed.urlencode(params))
@staticmethod
def parse_url(url):

12
apprise/plugins/NotifyPushjet.py
View File

@ -123,12 +123,8 @@ class NotifyPushjet(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
default_port = 443 if self.secure else 80
@ -141,7 +137,7 @@ class NotifyPushjet(NotifyBase):
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
)
return '{schema}://{auth}{hostname}{port}/{secret}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}/{secret}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyPushjet.quote(self.host, safe=''),
@ -149,7 +145,7 @@ class NotifyPushjet(NotifyBase):
else ':{}'.format(self.port),
secret=self.pprint(
self.secret_key, privacy, mode=PrivacyMode.Secret, safe=''),
args=NotifyPushjet.urlencode(args),
params=NotifyPushjet.urlencode(params),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):

17
apprise/plugins/NotifyPushover.py
View File

@ -497,19 +497,20 @@ class NotifyPushover(NotifyBase):
PushoverPriority.EMERGENCY: 'emergency',
}
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'priority':
_map[self.template_args['priority']['default']]
if self.priority not in _map else _map[self.priority],
'verify': 'yes' if self.verify_certificate else 'no',
}
# Only add expire and retry for emergency messages,
# pushover ignores for all other priorities
if self.priority == PushoverPriority.EMERGENCY:
args.update({'expire': self.expire, 'retry': self.retry})
params.update({'expire': self.expire, 'retry': self.retry})
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Escape our devices
devices = '/'.join([NotifyPushover.quote(x, safe='')
@ -520,12 +521,12 @@ class NotifyPushover(NotifyBase):
# it from the devices list
devices = ''
return '{schema}://{user_key}@{token}/{devices}/?{args}'.format(
return '{schema}://{user_key}@{token}/{devices}/?{params}'.format(
schema=self.secure_protocol,
user_key=self.pprint(self.user_key, privacy, safe=''),
token=self.pprint(self.token, privacy, safe=''),
devices=devices,
args=NotifyPushover.urlencode(args))
params=NotifyPushover.urlencode(params))
@staticmethod
def parse_url(url):

14
apprise/plugins/NotifyRocketChat.py
View File

@ -285,15 +285,15 @@ class NotifyRocketChat(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'avatar': 'yes' if self.avatar else 'no',
'mode': self.mode,
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine Authentication
if self.mode == RocketChatAuthMode.BASIC:
auth = '{user}:{password}@'.format(
@ -310,7 +310,7 @@ class NotifyRocketChat(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}/{targets}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}/{targets}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyRocketChat.quote(self.host, safe=''),
@ -325,7 +325,7 @@ class NotifyRocketChat(NotifyBase):
# Users
['@{}'.format(x) for x in self.users],
)]),
args=NotifyRocketChat.urlencode(args),
params=NotifyRocketChat.urlencode(params),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):

22
apprise/plugins/NotifyRyver.py
View File

@ -274,15 +274,15 @@ class NotifyRyver(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'mode': self.mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine if there is a botname present
botname = ''
if self.user:
@ -290,12 +290,12 @@ class NotifyRyver(NotifyBase):
botname=NotifyRyver.quote(self.user, safe=''),
)
return '{schema}://{botname}{organization}/{token}/?{args}'.format(
return '{schema}://{botname}{organization}/{token}/?{params}'.format(
schema=self.secure_protocol,
botname=botname,
organization=NotifyRyver.quote(self.organization, safe=''),
token=self.pprint(self.token, privacy, safe=''),
args=NotifyRyver.urlencode(args),
params=NotifyRyver.urlencode(params),
)
@staticmethod
@ -353,15 +353,15 @@ class NotifyRyver(NotifyBase):
result = re.match(
r'^https?://(?P<org>[A-Z0-9_-]+)\.ryver\.com/application/webhook/'
r'(?P<webhook_token>[A-Z0-9]+)/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
return NotifyRyver.parse_url(
'{schema}://{org}/{webhook_token}/{args}'.format(
'{schema}://{org}/{webhook_token}/{params}'.format(
schema=NotifyRyver.secure_protocol,
org=result.group('org'),
webhook_token=result.group('webhook_token'),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

12
apprise/plugins/NotifySNS.py
View File

@ -580,15 +580,11 @@ class NotifySNS(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{key_id}/{key_secret}/{region}/{targets}/'\
'?{args}'.format(
'?{params}'.format(
schema=self.secure_protocol,
key_id=self.pprint(self.aws_access_key_id, privacy, safe=''),
key_secret=self.pprint(
@ -602,7 +598,7 @@ class NotifySNS(NotifyBase):
# Topics are prefixed with a pound/hashtag symbol
['#{}'.format(x) for x in self.topics],
)]),
args=NotifySNS.urlencode(args),
params=NotifySNS.urlencode(params),
)
@staticmethod

24
apprise/plugins/NotifySendGrid.py
View File

@ -245,41 +245,37 @@ class NotifySendGrid(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
if len(self.cc) > 0:
# Handle our Carbon Copy Addresses
args['cc'] = ','.join(self.cc)
params['cc'] = ','.join(self.cc)
if len(self.bcc) > 0:
# Handle our Blind Carbon Copy Addresses
args['bcc'] = ','.join(self.bcc)
params['bcc'] = ','.join(self.bcc)
if self.template:
# Handle our Template ID if if was specified
args['template'] = self.template
params['template'] = self.template
# Append our template_data into our args
args.update({'+{}'.format(k): v
for k, v in self.template_data.items()})
# Append our template_data into our parameter list
params.update(
{'+{}'.format(k): v for k, v in self.template_data.items()})
# a simple boolean check as to whether we display our target emails
# or not
has_targets = \
not (len(self.targets) == 1 and self.targets[0] == self.from_email)
return '{schema}://{apikey}:{from_email}/{targets}?{args}'.format(
return '{schema}://{apikey}:{from_email}/{targets}?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
from_email=self.quote(self.from_email, safe='@'),
targets='' if not has_targets else '/'.join(
[NotifySendGrid.quote(x, safe='') for x in self.targets]),
args=NotifySendGrid.urlencode(args),
params=NotifySendGrid.urlencode(params),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):

14
apprise/plugins/NotifySimplePush.py
View File

@ -286,15 +286,11 @@ class NotifySimplePush(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
if self.event:
args['event'] = self.event
params['event'] = self.event
# Determine Authentication
auth = ''
@ -306,11 +302,11 @@ class NotifySimplePush(NotifyBase):
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
)
return '{schema}://{auth}{apikey}/?{args}'.format(
return '{schema}://{auth}{apikey}/?{params}'.format(
schema=self.secure_protocol,
auth=auth,
apikey=self.pprint(self.apikey, privacy, safe=''),
args=NotifySimplePush.urlencode(args),
params=NotifySimplePush.urlencode(params),
)
@staticmethod

14
apprise/plugins/NotifySinch.py
View File

@ -400,15 +400,15 @@ class NotifySinch(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
# Define any URL parameters
params = {
'region': self.region,
}
return '{schema}://{spi}:{token}@{source}/{targets}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{spi}:{token}@{source}/{targets}/?{params}'.format(
schema=self.secure_protocol,
spi=self.pprint(
self.service_plan_id, privacy, mode=PrivacyMode.Tail, safe=''),
@ -416,7 +416,7 @@ class NotifySinch(NotifyBase):
source=NotifySinch.quote(self.source, safe=''),
targets='/'.join(
[NotifySinch.quote(x, safe='') for x in self.targets]),
args=NotifySinch.urlencode(args))
params=NotifySinch.urlencode(params))
@staticmethod
def parse_url(url):

26
apprise/plugins/NotifySlack.py
View File

@ -649,15 +649,15 @@ class NotifySlack(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'footer': 'yes' if self.include_footer else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if self.mode == SlackMode.WEBHOOK:
# Determine if there is a botname present
botname = ''
@ -667,7 +667,7 @@ class NotifySlack(NotifyBase):
)
return '{schema}://{botname}{token_a}/{token_b}/{token_c}/'\
'{targets}/?{args}'.format(
'{targets}/?{params}'.format(
schema=self.secure_protocol,
botname=botname,
token_a=self.pprint(self.token_a, privacy, safe=''),
@ -676,16 +676,16 @@ class NotifySlack(NotifyBase):
targets='/'.join(
[NotifySlack.quote(x, safe='')
for x in self.channels]),
args=NotifySlack.urlencode(args),
params=NotifySlack.urlencode(params),
)
# else -> self.mode == SlackMode.BOT:
return '{schema}://{access_token}/{targets}/'\
'?{args}'.format(
'?{params}'.format(
schema=self.secure_protocol,
access_token=self.pprint(self.access_token, privacy, safe=''),
targets='/'.join(
[NotifySlack.quote(x, safe='') for x in self.channels]),
args=NotifySlack.urlencode(args),
params=NotifySlack.urlencode(params),
)
@staticmethod
@ -761,16 +761,16 @@ class NotifySlack(NotifyBase):
r'(?P<token_a>[A-Z0-9]+)/'
r'(?P<token_b>[A-Z0-9]+)/'
r'(?P<token_c>[A-Z0-9]+)/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
return NotifySlack.parse_url(
'{schema}://{token_a}/{token_b}/{token_c}/{args}'.format(
'{schema}://{token_a}/{token_b}/{token_c}/{params}'.format(
schema=NotifySlack.secure_protocol,
token_a=result.group('token_a'),
token_b=result.group('token_b'),
token_c=result.group('token_c'),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

14
apprise/plugins/NotifySyslog.py
View File

@ -233,21 +233,21 @@ class NotifySyslog(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
# Define any URL parameters
params = {
'logperror': 'yes' if self.log_perror else 'no',
'logpid': 'yes' if self.log_pid else 'no',
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://{facility}/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{facility}/?{params}'.format(
facility=self.template_tokens['facility']['default']
if self.facility not in SYSLOG_FACILITY_RMAP
else SYSLOG_FACILITY_RMAP[self.facility],
schema=self.secure_protocol,
args=NotifySyslog.urlencode(args),
params=NotifySyslog.urlencode(params),
)
@staticmethod

12
apprise/plugins/NotifyTechulusPush.py
View File

@ -186,17 +186,13 @@ class NotifyTechulusPush(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{apikey}/?{args}'.format(
return '{schema}://{apikey}/?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
args=NotifyTechulusPush.urlencode(args),
params=NotifyTechulusPush.urlencode(params),
)
@staticmethod

14
apprise/plugins/NotifyTelegram.py
View File

@ -666,23 +666,23 @@ class NotifyTelegram(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': self.include_image,
'verify': 'yes' if self.verify_certificate else 'no',
'detect': 'yes' if self.detect_owner else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# No need to check the user token because the user automatically gets
# appended into the list of chat ids
return '{schema}://{bot_token}/{targets}/?{args}'.format(
return '{schema}://{bot_token}/{targets}/?{params}'.format(
schema=self.secure_protocol,
bot_token=self.pprint(self.bot_token, privacy, safe=''),
targets='/'.join(
[NotifyTelegram.quote('@{}'.format(x)) for x in self.targets]),
args=NotifyTelegram.urlencode(args))
params=NotifyTelegram.urlencode(params))
@staticmethod
def parse_url(url):

12
apprise/plugins/NotifyTwilio.py
View File

@ -368,14 +368,10 @@ class NotifyTwilio(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{sid}:{token}@{source}/{targets}/?{args}'.format(
return '{schema}://{sid}:{token}@{source}/{targets}/?{params}'.format(
schema=self.secure_protocol,
sid=self.pprint(
self.account_sid, privacy, mode=PrivacyMode.Tail, safe=''),
@ -383,7 +379,7 @@ class NotifyTwilio(NotifyBase):
source=NotifyTwilio.quote(self.source, safe=''),
targets='/'.join(
[NotifyTwilio.quote(x, safe='') for x in self.targets]),
args=NotifyTwilio.urlencode(args))
params=NotifyTwilio.urlencode(params))
@staticmethod
def parse_url(url):

39
apprise/plugins/NotifyTwist.py
View File

@ -229,28 +229,25 @@ class NotifyTwist(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{password}:{user}@{host}/{targets}/?{args}'.format(
schema=self.secure_protocol,
password=self.pprint(
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
user=self.quote(self.user, safe=''),
host=self.host,
targets='/'.join(
[NotifyTwist.quote(x, safe='') for x in chain(
# Channels are prefixed with a pound/hashtag symbol
['#{}'.format(x) for x in self.channels],
# Channel IDs
self.channel_ids,
)]),
args=NotifyTwist.urlencode(args),
)
return '{schema}://{password}:{user}@{host}/{targets}/' \
'?{params}'.format(
schema=self.secure_protocol,
password=self.pprint(
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
user=self.quote(self.user, safe=''),
host=self.host,
targets='/'.join(
[NotifyTwist.quote(x, safe='') for x in chain(
# Channels are prefixed with a pound/hashtag symbol
['#{}'.format(x) for x in self.channels],
# Channel IDs
self.channel_ids,
)]),
params=NotifyTwist.urlencode(params),
)
def login(self):
"""

18
apprise/plugins/NotifyTwitter.py
View File

@ -579,20 +579,20 @@ class NotifyTwitter(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'mode': self.mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
if len(self.targets) > 0:
args['to'] = ','.join([NotifyTwitter.quote(x, safe='')
for x in self.targets])
params['to'] = ','.join(
[NotifyTwitter.quote(x, safe='') for x in self.targets])
return '{schema}://{ckey}/{csecret}/{akey}/{asecret}' \
'/{targets}/?{args}'.format(
'/{targets}/?{params}'.format(
schema=self.secure_protocol[0],
ckey=self.pprint(self.ckey, privacy, safe=''),
csecret=self.pprint(
@ -603,7 +603,7 @@ class NotifyTwitter(NotifyBase):
targets='/'.join(
[NotifyTwitter.quote('@{}'.format(target), safe='')
for target in self.targets]),
args=NotifyTwitter.urlencode(args))
params=NotifyTwitter.urlencode(params))
@staticmethod
def parse_url(url):

20
apprise/plugins/NotifyWebexTeams.py
View File

@ -209,17 +209,13 @@ class NotifyWebexTeams(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
return '{schema}://{token}/?{args}'.format(
return '{schema}://{token}/?{params}'.format(
schema=self.secure_protocol,
token=self.pprint(self.token, privacy, safe=''),
args=NotifyWebexTeams.urlencode(args),
params=NotifyWebexTeams.urlencode(params),
)
@staticmethod
@ -249,14 +245,14 @@ class NotifyWebexTeams(NotifyBase):
result = re.match(
r'^https?://api\.ciscospark\.com/v[1-9][0-9]*/webhooks/incoming/'
r'(?P<webhook_token>[A-Z0-9_-]+)/?'
r'(?P<args>\?.+)?$', url, re.I)
r'(?P<params>\?.+)?$', url, re.I)
if result:
return NotifyWebexTeams.parse_url(
'{schema}://{webhook_token}/{args}'.format(
'{schema}://{webhook_token}/{params}'.format(
schema=NotifyWebexTeams.secure_protocol,
webhook_token=result.group('webhook_token'),
args='' if not result.group('args')
else result.group('args')))
params='' if not result.group('params')
else result.group('params')))
return None

14
apprise/plugins/NotifyWindows.py
View File

@ -223,18 +223,18 @@ class NotifyWindows(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'duration': str(self.duration),
'verify': 'yes' if self.verify_certificate else 'no',
}
return '{schema}://_/?{args}'.format(
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://_/?{params}'.format(
schema=self.protocol,
args=NotifyWindows.urlencode(args),
params=NotifyWindows.urlencode(params),
)
@staticmethod

14
apprise/plugins/NotifyXBMC.py
View File

@ -303,15 +303,15 @@ class NotifyXBMC(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
# Define any URL parameters
params = {
'image': 'yes' if self.include_image else 'no',
'duration': str(self.duration),
'verify': 'yes' if self.verify_certificate else 'no',
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine Authentication
auth = ''
if self.user and self.password:
@ -332,13 +332,13 @@ class NotifyXBMC(NotifyBase):
# Append 's' to schema
default_schema += 's'
return '{schema}://{auth}{hostname}{port}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}/?{params}'.format(
schema=default_schema,
auth=auth,
hostname=NotifyXBMC.quote(self.host, safe=''),
port='' if not self.port or self.port == default_port
else ':{}'.format(self.port),
args=NotifyXBMC.urlencode(args),
params=NotifyXBMC.urlencode(params),
)
@staticmethod

16
apprise/plugins/NotifyXML.py
View File

@ -143,15 +143,11 @@ class NotifyXML(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Store our defined headers into our URL parameters
params = {'+{}'.format(k): v for k, v in self.headers.items()}
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine Authentication
auth = ''
@ -168,14 +164,14 @@ class NotifyXML(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}{fullpath}/?{args}'.format(
return '{schema}://{auth}{hostname}{port}{fullpath}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyXML.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
fullpath=NotifyXML.quote(self.fullpath, safe='/'),
args=NotifyXML.urlencode(args),
params=NotifyXML.urlencode(params),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):

16
apprise/plugins/NotifyXMPP/__init__.py
View File

@ -272,20 +272,16 @@ class NotifyXMPP(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
if self.jid:
args['jid'] = self.jid
params['jid'] = self.jid
if self.xep:
# xep are integers, so we need to just iterate over a list and
# switch them to a string
args['xep'] = ','.join([str(xep) for xep in self.xep])
params['xep'] = ','.join([str(xep) for xep in self.xep])
# Target JID(s) can clash with our existing paths, so we just use comma
# and/or space as a delimiters - %20 = space
@ -307,14 +303,14 @@ class NotifyXMPP(NotifyBase):
self.password if self.password else self.user, privacy,
mode=PrivacyMode.Secret, safe='')
return '{schema}://{auth}@{hostname}{port}/{jids}?{args}'.format(
return '{schema}://{auth}@{hostname}{port}/{jids}?{params}'.format(
auth=auth,
schema=default_schema,
hostname=NotifyXMPP.quote(self.host, safe=''),
port='' if not self.port or self.port == default_port
else ':{}'.format(self.port),
jids=jids,
args=NotifyXMPP.urlencode(args),
params=NotifyXMPP.urlencode(params),
)
@staticmethod

12
apprise/plugins/NotifyZulip.py
View File

@ -328,12 +328,8 @@ class NotifyZulip(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
# simplify our organization in our URL if we can
organization = '{}{}'.format(
@ -342,14 +338,14 @@ class NotifyZulip(NotifyBase):
if self.hostname != self.default_hostname else '')
return '{schema}://{botname}@{org}/{token}/' \
'{targets}?{args}'.format(
'{targets}?{params}'.format(
schema=self.secure_protocol,
botname=NotifyZulip.quote(self.botname, safe=''),
org=NotifyZulip.quote(organization, safe=''),
token=self.pprint(self.token, privacy, safe=''),
targets='/'.join(
[NotifyZulip.quote(x, safe='') for x in self.targets]),
args=NotifyZulip.urlencode(args),
params=NotifyZulip.urlencode(params),
)
@staticmethod

25
test/test_api.py
View File

@ -303,6 +303,23 @@ def test_apprise():
a.clear()
assert len(a) == 0
# Test our socket details
# rto = Socket Read Timeout
# cto = Socket Connect Timeout
plugin = a.instantiate('good://localhost?rto=5.1&cto=10')
assert isinstance(plugin, NotifyBase)
assert plugin.socket_connect_timeout == 10.0
assert plugin.socket_read_timeout == 5.1
plugin = a.instantiate('good://localhost?rto=invalid&cto=invalid')
assert isinstance(plugin, NotifyBase)
assert plugin.socket_connect_timeout == URLBase.socket_connect_timeout
assert plugin.socket_read_timeout == URLBase.socket_read_timeout
# Reset our object
a.clear()
assert len(a) == 0
# Instantiate a bad object
plugin = a.instantiate(object, tag="bad_object")
assert plugin is None
@ -1024,10 +1041,12 @@ def test_apprise_details_plugin_verification():
# Define acceptable map_to arguments that can be tied in with the
# kwargs function definitions.
valid_kwargs = set([
# URL prepared kwargs
# General Parameters
'user', 'password', 'port', 'host', 'schema', 'fullpath',
# URLBase and NotifyBase args:
'verify', 'format', 'overflow',
# NotifyBase parameters:
'format', 'overflow',
# URLBase parameters:
'verify', 'cto', 'rto',
])
# Valid Schema Entries: