mirror of
https://github.com/morpheus65535/bazarr.git
synced 2024-09-20 15:35:58 +08:00
1003 lines
36 KiB
Python
1003 lines
36 KiB
Python
# -*- coding: utf-8 -*-
|
|
#
|
|
# Copyright (C) 2020 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.
|
|
|
|
# For LaMetric to work, you need to first setup a custom application on their
|
|
# website. it can be done as follows:
|
|
|
|
# Cloud Mode:
|
|
# - Sign Up and login to the developer webpage https://developer.lametric.com
|
|
#
|
|
# - Create a **Indicator App** if you haven't already done so from here:
|
|
# https://developer.lametric.com/applications/sources
|
|
#
|
|
# There is a great official tutorial on how to do this here:
|
|
# https://lametric-documentation.readthedocs.io/en/latest/\
|
|
# guides/first-steps/first-lametric-indicator-app.html
|
|
#
|
|
# - Make sure to set the **Communication Type** to **PUSH**.
|
|
#
|
|
# - You will be able to **Publish** your app once you've finished setting it
|
|
# up. This will allow it to be accessible from the internet using the
|
|
# `cloud` mode of this Apprise Plugin. The **Publish** button shows up
|
|
# from within the settings of your Lametric App upon clicking on the
|
|
# **Draft Vx** folder (where `x` is the version - usually a 1)
|
|
#
|
|
# When you've completed, the site would have provided you a **PUSH URL** that
|
|
# looks like this:
|
|
# https://developer.lametric.com/api/v1/dev/widget/update/\
|
|
# com.lametric.{app_id}/{app_ver}
|
|
#
|
|
# You will need to record the `{app_id}` and `{app_ver}` to use the `cloud`
|
|
# mode.
|
|
#
|
|
# The same page should also provide you with an **Access Token**. It's
|
|
# approximately 86 characters with two equal (`=`) characters at the end of it.
|
|
# This becomes your `{app_token}`. Here is an example of what one might
|
|
# look like:
|
|
# K2MxWI0NzU0ZmI2NjJlZYTgViMDgDRiN8YjlmZjRmNTc4NDVhJzk0RiNjNh0EyKWW==`
|
|
#
|
|
# The syntax for the cloud mode is:
|
|
# * `lametric://{app_token}@{app_id}/{app_ver}?mode=cloud`
|
|
|
|
# Device Mode:
|
|
# - Sign Up and login to the developer webpage https://developer.lametric.com
|
|
# - Locate your Device API Key; you can find it here:
|
|
# https://developer.lametric.com/user/devices
|
|
# - From here you can get your your API Key for the device you plan to notify.
|
|
# - Your devices IP Address can be found in LaMetric Time app at:
|
|
# Settings -> Wi-Fi -> IP Address
|
|
#
|
|
# The syntax for the device mode is:
|
|
# * `lametric://{apikey}@{host}`
|
|
|
|
# A great source for API examples (Device Mode):
|
|
# - https://lametric-documentation.readthedocs.io/en/latest/reference-docs\
|
|
# /device-notifications.html
|
|
#
|
|
# A great source for API examples (Cloud Mode):
|
|
# - https://lametric-documentation.readthedocs.io/en/latest/reference-docs\
|
|
# /lametric-cloud-reference.html
|
|
|
|
# A great source for the icon reference:
|
|
# - https://developer.lametric.com/icons
|
|
|
|
|
|
import re
|
|
import six
|
|
import requests
|
|
from json import dumps
|
|
from .NotifyBase import NotifyBase
|
|
from ..common import NotifyType
|
|
from ..utils import validate_regex
|
|
from ..AppriseLocale import gettext_lazy as _
|
|
from ..utils import is_hostname
|
|
from ..utils import is_ipaddr
|
|
|
|
# A URL Parser to detect App ID
|
|
LAMETRIC_APP_ID_DETECTOR_RE = re.compile(
|
|
r'(com\.lametric\.)?(?P<app_id>[0-9a-z.-]{1,64})'
|
|
r'(/(?P<app_ver>[1-9][0-9]*))?', re.I)
|
|
|
|
# Tokens are huge
|
|
LAMETRIC_IS_APP_TOKEN = re.compile(r'^[a-z0-9]{80,}==$', re.I)
|
|
|
|
|
|
class LametricMode(object):
|
|
"""
|
|
Define Lametric Notification Modes
|
|
"""
|
|
# App posts upstream to the developer API on Lametric's website
|
|
CLOUD = "cloud"
|
|
|
|
# Device mode posts directly to the device that you identify
|
|
DEVICE = "device"
|
|
|
|
|
|
LAMETRIC_MODES = (
|
|
LametricMode.CLOUD,
|
|
LametricMode.DEVICE,
|
|
)
|
|
|
|
|
|
class LametricPriority(object):
|
|
"""
|
|
Priority of the message
|
|
"""
|
|
|
|
# info: this priority means that notification will be displayed on the
|
|
# same “level” as all other notifications on the device that come
|
|
# from apps (for example facebook app). This notification will not
|
|
# be shown when screensaver is active. By default message is sent
|
|
# with "info" priority. This level of notification should be used
|
|
# for notifications like news, weather, temperature, etc.
|
|
INFO = 'info'
|
|
|
|
# warning: notifications with this priority will interrupt ones sent with
|
|
# lower priority (“info”). Should be used to notify the user
|
|
# about something important but not critical. For example,
|
|
# events like “someone is coming home” should use this priority
|
|
# when sending notifications from smart home.
|
|
WARNING = 'warning'
|
|
|
|
# critical: the most important notifications. Interrupts notification
|
|
# with priority info or warning and is displayed even if
|
|
# screensaver is active. Use with care as these notifications
|
|
# can pop in the middle of the night. Must be used only for
|
|
# really important notifications like notifications from smoke
|
|
# detectors, water leak sensors, etc. Use it for events that
|
|
# require human interaction immediately.
|
|
CRITICAL = 'critical'
|
|
|
|
|
|
LAMETRIC_PRIORITIES = (
|
|
LametricPriority.INFO,
|
|
LametricPriority.WARNING,
|
|
LametricPriority.CRITICAL,
|
|
)
|
|
|
|
|
|
class LametricIconType(object):
|
|
"""
|
|
Represents the nature of notification.
|
|
"""
|
|
|
|
# info - "i" icon will be displayed prior to the notification. Means that
|
|
# notification contains information, no need to take actions on it.
|
|
INFO = 'info'
|
|
|
|
# alert: "!!!" icon will be displayed prior to the notification. Use it
|
|
# when you want the user to pay attention to that notification as
|
|
# it indicates that something bad happened and user must take
|
|
# immediate action.
|
|
ALERT = 'alert'
|
|
|
|
# none: no notification icon will be shown.
|
|
NONE = 'none'
|
|
|
|
|
|
LAMETRIC_ICON_TYPES = (
|
|
LametricIconType.INFO,
|
|
LametricIconType.ALERT,
|
|
LametricIconType.NONE,
|
|
)
|
|
|
|
|
|
class LametricSoundCategory(object):
|
|
"""
|
|
Define Sound Categories
|
|
"""
|
|
NOTIFICATIONS = "notifications"
|
|
ALARMS = "alarms"
|
|
|
|
|
|
class LametricSound(object):
|
|
"""
|
|
There are 2 categories of sounds, to make things simple we just lump them
|
|
all togther in one class object.
|
|
|
|
Syntax is (Category, (AlarmID, Alias1, Alias2, ...))
|
|
"""
|
|
|
|
# Alarm Category Sounds
|
|
ALARM01 = (LametricSoundCategory.ALARMS, ('alarm1', 'a1', 'a01'))
|
|
ALARM02 = (LametricSoundCategory.ALARMS, ('alarm2', 'a2', 'a02'))
|
|
ALARM03 = (LametricSoundCategory.ALARMS, ('alarm3', 'a3', 'a03'))
|
|
ALARM04 = (LametricSoundCategory.ALARMS, ('alarm4', 'a4', 'a04'))
|
|
ALARM05 = (LametricSoundCategory.ALARMS, ('alarm5', 'a5', 'a05'))
|
|
ALARM06 = (LametricSoundCategory.ALARMS, ('alarm6', 'a6', 'a06'))
|
|
ALARM07 = (LametricSoundCategory.ALARMS, ('alarm7', 'a7', 'a07'))
|
|
ALARM08 = (LametricSoundCategory.ALARMS, ('alarm8', 'a8', 'a08'))
|
|
ALARM09 = (LametricSoundCategory.ALARMS, ('alarm9', 'a9', 'a09'))
|
|
ALARM10 = (LametricSoundCategory.ALARMS, ('alarm10', 'a10'))
|
|
ALARM11 = (LametricSoundCategory.ALARMS, ('alarm11', 'a11'))
|
|
ALARM12 = (LametricSoundCategory.ALARMS, ('alarm12', 'a12'))
|
|
ALARM13 = (LametricSoundCategory.ALARMS, ('alarm13', 'a13'))
|
|
|
|
# Notification Category Sounds
|
|
BICYCLE = (LametricSoundCategory.NOTIFICATIONS, ('bicycle', 'bike'))
|
|
CAR = (LametricSoundCategory.NOTIFICATIONS, ('car', ))
|
|
CASH = (LametricSoundCategory.NOTIFICATIONS, ('cash', ))
|
|
CAT = (LametricSoundCategory.NOTIFICATIONS, ('cat', ))
|
|
DOG01 = (LametricSoundCategory.NOTIFICATIONS, ('dog', 'dog1', 'dog01'))
|
|
DOG02 = (LametricSoundCategory.NOTIFICATIONS, ('dog2', 'dog02'))
|
|
ENERGY = (LametricSoundCategory.NOTIFICATIONS, ('energy', ))
|
|
KNOCK = (LametricSoundCategory.NOTIFICATIONS, ('knock-knock', 'knock'))
|
|
EMAIL = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'letter_email', 'letter', 'email'))
|
|
LOSE01 = (LametricSoundCategory.NOTIFICATIONS, ('lose1', 'lose01', 'lose'))
|
|
LOSE02 = (LametricSoundCategory.NOTIFICATIONS, ('lose2', 'lose02'))
|
|
NEGATIVE01 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'negative1', 'negative01', 'neg01', 'neg1', '-'))
|
|
NEGATIVE02 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'negative2', 'negative02', 'neg02', 'neg2', '--'))
|
|
NEGATIVE03 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'negative3', 'negative03', 'neg03', 'neg3', '---'))
|
|
NEGATIVE04 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'negative4', 'negative04', 'neg04', 'neg4', '----'))
|
|
NEGATIVE05 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'negative5', 'negative05', 'neg05', 'neg5', '-----'))
|
|
NOTIFICATION01 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'notification', 'notification1', 'notification01', 'not01', 'not1'))
|
|
NOTIFICATION02 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'notification2', 'notification02', 'not02', 'not2'))
|
|
NOTIFICATION03 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'notification3', 'notification03', 'not03', 'not3'))
|
|
NOTIFICATION04 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'notification4', 'notification04', 'not04', 'not4'))
|
|
OPEN_DOOR = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'open_door', 'open', 'door'))
|
|
POSITIVE01 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'positive1', 'positive01', 'pos01', 'p1', '+'))
|
|
POSITIVE02 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'positive2', 'positive02', 'pos02', 'p2', '++'))
|
|
POSITIVE03 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'positive3', 'positive03', 'pos03', 'p3', '+++'))
|
|
POSITIVE04 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'positive4', 'positive04', 'pos04', 'p4', '++++'))
|
|
POSITIVE05 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'positive5', 'positive05', 'pos05', 'p5', '+++++'))
|
|
POSITIVE06 = (LametricSoundCategory.NOTIFICATIONS, (
|
|
'positive6', 'positive06', 'pos06', 'p6', '++++++'))
|
|
STATISTIC = (LametricSoundCategory.NOTIFICATIONS, ('statistic', 'stat'))
|
|
THUNDER = (LametricSoundCategory.NOTIFICATIONS, ('thunder'))
|
|
WATER01 = (LametricSoundCategory.NOTIFICATIONS, ('water1', 'water01'))
|
|
WATER02 = (LametricSoundCategory.NOTIFICATIONS, ('water2', 'water02'))
|
|
WIN01 = (LametricSoundCategory.NOTIFICATIONS, ('win', 'win01', 'win1'))
|
|
WIN02 = (LametricSoundCategory.NOTIFICATIONS, ('win2', 'win02'))
|
|
WIND = (LametricSoundCategory.NOTIFICATIONS, ('wind', ))
|
|
WIND_SHORT = (LametricSoundCategory.NOTIFICATIONS, ('wind_short', ))
|
|
|
|
|
|
# A listing of all the sounds; the order DOES matter, content is read from
|
|
# top down and then right to left (over aliases). Longer similar sounding
|
|
# elements should be placed higher in the list over others. for example
|
|
# ALARM10 should come before ALARM01 (because ALARM01 can match on 'alarm1'
|
|
# which is very close to 'alarm10'
|
|
LAMETRIC_SOUNDS = (
|
|
# Alarm Category Entries
|
|
LametricSound.ALARM13, LametricSound.ALARM12, LametricSound.ALARM11,
|
|
LametricSound.ALARM10, LametricSound.ALARM09, LametricSound.ALARM08,
|
|
LametricSound.ALARM07, LametricSound.ALARM06, LametricSound.ALARM05,
|
|
LametricSound.ALARM04, LametricSound.ALARM03, LametricSound.ALARM02,
|
|
LametricSound.ALARM01,
|
|
|
|
# Notification Category Entries
|
|
LametricSound.BICYCLE, LametricSound.CAR, LametricSound.CASH,
|
|
LametricSound.CAT, LametricSound.DOG02, LametricSound.DOG01,
|
|
LametricSound.ENERGY, LametricSound.KNOCK, LametricSound.EMAIL,
|
|
LametricSound.LOSE02, LametricSound.LOSE01, LametricSound.NEGATIVE01,
|
|
LametricSound.NEGATIVE02, LametricSound.NEGATIVE03,
|
|
LametricSound.NEGATIVE04, LametricSound.NEGATIVE05,
|
|
LametricSound.NOTIFICATION04, LametricSound.NOTIFICATION03,
|
|
LametricSound.NOTIFICATION02, LametricSound.NOTIFICATION01,
|
|
LametricSound.OPEN_DOOR, LametricSound.POSITIVE01,
|
|
LametricSound.POSITIVE02, LametricSound.POSITIVE03,
|
|
LametricSound.POSITIVE04, LametricSound.POSITIVE05,
|
|
LametricSound.POSITIVE01, LametricSound.STATISTIC, LametricSound.THUNDER,
|
|
LametricSound.WATER02, LametricSound.WATER01, LametricSound.WIND,
|
|
LametricSound.WIND_SHORT, LametricSound.WIN01, LametricSound.WIN02,
|
|
)
|
|
|
|
|
|
class NotifyLametric(NotifyBase):
|
|
"""
|
|
A wrapper for LaMetric Notifications
|
|
"""
|
|
|
|
# The default descriptive name associated with the Notification
|
|
service_name = 'LaMetric'
|
|
|
|
# The services URL
|
|
service_url = 'https://lametric.com'
|
|
|
|
# The default protocol
|
|
protocol = 'lametric'
|
|
|
|
# The default secure protocol
|
|
secure_protocol = 'lametrics'
|
|
|
|
# Allow 300 requests per minute.
|
|
# 60/300 = 0.2
|
|
request_rate_per_sec = 0.20
|
|
|
|
# A URL that takes you to the setup/help of the specific protocol
|
|
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_lametric'
|
|
|
|
# Lametric does have titles when creating a message
|
|
title_maxlen = 0
|
|
|
|
# URL used for notifying Lametric App's created in the Dev Portal
|
|
cloud_notify_url = 'https://developer.lametric.com/api/v1' \
|
|
'/dev/widget/update/com.lametric.{app_id}/{app_ver}'
|
|
|
|
# URL used for local notifications directly to the device
|
|
device_notify_url = '{schema}://{host}{port}/api/v2/device/notifications'
|
|
|
|
# The Device User ID
|
|
default_device_user = 'dev'
|
|
|
|
# Track all icon mappings back to Apprise Icon NotifyType's
|
|
# See: https://developer.lametric.com/icons
|
|
# Icon ID looks like <prefix>XXX, where <prefix> is:
|
|
# - "i" (for static icon)
|
|
# - "a" (for animation)
|
|
# - XXX - is the number of the icon and can be found at:
|
|
# https://developer.lametric.com/icons
|
|
lametric_icon_id_mapping = {
|
|
# 620/Info
|
|
NotifyType.INFO: 'i620',
|
|
# 9182/info_good
|
|
NotifyType.SUCCESS: 'i9182',
|
|
# 9183/info_caution
|
|
NotifyType.WARNING: 'i9183',
|
|
# 9184/info_error
|
|
NotifyType.FAILURE: 'i9184',
|
|
}
|
|
|
|
# Define object templates
|
|
templates = (
|
|
# Cloud (App) Mode
|
|
'{schema}://{app_token}@{app_id}',
|
|
'{schema}://{app_token}@{app_id}/{app_ver}',
|
|
|
|
# Device Mode
|
|
'{schema}://{apikey}@{host}',
|
|
'{schema}://{apikey}@{host}:{port}',
|
|
'{schema}://{user}:{apikey}@{host}:{port}',
|
|
)
|
|
|
|
# Define our template tokens
|
|
template_tokens = dict(NotifyBase.template_tokens, **{
|
|
# Used for Local Device mode
|
|
'apikey': {
|
|
'name': _('Device API Key'),
|
|
'type': 'string',
|
|
'private': True,
|
|
},
|
|
# Used for Cloud mode
|
|
'app_id': {
|
|
'name': _('App ID'),
|
|
'type': 'string',
|
|
'private': True,
|
|
},
|
|
# Used for Cloud mode
|
|
'app_ver': {
|
|
'name': _('App Version'),
|
|
'type': 'string',
|
|
'regex': (r'^[1-9][0-9]*$', ''),
|
|
'default': '1',
|
|
},
|
|
# Used for Cloud mode
|
|
'app_token': {
|
|
'name': _('App Access Token'),
|
|
'type': 'string',
|
|
'regex': (r'^[A-Z0-9]{80,}==$', 'i'),
|
|
},
|
|
'host': {
|
|
'name': _('Hostname'),
|
|
'type': 'string',
|
|
'required': True,
|
|
},
|
|
'port': {
|
|
'name': _('Port'),
|
|
'type': 'int',
|
|
'min': 1,
|
|
'max': 65535,
|
|
'default': 8080,
|
|
},
|
|
'user': {
|
|
'name': _('Username'),
|
|
'type': 'string',
|
|
},
|
|
})
|
|
|
|
# Define our template arguments
|
|
template_args = dict(NotifyBase.template_args, **{
|
|
'apikey': {
|
|
'alias_of': 'apikey',
|
|
},
|
|
'app_id': {
|
|
'alias_of': 'app_id',
|
|
},
|
|
'app_ver': {
|
|
'alias_of': 'app_ver',
|
|
},
|
|
'app_token': {
|
|
'alias_of': 'app_token',
|
|
},
|
|
'priority': {
|
|
'name': _('Priority'),
|
|
'type': 'choice:string',
|
|
'values': LAMETRIC_PRIORITIES,
|
|
'default': LametricPriority.INFO,
|
|
},
|
|
'icon': {
|
|
'name': _('Custom Icon'),
|
|
'type': 'string',
|
|
},
|
|
'icon_type': {
|
|
'name': _('Icon Type'),
|
|
'type': 'choice:string',
|
|
'values': LAMETRIC_ICON_TYPES,
|
|
'default': LametricIconType.NONE,
|
|
},
|
|
'mode': {
|
|
'name': _('Mode'),
|
|
'type': 'choice:string',
|
|
'values': LAMETRIC_MODES,
|
|
'default': LametricMode.DEVICE,
|
|
},
|
|
'sound': {
|
|
'name': _('Sound'),
|
|
'type': 'string',
|
|
},
|
|
# Lifetime is in seconds
|
|
'cycles': {
|
|
'name': _('Cycles'),
|
|
'type': 'int',
|
|
'min': 0,
|
|
'default': 1,
|
|
},
|
|
})
|
|
|
|
def __init__(self, apikey=None, app_token=None, app_id=None,
|
|
app_ver=None, priority=None, icon=None, icon_type=None,
|
|
sound=None, mode=None, cycles=None, **kwargs):
|
|
"""
|
|
Initialize LaMetric Object
|
|
"""
|
|
super(NotifyLametric, self).__init__(**kwargs)
|
|
|
|
self.mode = mode.strip().lower() \
|
|
if isinstance(mode, six.string_types) \
|
|
else self.template_args['mode']['default']
|
|
|
|
# Default Cloud Argument
|
|
self.lametric_app_id = None
|
|
self.lametric_app_ver = None
|
|
self.lametric_app_access_token = None
|
|
|
|
# Default Device/Cloud Argument
|
|
self.lametric_apikey = None
|
|
|
|
if self.mode not in LAMETRIC_MODES:
|
|
msg = 'An invalid LaMetric Mode ({}) was specified.'.format(mode)
|
|
self.logger.warning(msg)
|
|
raise TypeError(msg)
|
|
|
|
if self.mode == LametricMode.CLOUD:
|
|
try:
|
|
results = LAMETRIC_APP_ID_DETECTOR_RE.match(app_id)
|
|
except TypeError:
|
|
msg = 'An invalid LaMetric Application ID ' \
|
|
'({}) was specified.'.format(app_id)
|
|
self.logger.warning(msg)
|
|
raise TypeError(msg)
|
|
|
|
# Detect our Access Token
|
|
self.lametric_app_access_token = validate_regex(
|
|
app_token,
|
|
*self.template_tokens['app_token']['regex'])
|
|
if not self.lametric_app_access_token:
|
|
msg = 'An invalid LaMetric Application Access Token ' \
|
|
'({}) was specified.'.format(app_token)
|
|
self.logger.warning(msg)
|
|
raise TypeError(msg)
|
|
|
|
# If app_ver is specified, it over-rides all
|
|
if app_ver:
|
|
self.lametric_app_ver = validate_regex(
|
|
app_ver, *self.template_tokens['app_ver']['regex'])
|
|
if not self.lametric_app_ver:
|
|
msg = 'An invalid LaMetric Application Version ' \
|
|
'({}) was specified.'.format(app_ver)
|
|
self.logger.warning(msg)
|
|
raise TypeError(msg)
|
|
|
|
else:
|
|
# If app_ver wasn't specified, we parse it from the
|
|
# Application ID
|
|
self.lametric_app_ver = results.group('app_ver') \
|
|
if results.group('app_ver') else \
|
|
self.template_tokens['app_ver']['default']
|
|
|
|
# Store our Application ID
|
|
self.lametric_app_id = results.group('app_id')
|
|
|
|
if self.mode == LametricMode.DEVICE:
|
|
self.lametric_apikey = validate_regex(apikey)
|
|
if not self.lametric_apikey:
|
|
msg = 'An invalid LaMetric Device API Key ' \
|
|
'({}) was specified.'.format(apikey)
|
|
self.logger.warning(msg)
|
|
raise TypeError(msg)
|
|
|
|
if priority not in LAMETRIC_PRIORITIES:
|
|
self.priority = self.template_args['priority']['default']
|
|
|
|
else:
|
|
self.priority = priority
|
|
|
|
# assign our icon (if it was defined); we also eliminate
|
|
# any hashtag (#) entries that might be present
|
|
self.icon = re.search(r'[#\s]*(?P<value>.+?)\s*$', icon) \
|
|
.group('value') if isinstance(icon, six.string_types) else None
|
|
|
|
if icon_type not in LAMETRIC_ICON_TYPES:
|
|
self.icon_type = self.template_args['icon_type']['default']
|
|
|
|
else:
|
|
self.icon_type = icon_type
|
|
|
|
# The number of times the message should be displayed
|
|
self.cycles = self.template_args['cycles']['default'] \
|
|
if not (isinstance(cycles, int) and
|
|
cycles > self.template_args['cycles']['min']) else cycles
|
|
|
|
self.sound = None
|
|
if isinstance(sound, six.string_types):
|
|
# If sound is set, get it's match
|
|
self.sound = self.sound_lookup(sound.strip().lower())
|
|
if self.sound is None:
|
|
self.logger.warning(
|
|
'An invalid LaMetric sound ({}) was specified.'.format(
|
|
sound))
|
|
return
|
|
|
|
@staticmethod
|
|
def sound_lookup(lookup):
|
|
"""
|
|
A simple match function that takes string and returns the
|
|
LametricSound object it was found in.
|
|
|
|
"""
|
|
|
|
for x in LAMETRIC_SOUNDS:
|
|
match = next((f for f in x[1] if f.startswith(lookup)), None)
|
|
if match:
|
|
# We're done
|
|
return x
|
|
|
|
# No match was found
|
|
return None
|
|
|
|
def _cloud_notification_payload(self, body, notify_type, headers):
|
|
"""
|
|
Return URL and payload for cloud directed requests
|
|
"""
|
|
|
|
# Update header entries
|
|
headers.update({
|
|
'X-Access-Token': self.lametric_apikey,
|
|
})
|
|
|
|
if self.sound:
|
|
self.logger.warning(
|
|
'LaMetric sound setting is unavailable in Cloud mode')
|
|
|
|
if self.priority != self.template_args['priority']['default']:
|
|
self.logger.warning(
|
|
'LaMetric priority setting is unavailable in Cloud mode')
|
|
|
|
if self.icon_type != self.template_args['icon_type']['default']:
|
|
self.logger.warning(
|
|
'LaMetric icon_type setting is unavailable in Cloud mode')
|
|
|
|
if self.cycles != self.template_args['cycles']['default']:
|
|
self.logger.warning(
|
|
'LaMetric cycle settings is unavailable in Cloud mode')
|
|
|
|
# Assign our icon if the user specified a custom one, otherwise
|
|
# choose from our pre-set list (based on notify_type)
|
|
icon = self.icon if self.icon \
|
|
else self.lametric_icon_id_mapping[notify_type]
|
|
|
|
# Our Payload
|
|
# Cloud Notifications don't have as much functionality
|
|
# You can not set priority and/or sound
|
|
payload = {
|
|
"frames": [
|
|
{
|
|
"icon": icon,
|
|
"text": body,
|
|
"index": 0,
|
|
}
|
|
]
|
|
}
|
|
|
|
# Prepare our Cloud Notify URL
|
|
notify_url = self.cloud_notify_url.format(
|
|
app_id=self.lametric_app_id, app_ver=self.lametric_app_ver)
|
|
|
|
# Return request parameters
|
|
return (notify_url, None, payload)
|
|
|
|
def _device_notification_payload(self, body, notify_type, headers):
|
|
"""
|
|
Return URL and Payload for Device directed requests
|
|
"""
|
|
|
|
# Assign our icon if the user specified a custom one, otherwise
|
|
# choose from our pre-set list (based on notify_type)
|
|
icon = self.icon if self.icon \
|
|
else self.lametric_icon_id_mapping[notify_type]
|
|
|
|
# Our Payload
|
|
payload = {
|
|
# Priority of the message
|
|
"priority": self.priority,
|
|
|
|
# Icon Type: Represents the nature of notification
|
|
"icon_type": self.icon_type,
|
|
|
|
# The time notification lives in queue to be displayed in
|
|
# milliseconds (ms). The default lifetime is 2 minutes (120000ms).
|
|
# If notification stayed in queue for longer than lifetime
|
|
# milliseconds - it will not be displayed.
|
|
"lifetime": 120000,
|
|
|
|
"model": {
|
|
# cycles - the number of times message should be displayed. If
|
|
# cycles is set to 0, notification will stay on the screen
|
|
# until user dismisses it manually. By default it is set to 1.
|
|
"cycles": self.cycles,
|
|
"frames": [
|
|
{
|
|
"icon": icon,
|
|
"text": body,
|
|
}
|
|
]
|
|
}
|
|
}
|
|
|
|
if self.sound:
|
|
# Sound was set, so add it to the payload
|
|
payload["model"]["sound"] = {
|
|
# The sound category
|
|
"category": self.sound[0],
|
|
|
|
# The first element of our tuple is always the id
|
|
"id": self.sound[1][0],
|
|
|
|
# repeat - defines the number of times sound must be played.
|
|
# If set to 0 sound will be played until notification is
|
|
# dismissed. By default the value is set to 1.
|
|
"repeat": 1,
|
|
}
|
|
|
|
if not self.user:
|
|
# Use default user if there wasn't one otherwise specified
|
|
self.user = self.default_device_user
|
|
|
|
# Prepare our authentication
|
|
auth = (self.user, self.password)
|
|
|
|
# Prepare our Direct Access Notify URL
|
|
notify_url = self.device_notify_url.format(
|
|
schema="https" if self.secure else "http",
|
|
host=self.host,
|
|
port=':{}'.format(
|
|
self.port if self.port
|
|
else self.template_tokens['port']['default']))
|
|
|
|
# Return request parameters
|
|
return (notify_url, auth, payload)
|
|
|
|
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
|
|
"""
|
|
Perform LaMetric Notification
|
|
"""
|
|
|
|
# Prepare our headers:
|
|
headers = {
|
|
'User-Agent': self.app_id,
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'Cache-Control': 'no-cache',
|
|
}
|
|
|
|
# Depending on the mode, the payload is gathered by
|
|
# - _device_notification_payload()
|
|
# - _cloud_notification_payload()
|
|
(notify_url, auth, payload) = getattr(
|
|
self, '_{}_notification_payload'.format(self.mode))(
|
|
body=body, notify_type=notify_type, headers=headers)
|
|
|
|
self.logger.debug('LaMetric POST URL: %s (cert_verify=%r)' % (
|
|
notify_url, self.verify_certificate,
|
|
))
|
|
self.logger.debug('LaMetric Payload: %s' % str(payload))
|
|
|
|
# Always call throttle before any remote server i/o is made
|
|
self.throttle()
|
|
|
|
try:
|
|
r = requests.post(
|
|
notify_url,
|
|
data=dumps(payload),
|
|
headers=headers,
|
|
auth=auth,
|
|
verify=self.verify_certificate,
|
|
timeout=self.request_timeout,
|
|
)
|
|
# An ideal response would be:
|
|
# {
|
|
# "success": {
|
|
# "id": "<notification id>"
|
|
# }
|
|
# }
|
|
|
|
if r.status_code not in (
|
|
requests.codes.created, requests.codes.ok):
|
|
# We had a problem
|
|
status_str = \
|
|
NotifyLametric.http_response_code_lookup(r.status_code)
|
|
|
|
self.logger.warning(
|
|
'Failed to send LaMetric notification: '
|
|
'{}{}error={}.'.format(
|
|
status_str,
|
|
', ' if status_str else '',
|
|
r.status_code))
|
|
|
|
self.logger.debug('Response Details:\r\n{}'.format(r.content))
|
|
|
|
# Return; we're done
|
|
return False
|
|
|
|
else:
|
|
self.logger.info('Sent LaMetric notification.')
|
|
|
|
except requests.RequestException as e:
|
|
self.logger.warning(
|
|
'A Connection error occurred sending LaMetric '
|
|
'notification to %s.' % self.host)
|
|
self.logger.debug('Socket Exception: %s' % str(e))
|
|
|
|
# Return; we're done
|
|
return False
|
|
|
|
return True
|
|
|
|
def url(self, privacy=False, *args, **kwargs):
|
|
"""
|
|
Returns the URL built dynamically based on specified arguments.
|
|
"""
|
|
|
|
# Define any URL parameters
|
|
params = {
|
|
'mode': self.mode,
|
|
}
|
|
|
|
# Extend our parameters
|
|
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
|
|
|
|
if self.icon:
|
|
# Assign our icon IF one was specified
|
|
params['icon'] = self.icon
|
|
|
|
if self.mode == LametricMode.CLOUD:
|
|
# Upstream/LaMetric App Return
|
|
return '{schema}://{token}@{app_id}/{app_ver}/?{params}'.format(
|
|
schema=self.protocol,
|
|
token=self.pprint(
|
|
self.lametric_app_access_token, privacy, safe=''),
|
|
app_id=self.pprint(self.lametric_app_id, privacy, safe=''),
|
|
app_ver=NotifyLametric.quote(self.lametric_app_ver, safe=''),
|
|
params=NotifyLametric.urlencode(params))
|
|
|
|
#
|
|
# If we reach here then we're dealing with LametricMode.DEVICE
|
|
#
|
|
if self.priority != self.template_args['priority']['default']:
|
|
params['priority'] = self.priority
|
|
|
|
if self.icon_type != self.template_args['icon_type']['default']:
|
|
params['icon_type'] = self.icon_type
|
|
|
|
if self.cycles != self.template_args['cycles']['default']:
|
|
params['cycles'] = self.cycles
|
|
|
|
if self.sound:
|
|
# Store our sound entry
|
|
# The first element of our tuple is always the id
|
|
params['sound'] = self.sound[1][0]
|
|
|
|
auth = ''
|
|
if self.user and self.password:
|
|
auth = '{user}:{apikey}@'.format(
|
|
user=NotifyLametric.quote(self.user, safe=''),
|
|
apikey=self.pprint(self.lametric_apikey, privacy, safe=''),
|
|
)
|
|
else: # self.apikey is set
|
|
auth = '{apikey}@'.format(
|
|
apikey=self.pprint(self.lametric_apikey, privacy, safe=''),
|
|
)
|
|
|
|
# Local Return
|
|
return '{schema}://{auth}{hostname}{port}/?{params}'.format(
|
|
schema=self.secure_protocol if self.secure else self.protocol,
|
|
auth=auth,
|
|
# never encode hostname since we're expecting it to be a valid one
|
|
hostname=self.host,
|
|
port='' if self.port is None
|
|
or self.port == self.template_tokens['port']['default']
|
|
else ':{}'.format(self.port),
|
|
params=NotifyLametric.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
|
|
|
|
if results.get('user') and not results.get('password'):
|
|
# Handle URL like:
|
|
# schema://user@host
|
|
|
|
# This becomes the password
|
|
results['password'] = results['user']
|
|
results['user'] = None
|
|
|
|
# Priority Handling
|
|
if 'priority' in results['qsd'] and results['qsd']['priority']:
|
|
results['priority'] = NotifyLametric.unquote(
|
|
results['qsd']['priority'].strip().lower())
|
|
|
|
# Icon Type
|
|
if 'icon' in results['qsd'] and results['qsd']['icon']:
|
|
results['icon'] = NotifyLametric.unquote(
|
|
results['qsd']['icon'].strip().lower())
|
|
|
|
# Icon Type
|
|
if 'icon_type' in results['qsd'] and results['qsd']['icon_type']:
|
|
results['icon_type'] = NotifyLametric.unquote(
|
|
results['qsd']['icon_type'].strip().lower())
|
|
|
|
# Sound
|
|
if 'sound' in results['qsd'] and results['qsd']['sound']:
|
|
results['sound'] = NotifyLametric.unquote(
|
|
results['qsd']['sound'].strip().lower())
|
|
|
|
# API Key (Device Mode)
|
|
if 'apikey' in results['qsd'] and results['qsd']['apikey']:
|
|
# Extract API Key from an argument
|
|
results['apikey'] = \
|
|
NotifyLametric.unquote(results['qsd']['apikey'])
|
|
|
|
# App ID
|
|
if 'app' in results['qsd'] \
|
|
and results['qsd']['app']:
|
|
|
|
# Extract the App ID from an argument
|
|
results['app_id'] = \
|
|
NotifyLametric.unquote(results['qsd']['app'])
|
|
|
|
# App Version
|
|
if 'app_ver' in results['qsd'] \
|
|
and results['qsd']['app_ver']:
|
|
|
|
# Extract the App ID from an argument
|
|
results['app_ver'] = \
|
|
NotifyLametric.unquote(results['qsd']['app_ver'])
|
|
|
|
if 'token' in results['qsd'] and results['qsd']['token']:
|
|
# Extract Application Access Token from an argument
|
|
results['app_token'] = \
|
|
NotifyLametric.unquote(results['qsd']['token'])
|
|
|
|
# Mode override
|
|
if 'mode' in results['qsd'] and results['qsd']['mode']:
|
|
results['mode'] = NotifyLametric.unquote(
|
|
results['qsd']['mode'].strip().lower())
|
|
else:
|
|
# We can try to detect the mode based on the validity of the
|
|
# hostname. We can also scan the validity of the Application
|
|
# Access token
|
|
#
|
|
# This isn't a surfire way to do things though; it's best to
|
|
# specify the mode= flag
|
|
results['mode'] = LametricMode.DEVICE \
|
|
if ((is_hostname(results['host']) or
|
|
is_ipaddr(results['host'])) and
|
|
|
|
# make sure password is not an Access Token
|
|
(results['password'] and not
|
|
LAMETRIC_IS_APP_TOKEN.match(results['password'])) and
|
|
|
|
# Scan for app_ flags
|
|
next((f for f in results.keys() \
|
|
if f.startswith('app_')), None) is None) \
|
|
else LametricMode.CLOUD
|
|
|
|
# Handle defaults if not set
|
|
if results['mode'] == LametricMode.DEVICE:
|
|
# Device Mode Defaults
|
|
if 'apikey' not in results:
|
|
results['apikey'] = \
|
|
NotifyLametric.unquote(results['password'])
|
|
|
|
else:
|
|
# CLOUD Mode Defaults
|
|
if 'app_id' not in results:
|
|
results['app_id'] = \
|
|
NotifyLametric.unquote(results['host'])
|
|
if 'app_token' not in results:
|
|
results['app_token'] = \
|
|
NotifyLametric.unquote(results['password'])
|
|
|
|
# Set cycles
|
|
try:
|
|
results['cycles'] = abs(int(results['qsd'].get('cycles')))
|
|
|
|
except (TypeError, ValueError):
|
|
# Not a valid integer; ignore entry
|
|
pass
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def parse_native_url(url):
|
|
"""
|
|
Support
|
|
https://developer.lametric.com/api/v1/dev/\
|
|
widget/update/com.lametric.{APP_ID}/1
|
|
|
|
https://developer.lametric.com/api/v1/dev/\
|
|
widget/update/com.lametric.{APP_ID}/{APP_VER}
|
|
"""
|
|
|
|
# If users do provide the Native URL they wll also want to add
|
|
# ?token={APP_ACCESS_TOKEN} to the parameters at the end or the
|
|
# URL will fail to load in later stages.
|
|
result = re.match(
|
|
r'^http(?P<secure>s)?://(?P<host>[^/]+)'
|
|
r'/api/(?P<api_ver>v[1-9]*[0-9]+)'
|
|
r'/dev/widget/update/'
|
|
r'com\.lametric\.(?P<app_id>[0-9a-z.-]{1,64})'
|
|
r'(/(?P<app_ver>[1-9][0-9]*))?/?'
|
|
r'(?P<params>\?.+)?$', url, re.I)
|
|
|
|
if result:
|
|
return NotifyLametric.parse_url(
|
|
'{schema}://{app_id}{app_ver}/{params}'.format(
|
|
schema=NotifyLametric.secure_protocol
|
|
if result.group('secure') else NotifyLametric.protocol,
|
|
app_id=result.group('app_id'),
|
|
app_ver='/{}'.format(result.group('app_ver'))
|
|
if result.group('app_ver') else '',
|
|
params='' if not result.group('params')
|
|
else result.group('params')))
|
|
|
|
return None
|