1# -*- coding: utf-8 -*-
2#
3# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
4# All rights reserved.
5#
6# This code is licensed under the MIT License.
7#
8# Permission is hereby granted, free of charge, to any person obtaining a copy
9# of this software and associated documentation files(the "Software"), to deal
10# in the Software without restriction, including without limitation the rights
11# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
12# copies of the Software, and to permit persons to whom the Software is
13# furnished to do so, subject to the following conditions :
14#
15# The above copyright notice and this permission notice shall be included in
16# all copies or substantial portions of the Software.
17#
18# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
19# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
20# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
21# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
22# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
23# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
24# THE SOFTWARE.
25
26from __future__ import absolute_import
27from __future__ import print_function
28
29from .NotifyBase import NotifyBase
30from ..common import NotifyImageSize
31from ..common import NotifyType
32from ..utils import parse_bool
33from ..AppriseLocale import gettext_lazy as _
34
35# Default our global support flag
36NOTIFY_DBUS_SUPPORT_ENABLED = False
37
38# Image support is dependant on the GdkPixbuf library being available
39NOTIFY_DBUS_IMAGE_SUPPORT = False
40
41# Initialize our mainloops
42LOOP_GLIB = None
43LOOP_QT = None
44
45
46try:
47    # dbus essentials
48    from dbus import SessionBus
49    from dbus import Interface
50    from dbus import Byte
51    from dbus import ByteArray
52    from dbus import DBusException
53
54    #
55    # now we try to determine which mainloop(s) we can access
56    #
57
58    # glib
59    try:
60        from dbus.mainloop.glib import DBusGMainLoop
61        LOOP_GLIB = DBusGMainLoop()
62
63    except ImportError:
64        # No problem
65        pass
66
67    # qt
68    try:
69        from dbus.mainloop.qt import DBusQtMainLoop
70        LOOP_QT = DBusQtMainLoop(set_as_default=True)
71
72    except ImportError:
73        # No problem
74        pass
75
76    # We're good as long as at least one
77    NOTIFY_DBUS_SUPPORT_ENABLED = (
78        LOOP_GLIB is not None or LOOP_QT is not None)
79
80    try:
81        # The following is required for Image/Icon loading only
82        import gi
83        gi.require_version('GdkPixbuf', '2.0')
84        from gi.repository import GdkPixbuf
85        NOTIFY_DBUS_IMAGE_SUPPORT = True
86
87    except (ImportError, ValueError, AttributeError):
88        # No problem; this will get caught in outer try/catch
89
90        # A ValueError will get thrown upon calling gi.require_version() if
91        # GDK/GTK isn't installed on the system but gi is.
92        pass
93
94except ImportError:
95    # No problem; we just simply can't support this plugin; we could
96    # be in microsoft windows, or we just don't have the python-gobject
97    # library available to us (or maybe one we don't support)?
98    pass
99
100# Define our supported protocols and the loop to assign them.
101# The key to value pairs are the actual supported schema's matched
102# up with the Main Loop they should reference when accessed.
103MAINLOOP_MAP = {
104    'qt': LOOP_QT,
105    'kde': LOOP_QT,
106    'glib': LOOP_GLIB,
107    'dbus': LOOP_QT if LOOP_QT else LOOP_GLIB,
108}
109
110
111# Urgencies
112class DBusUrgency(object):
113    LOW = 0
114    NORMAL = 1
115    HIGH = 2
116
117
118# Define our urgency levels
119DBUS_URGENCIES = (
120    DBusUrgency.LOW,
121    DBusUrgency.NORMAL,
122    DBusUrgency.HIGH,
123)
124
125
126class NotifyDBus(NotifyBase):
127    """
128    A wrapper for local DBus/Qt Notifications
129    """
130
131    # Set our global enabled flag
132    enabled = NOTIFY_DBUS_SUPPORT_ENABLED
133
134    requirements = {
135        # Define our required packaging in order to work
136        'details': _('libdbus-1.so.x must be installed.')
137    }
138
139    # The default descriptive name associated with the Notification
140    service_name = _('DBus Notification')
141
142    # The services URL
143    service_url = 'http://www.freedesktop.org/Software/dbus/'
144
145    # The default protocols
146    # Python 3 keys() does not return a list object, it's it's own dict_keys()
147    # object if we were to reference, we wouldn't be backwards compatible with
148    # Python v2.  So converting the result set back into a list makes us
149    # compatible
150    protocol = list(MAINLOOP_MAP.keys())
151
152    # A URL that takes you to the setup/help of the specific protocol
153    setup_url = 'https://github.com/caronc/apprise/wiki/Notify_dbus'
154
155    # No throttling required for DBus queries
156    request_rate_per_sec = 0
157
158    # Allows the user to specify the NotifyImageSize object
159    image_size = NotifyImageSize.XY_128
160
161    # The number of milliseconds to keep the message present for
162    message_timeout_ms = 13000
163
164    # Limit results to just the first 10 line otherwise there is just to much
165    # content to display
166    body_max_line_count = 10
167
168    # The following are required to hook into the notifications:
169    dbus_interface = 'org.freedesktop.Notifications'
170    dbus_setting_location = '/org/freedesktop/Notifications'
171
172    # Define object templates
173    templates = (
174        '{schema}://',
175    )
176
177    # Define our template arguments
178    template_args = dict(NotifyBase.template_args, **{
179        'urgency': {
180            'name': _('Urgency'),
181            'type': 'choice:int',
182            'values': DBUS_URGENCIES,
183            'default': DBusUrgency.NORMAL,
184        },
185        'x': {
186            'name': _('X-Axis'),
187            'type': 'int',
188            'min': 0,
189            'map_to': 'x_axis',
190        },
191        'y': {
192            'name': _('Y-Axis'),
193            'type': 'int',
194            'min': 0,
195            'map_to': 'y_axis',
196        },
197        'image': {
198            'name': _('Include Image'),
199            'type': 'bool',
200            'default': True,
201            'map_to': 'include_image',
202        },
203    })
204
205    def __init__(self, urgency=None, x_axis=None, y_axis=None,
206                 include_image=True, **kwargs):
207        """
208        Initialize DBus Object
209        """
210
211        super(NotifyDBus, self).__init__(**kwargs)
212
213        # Track our notifications
214        self.registry = {}
215
216        # Store our schema; default to dbus
217        self.schema = kwargs.get('schema', 'dbus')
218
219        if self.schema not in MAINLOOP_MAP:
220            msg = 'The schema specified ({}) is not supported.' \
221                .format(self.schema)
222            self.logger.warning(msg)
223            raise TypeError(msg)
224
225        # The urgency of the message
226        if urgency not in DBUS_URGENCIES:
227            self.urgency = DBusUrgency.NORMAL
228
229        else:
230            self.urgency = urgency
231
232        # Our x/y axis settings
233        self.x_axis = x_axis if isinstance(x_axis, int) else None
234        self.y_axis = y_axis if isinstance(y_axis, int) else None
235
236        # Track whether or not we want to send an image with our notification
237        # or not.
238        self.include_image = include_image
239
240        return
241
242    def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
243        """
244        Perform DBus Notification
245        """
246        # Acquire our session
247        try:
248            session = SessionBus(mainloop=MAINLOOP_MAP[self.schema])
249
250        except DBusException:
251            # Handle exception
252            self.logger.warning('Failed to send DBus notification.')
253            self.logger.exception('DBus Exception')
254            return False
255
256        # If there is no title, but there is a body, swap the two to get rid
257        # of the weird whitespace
258        if not title:
259            title = body
260            body = ''
261
262        # acquire our dbus object
263        dbus_obj = session.get_object(
264            self.dbus_interface,
265            self.dbus_setting_location,
266        )
267
268        # Acquire our dbus interface
269        dbus_iface = Interface(
270            dbus_obj,
271            dbus_interface=self.dbus_interface,
272        )
273
274        # image path
275        icon_path = None if not self.include_image \
276            else self.image_path(notify_type, extension='.ico')
277
278        # Our meta payload
279        meta_payload = {
280            "urgency": Byte(self.urgency)
281        }
282
283        if not (self.x_axis is None and self.y_axis is None):
284            # Set x/y access if these were set
285            meta_payload['x'] = self.x_axis
286            meta_payload['y'] = self.y_axis
287
288        if NOTIFY_DBUS_IMAGE_SUPPORT and icon_path:
289            try:
290                # Use Pixbuf to create the proper image type
291                image = GdkPixbuf.Pixbuf.new_from_file(icon_path)
292
293                # Associate our image to our notification
294                meta_payload['icon_data'] = (
295                    image.get_width(),
296                    image.get_height(),
297                    image.get_rowstride(),
298                    image.get_has_alpha(),
299                    image.get_bits_per_sample(),
300                    image.get_n_channels(),
301                    ByteArray(image.get_pixels())
302                )
303
304            except Exception as e:
305                self.logger.warning(
306                    "Could not load Gnome notification icon ({}): {}"
307                    .format(icon_path, e))
308
309        try:
310            # Always call throttle() before any remote execution is made
311            self.throttle()
312
313            dbus_iface.Notify(
314                # Application Identifier
315                self.app_id,
316                # Message ID (0 = New Message)
317                0,
318                # Icon (str) - not used
319                '',
320                # Title
321                str(title),
322                # Body
323                str(body),
324                # Actions
325                list(),
326                # Meta
327                meta_payload,
328                # Message Timeout
329                self.message_timeout_ms,
330            )
331
332            self.logger.info('Sent DBus notification.')
333
334        except Exception:
335            self.logger.warning('Failed to send DBus notification.')
336            self.logger.exception('DBus Exception')
337            return False
338
339        return True
340
341    def url(self, privacy=False, *args, **kwargs):
342        """
343        Returns the URL built dynamically based on specified arguments.
344        """
345
346        _map = {
347            DBusUrgency.LOW: 'low',
348            DBusUrgency.NORMAL: 'normal',
349            DBusUrgency.HIGH: 'high',
350        }
351
352        # Define any URL parameters
353        params = {
354            'image': 'yes' if self.include_image else 'no',
355            'urgency': 'normal' if self.urgency not in _map
356                       else _map[self.urgency],
357        }
358
359        # Extend our parameters
360        params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
361
362        # x in (x,y) screen coordinates
363        if self.x_axis:
364            params['x'] = str(self.x_axis)
365
366        # y in (x,y) screen coordinates
367        if self.y_axis:
368            params['y'] = str(self.y_axis)
369
370        return '{schema}://_/?{params}'.format(
371            schema=self.schema,
372            params=NotifyDBus.urlencode(params),
373        )
374
375    @staticmethod
376    def parse_url(url):
377        """
378        There are no parameters nessisary for this protocol; simply having
379        gnome:// is all you need.  This function just makes sure that
380        is in place.
381
382        """
383
384        results = NotifyBase.parse_url(url, verify_host=False)
385
386        # Include images with our message
387        results['include_image'] = \
388            parse_bool(results['qsd'].get('image', True))
389
390        # DBus supports urgency, but we we also support the keyword priority
391        # so that it is consistent with some of the other plugins
392        urgency = results['qsd'].get('urgency', results['qsd'].get('priority'))
393        if urgency and len(urgency):
394            _map = {
395                '0': DBusUrgency.LOW,
396                'l': DBusUrgency.LOW,
397                'n': DBusUrgency.NORMAL,
398                '1': DBusUrgency.NORMAL,
399                'h': DBusUrgency.HIGH,
400                '2': DBusUrgency.HIGH,
401            }
402
403            try:
404                # Attempt to index/retrieve our urgency
405                results['urgency'] = _map[urgency[0].lower()]
406
407            except KeyError:
408                # No priority was set
409                pass
410
411        # handle x,y coordinates
412        try:
413            results['x_axis'] = int(results['qsd'].get('x'))
414
415        except (TypeError, ValueError):
416            # No x was set
417            pass
418
419        try:
420            results['y_axis'] = int(results['qsd'].get('y'))
421
422        except (TypeError, ValueError):
423            # No y was set
424            pass
425
426        return results
427