pull/2427/head
v1.4.3-beta.10
morpheus65535
9 months ago
parent
35eabb6a83
commit
aedf2d4d89
@ -0,0 +1,440 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# BSD 2-Clause License
|
||||
#
|
||||
# Apprise - Push Notification Library.
|
||||
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# Redistribution and use in source and binary forms, with or without
|
||||
# modification, are permitted provided that the following conditions are met:
|
||||
#
|
||||
# 1. Redistributions of source code must retain the above copyright notice,
|
||||
# this list of conditions and the following disclaimer.
|
||||
#
|
||||
# 2. Redistributions in binary form must reproduce the above copyright notice,
|
||||
# this list of conditions and the following disclaimer in the documentation
|
||||
# and/or other materials provided with the distribution.
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
||||
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
||||
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
||||
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
|
||||
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
||||
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
||||
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
||||
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
||||
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
#
|
||||
# API:
|
||||
# https://docs.lunasea.app/lunasea/notifications/custom-notifications
|
||||
#
|
||||
import re
|
||||
import requests
|
||||
from json import dumps
|
||||
|
||||
from .NotifyBase import NotifyBase
|
||||
from ..common import NotifyType
|
||||
from ..common import NotifyImageSize
|
||||
from ..utils import parse_list
|
||||
from ..utils import is_hostname
|
||||
from ..utils import is_ipaddr
|
||||
from ..utils import parse_bool
|
||||
from ..AppriseLocale import gettext_lazy as _
|
||||
from ..URLBase import PrivacyMode
|
||||
|
||||
|
||||
class LunaSeaMode:
|
||||
"""
|
||||
Define LunaSea Notification Modes
|
||||
"""
|
||||
# App posts upstream to the developer API on LunaSea's website
|
||||
CLOUD = "cloud"
|
||||
|
||||
# Running a dedicated private ntfy Server
|
||||
PRIVATE = "private"
|
||||
|
||||
|
||||
LUNASEA_MODES = (
|
||||
LunaSeaMode.CLOUD,
|
||||
LunaSeaMode.PRIVATE,
|
||||
)
|
||||
|
||||
|
||||
class NotifyLunaSea(NotifyBase):
|
||||
"""
|
||||
A wrapper for LunaSea Notifications
|
||||
"""
|
||||
|
||||
# The default descriptive name associated with the Notification
|
||||
service_name = 'LunaSea'
|
||||
|
||||
# The services URL
|
||||
service_url = 'https://luasea.app'
|
||||
|
||||
# The default insecure protocol
|
||||
protocol = ('lunasea', 'lsea')
|
||||
|
||||
# The default secure protocol
|
||||
secure_protocol = ('lunaseas', 'lseas')
|
||||
|
||||
# A URL that takes you to the setup/help of the specific protocol
|
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_lunasea'
|
||||
|
||||
# Allows the user to specify the NotifyImageSize object
|
||||
image_size = NotifyImageSize.XY_256
|
||||
|
||||
# LunaSea Notification Details
|
||||
cloud_notify_url = 'https://notify.lunasea.app'
|
||||
notify_user_path = '/v1/custom/user/{}'
|
||||
notify_device_path = '/v1/custom/device/{}'
|
||||
|
||||
# if our hostname matches the following we automatically enforce
|
||||
# cloud mode
|
||||
__auto_cloud_host = re.compile(r'(notify\.)?lunasea\.app', re.IGNORECASE)
|
||||
|
||||
# Define object templates
|
||||
templates = (
|
||||
'{schema}://{targets}',
|
||||
'{schema}://{host}/{targets}',
|
||||
'{schema}://{host}:{port}/{targets}',
|
||||
'{schema}://{user}@{host}/{targets}',
|
||||
'{schema}://{user}@{host}:{port}/{targets}',
|
||||
'{schema}://{user}:{password}@{host}/{targets}',
|
||||
'{schema}://{user}:{password}@{host}:{port}/{targets}',
|
||||
)
|
||||
|
||||
# Define our template tokens
|
||||
template_tokens = dict(NotifyBase.template_tokens, **{
|
||||
'host': {
|
||||
'name': _('Hostname'),
|
||||
'type': 'string',
|
||||
},
|
||||
'port': {
|
||||
'name': _('Port'),
|
||||
'type': 'int',
|
||||
'min': 1,
|
||||
'max': 65535,
|
||||
},
|
||||
'user': {
|
||||
'name': _('Username'),
|
||||
'type': 'string',
|
||||
},
|
||||
'password': {
|
||||
'name': _('Password'),
|
||||
'type': 'string',
|
||||
'private': True,
|
||||
},
|
||||
'token': {
|
||||
'name': _('Token'),
|
||||
'type': 'string',
|
||||
'private': True,
|
||||
},
|
||||
'target_user': {
|
||||
'name': _('Target User'),
|
||||
'type': 'string',
|
||||
'prefix': '@',
|
||||
'map_to': 'targets',
|
||||
},
|
||||
'target_device': {
|
||||
'name': _('Target Device'),
|
||||
'type': 'string',
|
||||
'prefix': '+',
|
||||
'map_to': 'targets',
|
||||
},
|
||||
'targets': {
|
||||
'name': _('Targets'),
|
||||
'type': 'list:string',
|
||||
'required': True,
|
||||
},
|
||||
})
|
||||
|
||||
# Define our template arguments
|
||||
template_args = dict(NotifyBase.template_args, **{
|
||||
'to': {
|
||||
'alias_of': 'targets',
|
||||
},
|
||||
'image': {
|
||||
'name': _('Include Image'),
|
||||
'type': 'bool',
|
||||
'default': False,
|
||||
'map_to': 'include_image',
|
||||
},
|
||||
'mode': {
|
||||
'name': _('Mode'),
|
||||
'type': 'choice:string',
|
||||
'values': LUNASEA_MODES,
|
||||
'default': LunaSeaMode.PRIVATE,
|
||||
},
|
||||
})
|
||||
|
||||
def __init__(self, targets=None, mode=None, token=None,
|
||||
include_image=False, **kwargs):
|
||||
"""
|
||||
Initialize LunaSea Object
|
||||
"""
|
||||
super().__init__(**kwargs)
|
||||
|
||||
# Show image associated with notification
|
||||
self.include_image = \
|
||||
self.template_args['image']['default'] \
|
||||
if include_image is None else include_image
|
||||
|
||||
# Prepare our mode
|
||||
self.mode = mode.strip().lower() \
|
||||
if isinstance(mode, str) \
|
||||
else self.template_args['mode']['default']
|
||||
|
||||
if self.mode not in LUNASEA_MODES:
|
||||
msg = 'An invalid LunaSea mode ({}) was specified.'.format(mode)
|
||||
self.logger.warning(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
self.targets = []
|
||||
for target in parse_list(targets):
|
||||
if len(target) < 4:
|
||||
self.logger.warning(
|
||||
'A specified target ({}) is invalid and will be '
|
||||
'ignored'.format(target))
|
||||
continue
|
||||
|
||||
if target[0] == '+':
|
||||
# Device
|
||||
self.targets.append(('+', target[1:]))
|
||||
|
||||
elif target[0] == '@':
|
||||
# User
|
||||
self.targets.append(('@', target[1:]))
|
||||
|
||||
else:
|
||||
# User
|
||||
self.targets.append(('@', target))
|
||||
|
||||
return
|
||||
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
|
||||
"""
|
||||
Perform LunaSea Notification
|
||||
"""
|
||||
|
||||
# error tracking (used for function return)
|
||||
has_error = False
|
||||
|
||||
if not len(self.targets):
|
||||
# We have nothing to notify; we're done
|
||||
self.logger.warning('There are no LunaSea targets to notify')
|
||||
return False
|
||||
|
||||
# Prepare our headers
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json',
|
||||
}
|
||||
|
||||
# prepare payload
|
||||
payload = {
|
||||
'title': title if title else self.app_desc,
|
||||
'body': body,
|
||||
}
|
||||
|
||||
# Acquire image_url
|
||||
image_url = None if not self.include_image \
|
||||
else self.image_url(notify_type)
|
||||
|
||||
if image_url:
|
||||
payload['image'] = image_url
|
||||
|
||||
# Prepare our Authentication (if defined)
|
||||
if self.user and self.password:
|
||||
auth = (self.user, self.password)
|
||||
|
||||
else:
|
||||
# No Auth
|
||||
auth = None
|
||||
|
||||
if self.mode == LunaSeaMode.CLOUD:
|
||||
# Cloud Service
|
||||
notify_url = self.cloud_notify_url
|
||||
|
||||
else:
|
||||
# Local Hosting
|
||||
schema = 'https' if self.secure else 'http'
|
||||
|
||||
notify_url = '%s://%s' % (schema, self.host)
|
||||
if isinstance(self.port, int):
|
||||
notify_url += ':%d' % self.port
|
||||
|
||||
# Create a copy of the targets list
|
||||
targets = list(self.targets)
|
||||
while len(targets):
|
||||
target = targets.pop(0)
|
||||
|
||||
if target[0] == '+':
|
||||
url = notify_url + self.notify_device_path.format(target[1])
|
||||
|
||||
else:
|
||||
url = notify_url + self.notify_user_path.format(target[1])
|
||||
|
||||
self.logger.debug('LunaSea POST URL: %s (cert_verify=%r)' % (
|
||||
url, self.verify_certificate,
|
||||
))
|
||||
self.logger.debug('LunaSea Payload: %s' % str(payload))
|
||||
|
||||
# Always call throttle before any remote server i/o is made
|
||||
self.throttle()
|
||||
|
||||
try:
|
||||
r = requests.post(
|
||||
url,
|
||||
data=dumps(payload),
|
||||
headers=headers,
|
||||
auth=auth,
|
||||
verify=self.verify_certificate,
|
||||
timeout=self.request_timeout,
|
||||
)
|
||||
|
||||
if r.status_code not in (
|
||||
requests.codes.ok, requests.codes.no_content):
|
||||
# We had a problem
|
||||
status_str = \
|
||||
NotifyLunaSea.http_response_code_lookup(r.status_code)
|
||||
|
||||
self.logger.warning(
|
||||
'Failed to deliver payload to LunaSea:'
|
||||
'{}{}error={}.'.format(
|
||||
status_str,
|
||||
', ' if status_str else '',
|
||||
r.status_code))
|
||||
|
||||
self.logger.debug(
|
||||
'Response Details:\r\n{}'.format(r.content))
|
||||
|
||||
has_error = True
|
||||
|
||||
# otherwise we were successful
|
||||
continue
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A Connection error occurred communicating with LunaSea.')
|
||||
self.logger.debug('Socket Exception: %s' % str(e))
|
||||
|
||||
has_error = True
|
||||
|
||||
return not has_error
|
||||
|
||||
def url(self, privacy=False, *args, **kwargs):
|
||||
"""
|
||||
Returns the URL built dynamically based on specified arguments.
|
||||
"""
|
||||
|
||||
params = {
|
||||
'mode': self.mode,
|
||||
'image': 'yes' if self.include_image else 'no',
|
||||
}
|
||||
|
||||
# Our URL parameters
|
||||
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
|
||||
|
||||
auth = ''
|
||||
if self.user and self.password:
|
||||
auth = '{user}:{password}@'.format(
|
||||
user=NotifyLunaSea.quote(self.user, safe=''),
|
||||
password=self.pprint(
|
||||
self.password, privacy, mode=PrivacyMode.Secret,
|
||||
safe=''),
|
||||
)
|
||||
elif self.user:
|
||||
auth = '{user}@'.format(
|
||||
user=NotifyLunaSea.quote(self.user, safe=''),
|
||||
)
|
||||
|
||||
if self.mode == LunaSeaMode.PRIVATE:
|
||||
default_port = 443 if self.secure else 80
|
||||
return '{schema}://{auth}{host}{port}/{targets}?{params}'.format(
|
||||
schema=self.secure_protocol[0]
|
||||
if self.secure else self.protocol[0],
|
||||
auth=auth,
|
||||
host=self.host,
|
||||
port='' if self.port is None or self.port == default_port
|
||||
else ':{}'.format(self.port),
|
||||
targets='/'.join(
|
||||
[NotifyLunaSea.quote(x[0] + x[1], safe='@+')
|
||||
for x in self.targets]),
|
||||
params=NotifyLunaSea.urlencode(params)
|
||||
)
|
||||
|
||||
else: # Cloud mode
|
||||
return '{schema}://{auth}{targets}?{params}'.format(
|
||||
schema=self.protocol[0],
|
||||
auth=auth,
|
||||
targets='/'.join(
|
||||
[NotifyLunaSea.quote(x[0] + x[1], safe='@+')
|
||||
for x in self.targets]),
|
||||
params=NotifyLunaSea.urlencode(params)
|
||||
)
|
||||
|
||||
def __len__(self):
|
||||
"""
|
||||
Returns the number of targets associated with this notification
|
||||
"""
|
||||
# always return 1
|
||||
return 1 if not self.targets else len(self.targets)
|
||||
|
||||
@staticmethod
|
||||
def parse_url(url):
|
||||
"""
|
||||
Parses the URL and returns enough arguments that can allow
|
||||
us to re-instantiate this object.
|
||||
|
||||
"""
|
||||
results = NotifyBase.parse_url(url, verify_host=False)
|
||||
if not results:
|
||||
# We're done early as we couldn't load the results
|
||||
return results
|
||||
|
||||
# Fetch our targets
|
||||
results['targets'] = NotifyLunaSea.split_path(results['fullpath'])
|
||||
|
||||
# Boolean to include an image or not
|
||||
results['include_image'] = parse_bool(results['qsd'].get(
|
||||
'image', NotifyLunaSea.template_args['image']['default']))
|
||||
|
||||
# The 'to' makes it easier to use yaml configuration
|
||||
if 'to' in results['qsd'] and len(results['qsd']['to']):
|
||||
results['targets'] += \
|
||||
NotifyLunaSea.parse_list(results['qsd']['to'])
|
||||
|
||||
# Mode override
|
||||
if 'mode' in results['qsd'] and results['qsd']['mode']:
|
||||
results['mode'] = NotifyLunaSea.unquote(
|
||||
results['qsd']['mode'].strip().lower())
|
||||
|
||||
else:
|
||||
# We can try to detect the mode based on the validity of the
|
||||
# hostname.
|
||||
#
|
||||
# This isn't a surfire way to do things though; it's best to
|
||||
# specify the mode= flag
|
||||
results['mode'] = LunaSeaMode.PRIVATE \
|
||||
if ((is_hostname(results['host'])
|
||||
or is_ipaddr(results['host'])) and results['targets']) \
|
||||
else LunaSeaMode.CLOUD
|
||||
|
||||
if results['mode'] == LunaSeaMode.CLOUD:
|
||||
# Store first entry as it can be a topic too in this case
|
||||
# But only if we also rule it out not being the words
|
||||
# lunasea.app itself, something that starts wiht an non-alpha
|
||||
# numeric character:
|
||||
if not NotifyLunaSea.__auto_cloud_host.search(results['host']):
|
||||
# Add it to the front of the list for consistency
|
||||
results['targets'].insert(0, results['host'])
|
||||
|
||||
elif results['mode'] == LunaSeaMode.PRIVATE and \
|
||||
not (is_hostname(results['host'] or
|
||||
is_ipaddr(results['host']))):
|
||||
# Invalid Host for LunaSeaMode.PRIVATE
|
||||
return None
|
||||
|
||||
return results
|
||||
Loading…
Reference in new issue