mirror/bazarr
1
0
Fork 0
mirror of https://github.com/morpheus65535/bazarr synced 2024-12-27 01:57:33 +00:00
Code Issues Releases Wiki Activity
bazarr/libs/apprise/plugins/NotifyPushSafer.py
Louis Vézina e6b8b1ad19 Updated Apprise to 0.8.8
2020-09-14 08:24:26 -04:00

829 lines
26 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
# We use io because it allows us to test the open() call
import io
import base64
import requests
from json import loads
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..utils import validate_regex
from ..AppriseLocale import gettext_lazy as _
class PushSaferSound(object):
"""
Defines all of the supported PushSafe sounds
"""
# Silent
SILENT = 0
# Ahem (IM)
AHEM = 1
# Applause (Mail)
APPLAUSE = 2
# Arrow (Reminder)
ARROW = 3
# Baby (SMS)
BABY = 4
# Bell (Alarm)
BELL = 5
# Bicycle (Alarm2)
BICYCLE = 6
# Boing (Alarm3)
BOING = 7
# Buzzer (Alarm4)
BUZZER = 8
# Camera (Alarm5)
CAMERA = 9
# Car Horn (Alarm6)
CAR_HORN = 10
# Cash Register (Alarm7)
CASH_REGISTER = 11
# Chime (Alarm8)
CHIME = 12
# Creaky Door (Alarm9)
CREAKY_DOOR = 13
# Cuckoo Clock (Alarm10)
CUCKOO_CLOCK = 14
# Disconnect (Call)
DISCONNECT = 15
# Dog (Call2)
DOG = 16
# Doorbell (Call3)
DOORBELL = 17
# Fanfare (Call4)
FANFARE = 18
# Gun Shot (Call5)
GUN_SHOT = 19
# Honk (Call6)
HONK = 20
# Jaw Harp (Call7)
JAW_HARP = 21
# Morse (Call8)
MORSE = 22
# Electricity (Call9)
ELECTRICITY = 23
# Radio Tuner (Call10)
RADIO_TURNER = 24
# Sirens
SIRENS = 25
# Military Trumpets
MILITARY_TRUMPETS = 26
# Ufo
UFO = 27
# Whah Whah Whah
LONG_WHAH = 28
# Man Saying Goodbye
GOODBYE = 29
# Man Saying Hello
HELLO = 30
# Man Saying No
NO = 31
# Man Saying Ok
OKAY = 32
# Man Saying Ooohhhweee
OOOHHHWEEE = 33
# Man Saying Warning
WARNING = 34
# Man Saying Welcome
WELCOME = 35
# Man Saying Yeah
YEAH = 36
# Man Saying Yes
YES = 37
# Beep short
BEEP1 = 38
# Weeeee short
WEEE = 39
# Cut in and out short
CUTINOUT = 40
# Finger flicking glas short
FLICK_GLASS = 41
# Wa Wa Waaaa short
SHORT_WHAH = 42
# Laser short
LASER = 43
# Wind Chime short
WIND_CHIME = 44
# Echo short
ECHO = 45
# Zipper short
ZIPPER = 46
# HiHat short
HIHAT = 47
# Beep 2 short
BEEP2 = 48
# Beep 3 short
BEEP3 = 49
# Beep 4 short
BEEP4 = 50
# The Alarm is armed
ALARM_ARMED = 51
# The Alarm is disarmed
ALARM_DISARMED = 52
# The Backup is ready
BACKUP_READY = 53
# The Door is closed
DOOR_CLOSED = 54
# The Door is opend
DOOR_OPENED = 55
# The Window is closed
WINDOW_CLOSED = 56
# The Window is open
WINDOW_OPEN = 57
# The Light is off
LIGHT_ON = 58
# The Light is on
LIGHT_OFF = 59
# The Doorbell rings
DOORBELL_RANG = 60
PUSHSAFER_SOUND_MAP = {
# Device Default,
'silent': PushSaferSound.SILENT,
'ahem': PushSaferSound.AHEM,
'applause': PushSaferSound.APPLAUSE,
'arrow': PushSaferSound.ARROW,
'baby': PushSaferSound.BABY,
'bell': PushSaferSound.BELL,
'bicycle': PushSaferSound.BICYCLE,
'bike': PushSaferSound.BICYCLE,
'boing': PushSaferSound.BOING,
'buzzer': PushSaferSound.BUZZER,
'camera': PushSaferSound.CAMERA,
'carhorn': PushSaferSound.CAR_HORN,
'horn': PushSaferSound.CAR_HORN,
'cashregister': PushSaferSound.CASH_REGISTER,
'chime': PushSaferSound.CHIME,
'creakydoor': PushSaferSound.CREAKY_DOOR,
'cuckooclock': PushSaferSound.CUCKOO_CLOCK,
'cuckoo': PushSaferSound.CUCKOO_CLOCK,
'disconnect': PushSaferSound.DISCONNECT,
'dog': PushSaferSound.DOG,
'doorbell': PushSaferSound.DOORBELL,
'fanfare': PushSaferSound.FANFARE,
'gunshot': PushSaferSound.GUN_SHOT,
'honk': PushSaferSound.HONK,
'jawharp': PushSaferSound.JAW_HARP,
'morse': PushSaferSound.MORSE,
'electric': PushSaferSound.ELECTRICITY,
'radiotuner': PushSaferSound.RADIO_TURNER,
'sirens': PushSaferSound.SIRENS,
'militarytrumpets': PushSaferSound.MILITARY_TRUMPETS,
'military': PushSaferSound.MILITARY_TRUMPETS,
'trumpets': PushSaferSound.MILITARY_TRUMPETS,
'ufo': PushSaferSound.UFO,
'whahwhah': PushSaferSound.LONG_WHAH,
'whah': PushSaferSound.SHORT_WHAH,
'goodye': PushSaferSound.GOODBYE,
'hello': PushSaferSound.HELLO,
'no': PushSaferSound.NO,
'okay': PushSaferSound.OKAY,
'ok': PushSaferSound.OKAY,
'ooohhhweee': PushSaferSound.OOOHHHWEEE,
'warn': PushSaferSound.WARNING,
'warning': PushSaferSound.WARNING,
'welcome': PushSaferSound.WELCOME,
'yeah': PushSaferSound.YEAH,
'yes': PushSaferSound.YES,
'beep': PushSaferSound.BEEP1,
'beep1': PushSaferSound.BEEP1,
'weee': PushSaferSound.WEEE,
'wee': PushSaferSound.WEEE,
'cutinout': PushSaferSound.CUTINOUT,
'flickglass': PushSaferSound.FLICK_GLASS,
'laser': PushSaferSound.LASER,
'windchime': PushSaferSound.WIND_CHIME,
'echo': PushSaferSound.ECHO,
'zipper': PushSaferSound.ZIPPER,
'hihat': PushSaferSound.HIHAT,
'beep2': PushSaferSound.BEEP2,
'beep3': PushSaferSound.BEEP3,
'beep4': PushSaferSound.BEEP4,
'alarmarmed': PushSaferSound.ALARM_ARMED,
'armed': PushSaferSound.ALARM_ARMED,
'alarmdisarmed': PushSaferSound.ALARM_DISARMED,
'disarmed': PushSaferSound.ALARM_DISARMED,
'backupready': PushSaferSound.BACKUP_READY,
'dooropen': PushSaferSound.DOOR_OPENED,
'dopen': PushSaferSound.DOOR_OPENED,
'doorclosed': PushSaferSound.DOOR_CLOSED,
'dclosed': PushSaferSound.DOOR_CLOSED,
'windowopen': PushSaferSound.WINDOW_OPEN,
'wopen': PushSaferSound.WINDOW_OPEN,
'windowclosed': PushSaferSound.WINDOW_CLOSED,
'wclosed': PushSaferSound.WINDOW_CLOSED,
'lighton': PushSaferSound.LIGHT_ON,
'lon': PushSaferSound.LIGHT_ON,
'lightoff': PushSaferSound.LIGHT_OFF,
'loff': PushSaferSound.LIGHT_OFF,
'doorbellrang': PushSaferSound.DOORBELL_RANG,
}
# Priorities
class PushSaferPriority(object):
LOW = -2
MODERATE = -1
NORMAL = 0
HIGH = 1
EMERGENCY = 2
PUSHSAFER_PRIORITIES = (
PushSaferPriority.LOW,
PushSaferPriority.MODERATE,
PushSaferPriority.NORMAL,
PushSaferPriority.HIGH,
PushSaferPriority.EMERGENCY,
)
PUSHSAFER_PRIORITY_MAP = {
# short for 'low'
'low': PushSaferPriority.LOW,
# short for 'medium'
'medium': PushSaferPriority.MODERATE,
# short for 'normal'
'normal': PushSaferPriority.NORMAL,
# short for 'high'
'high': PushSaferPriority.HIGH,
# short for 'emergency'
'emergency': PushSaferPriority.EMERGENCY,
}
# Identify the priority ou want to designate as the fall back
DEFAULT_PRIORITY = "normal"
# Vibrations
class PushSaferVibration(object):
"""
Defines the acceptable vibration settings for notification
"""
# x1
LOW = 1
# x2
NORMAL = 2
# x3
HIGH = 3
# Identify all of the vibrations in one place
PUSHSAFER_VIBRATIONS = (
PushSaferVibration.LOW,
PushSaferVibration.NORMAL,
PushSaferVibration.HIGH,
)
# At this time, the following pictures can be attached to each notification
# at one time. When more are supported, just add their argument below
PICTURE_PARAMETER = (
'p',
'p2',
'p3',
)
# Flag used as a placeholder to sending to all devices
PUSHSAFER_SEND_TO_ALL = 'a'
class NotifyPushSafer(NotifyBase):
"""
A wrapper for PushSafer Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Pushsafer'
# The services URL
service_url = 'https://www.pushsafer.com/'
# The default insecure protocol
protocol = 'psafer'
# The default secure protocol
secure_protocol = 'psafers'
# Number of requests to a allow per second
request_rate_per_sec = 1.2
# The icon ID of 25 looks like a megaphone
default_pushsafer_icon = 25
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushsafer'
# Defines the hostname to post content to; since this service supports
# both insecure and secure methods, we set the {schema} just before we
# post the message upstream.
notify_url = '{schema}://www.pushsafer.com/api'
# Define object templates
templates = (
'{schema}://{privatekey}',
'{schema}://{privatekey}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'privatekey': {
'name': _('Private Key'),
'type': 'string',
'private': True,
'required': True,
},
'target_device': {
'name': _('Target Device'),
'type': 'string',
'map_to': 'targets',
},
'target_email': {
'name': _('Target Email'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'priority': {
'name': _('Priority'),
'type': 'choice:int',
'values': PUSHSAFER_PRIORITIES,
},
'sound': {
'name': _('Sound'),
'type': 'choice:string',
'values': PUSHSAFER_SOUND_MAP,
},
'vibration': {
'name': _('Vibration'),
'type': 'choice:int',
'values': PUSHSAFER_VIBRATIONS,
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, privatekey, targets=None, priority=None, sound=None,
vibration=None, **kwargs):
"""
Initialize PushSafer Object
"""
super(NotifyPushSafer, self).__init__(**kwargs)
#
# Priority
#
try:
# Acquire our priority if we can:
# - We accept both the integer form as well as a string
# representation
self.priority = int(priority)
except TypeError:
# NoneType means use Default; this is an okay exception
self.priority = None
except ValueError:
# Input is a string; attempt to get the lookup from our
# priority mapping
priority = priority.lower().strip()
# This little bit of black magic allows us to match against
# low, lo, l (for low);
# normal, norma, norm, nor, no, n (for normal)
# ... etc
match = next((key for key in PUSHSAFER_PRIORITY_MAP.keys()
if key.startswith(priority)), None) \
if priority else None
# Now test to see if we got a match
if not match:
msg = 'An invalid PushSafer priority ' \
'({}) was specified.'.format(priority)
self.logger.warning(msg)
raise TypeError(msg)
# store our successfully looked up priority
self.priority = PUSHSAFER_PRIORITY_MAP[match]
if self.priority is not None and \
self.priority not in PUSHSAFER_PRIORITY_MAP.values():
msg = 'An invalid PushSafer priority ' \
'({}) was specified.'.format(priority)
self.logger.warning(msg)
raise TypeError(msg)
#
# Sound
#
try:
# Acquire our sound if we can:
# - We accept both the integer form as well as a string
# representation
self.sound = int(sound)
except TypeError:
# NoneType means use Default; this is an okay exception
self.sound = None
except ValueError:
# Input is a string; attempt to get the lookup from our
# sound mapping
sound = sound.lower().strip()
# This little bit of black magic allows us to match against
# against multiple versions of the same string
# ... etc
match = next((key for key in PUSHSAFER_SOUND_MAP.keys()
if key.startswith(sound)), None) \
if sound else None
# Now test to see if we got a match
if not match:
msg = 'An invalid PushSafer sound ' \
'({}) was specified.'.format(sound)
self.logger.warning(msg)
raise TypeError(msg)
# store our successfully looked up sound
self.sound = PUSHSAFER_SOUND_MAP[match]
if self.sound is not None and \
self.sound not in PUSHSAFER_SOUND_MAP.values():
msg = 'An invalid PushSafer sound ' \
'({}) was specified.'.format(sound)
self.logger.warning(msg)
raise TypeError(msg)
#
# Vibration
#
try:
# Use defined integer as is if defined, no further error checking
# is performed
self.vibration = int(vibration)
except TypeError:
# NoneType means use Default; this is an okay exception
self.vibration = None
except ValueError:
msg = 'An invalid PushSafer vibration ' \
'({}) was specified.'.format(vibration)
self.logger.warning(msg)
raise TypeError(msg)
if self.vibration and self.vibration not in PUSHSAFER_VIBRATIONS:
msg = 'An invalid PushSafer vibration ' \
'({}) was specified.'.format(vibration)
self.logger.warning(msg)
raise TypeError(msg)
#
# Private Key (associated with project)
#
self.privatekey = validate_regex(privatekey)
if not self.privatekey:
msg = 'An invalid PushSafer Private Key ' \
'({}) was specified.'.format(privatekey)
self.logger.warning(msg)
raise TypeError(msg)
self.targets = parse_list(targets)
if len(self.targets) == 0:
self.targets = (PUSHSAFER_SEND_TO_ALL, )
return
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
**kwargs):
"""
Perform PushSafer Notification
"""
# error tracking (used for function return)
has_error = False
# Initialize our list of attachments
attachments = []
if attach:
# We need to upload our payload first so that we can source it
# in remaining messages
for attachment in attach:
# prepare payload
if not attachment:
# We could not access the attachment
self.logger.error(
'Could not access attachment {}.'.format(
attachment.url(privacy=True)))
return False
if not attachment.mimetype.startswith('image/'):
# Attachment not supported; continue peacefully
self.logger.debug(
'Ignoring unsupported PushSafer attachment {}.'.format(
attachment.url(privacy=True)))
continue
self.logger.debug(
'Posting PushSafer attachment {}'.format(
attachment.url(privacy=True)))
try:
with io.open(attachment.path, 'rb') as f:
# Output must be in a DataURL format (that's what
# PushSafer calls it):
attachment = (
attachment.name,
'data:{};base64,{}'.format(
attachment.mimetype,
base64.b64encode(f.read())))
except (OSError, IOError) as e:
self.logger.warning(
'An I/O error occurred while reading {}.'.format(
attachment.name if attachment else 'attachment'))
self.logger.debug('I/O Exception: %s' % str(e))
return False
# Save our pre-prepared payload for attachment posting
attachments.append(attachment)
# Create a copy of the targets list
targets = list(self.targets)
while len(targets):
recipient = targets.pop(0)
# prepare payload
payload = {
't': title,
'm': body,
# Our default icon to use
'i': self.default_pushsafer_icon,
# Notification Color
'c': self.color(notify_type),
# Target Recipient
'd': recipient,
}
if self.sound is not None:
# Only apply sound setting if it was specified
payload['s'] = str(self.sound)
if self.vibration is not None:
# Only apply vibration setting
payload['v'] = str(self.vibration)
if not attachments:
okay, response = self._send(payload)
if not okay:
has_error = True
continue
self.logger.info(
'Sent PushSafer notification to "%s".' % (recipient))
else:
# Create a copy of our payload object
_payload = payload.copy()
for idx in range(
0, len(attachments), len(PICTURE_PARAMETER)):
# Send our attachments to our same user (already prepared
# as our payload object)
for c, attachment in enumerate(
attachments[idx:idx + len(PICTURE_PARAMETER)]):
# Get our attachment information
filename, dataurl = attachment
_payload.update({PICTURE_PARAMETER[c]: dataurl})
self.logger.debug(
'Added attachment (%s) to "%s".' % (
filename, recipient))
okay, response = self._send(_payload)
if not okay:
has_error = True
continue
self.logger.info(
'Sent PushSafer attachment (%s) to "%s".' % (
filename, recipient))
# More then the maximum messages shouldn't cause all of
# the text to loop on future iterations
_payload = payload.copy()
_payload['t'] = ''
_payload['m'] = '...'
return not has_error
def _send(self, payload, **kwargs):
"""
Wrapper to the requests (post) object
"""
headers = {
'User-Agent': self.app_id,
}
# Prepare the notification URL to post to
notify_url = self.notify_url.format(
schema='https' if self.secure else 'http'
)
# Store the payload key
payload['k'] = self.privatekey
self.logger.debug('PushSafer POST URL: %s (cert_verify=%r)' % (
notify_url, self.verify_certificate,
))
self.logger.debug('PushSafer Payload: %s' % str(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
# Default response type
response = None
# Initialize our Pushsafer expected responses
_code = None
_str = 'Unknown'
try:
# Open our attachment path if required:
r = requests.post(
notify_url,
data=payload,
headers=headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
try:
response = loads(r.content)
_code = response.get('status')
_str = response.get('success', _str) \
if _code == 1 else response.get('error', _str)
except (AttributeError, TypeError, ValueError):
# ValueError = r.content is Unparsable
# TypeError = r.content is None
# AttributeError = r is None
# Fall back to the existing unparsed value
response = r.content
if r.status_code not in (
requests.codes.ok, requests.codes.no_content):
# We had a problem
status_str = \
NotifyPushSafer.http_response_code_lookup(
r.status_code)
self.logger.warning(
'Failed to deliver payload to PushSafer:'
'{}{}error={}.'.format(
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
return False, response
elif _code != 1:
# It's a bit backwards, but:
# 1 is returned if we succeed
# 0 is returned if we fail
self.logger.warning(
'Failed to deliver payload to PushSafer;'
' error={}.'.format(_str))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
return False, response
# otherwise we were successful
return True, response
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred communicating with PushSafer.')
self.logger.debug('Socket Exception: %s' % str(e))
return False, response
def url(self, privacy=False, *args, **kwargs):
"""
Returns the URL built dynamically based on specified arguments.
"""
# 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
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
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
params['vibration'] = str(self.vibration)
targets = '/'.join([NotifyPushSafer.quote(x) for x in self.targets])
if targets == PUSHSAFER_SEND_TO_ALL:
# keyword is reserved for internal usage only; it's safe to remove
# it from the recipients list
targets = ''
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,
params=NotifyPushSafer.urlencode(params))
@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'] = \
NotifyPushSafer.split_path(results['fullpath'])
# The 'to' makes it easier to use yaml configuration
if 'to' in results['qsd'] and len(results['qsd']['to']):
results['targets'] += \
NotifyPushSafer.parse_list(results['qsd']['to'])
# Setup the token; we store it in Private Key for global
# plugin consistency with naming conventions
results['privatekey'] = NotifyPushSafer.unquote(results['host'])
if 'priority' in results['qsd'] and len(results['qsd']['priority']):
results['priority'] = \
NotifyPushSafer.unquote(results['qsd']['priority'])
if 'sound' in results['qsd'] and len(results['qsd']['sound']):
results['sound'] = \
NotifyPushSafer.unquote(results['qsd']['sound'])
if 'vibration' in results['qsd'] and len(results['qsd']['vibration']):
results['vibration'] = \
NotifyPushSafer.unquote(results['qsd']['vibration'])
return results
Reference in a new issue View git blame Copy permalink