Permalink
Cannot retrieve contributors at this time
Fetching contributors…
Cannot retrieve contributors at this time
| # Copyright: 2013 Paul Traylor | |
| # These sources are released under the terms of the MIT license: see LICENSE | |
| """ | |
| The gntp.notifier module is provided as a simple way to send notifications | |
| using GNTP | |
| .. note:: | |
| This class is intended to mostly mirror the older Python bindings such | |
| that you should be able to replace instances of the old bindings with | |
| this class. | |
| `Original Python bindings <http://code.google.com/p/growl/source/browse/Bindings/python/Growl.py>`_ | |
| """ | |
| import logging | |
| import platform | |
| import socket | |
| import sys | |
| from gntp.version import __version__ | |
| import gntp.core | |
| import gntp.errors as errors | |
| import gntp.shim | |
| __all__ = [ | |
| 'mini', | |
| 'GrowlNotifier', | |
| ] | |
| logger = logging.getLogger(__name__) | |
| class GrowlNotifier(object): | |
| """Helper class to simplfy sending Growl messages | |
| :param string applicationName: Sending application name | |
| :param list notification: List of valid notifications | |
| :param list defaultNotifications: List of notifications that should be enabled | |
| by default | |
| :param string applicationIcon: Icon URL | |
| :param string hostname: Remote host | |
| :param integer port: Remote port | |
| """ | |
| passwordHash = 'MD5' | |
| socketTimeout = 3 | |
| def __init__(self, applicationName='Python GNTP', notifications=[], | |
| defaultNotifications=None, applicationIcon=None, hostname='localhost', | |
| password=None, port=23053): | |
| self.applicationName = applicationName | |
| self.notifications = list(notifications) | |
| if defaultNotifications: | |
| self.defaultNotifications = list(defaultNotifications) | |
| else: | |
| self.defaultNotifications = self.notifications | |
| self.applicationIcon = applicationIcon | |
| self.password = password | |
| self.hostname = hostname | |
| self.port = int(port) | |
| def _checkIcon(self, data): | |
| ''' | |
| Check the icon to see if it's valid | |
| If it's a simple URL icon, then we return True. If it's a data icon | |
| then we return False | |
| ''' | |
| logger.info('Checking icon') | |
| return gntp.shim.u(data)[:4] in ['http', 'file'] | |
| def register(self): | |
| """Send GNTP Registration | |
| .. warning:: | |
| Before sending notifications to Growl, you need to have | |
| sent a registration message at least once | |
| """ | |
| logger.info('Sending registration to %s:%s', self.hostname, self.port) | |
| register = gntp.core.GNTPRegister() | |
| register.add_header('Application-Name', self.applicationName) | |
| for notification in self.notifications: | |
| enabled = notification in self.defaultNotifications | |
| register.add_notification(notification, enabled) | |
| if self.applicationIcon: | |
| if self._checkIcon(self.applicationIcon): | |
| register.add_header('Application-Icon', self.applicationIcon) | |
| else: | |
| resource = register.add_resource(self.applicationIcon) | |
| register.add_header('Application-Icon', resource) | |
| if self.password: | |
| register.set_password(self.password, self.passwordHash) | |
| self.add_origin_info(register) | |
| self.register_hook(register) | |
| return self._send('register', register) | |
| def notify(self, noteType, title, description, icon=None, sticky=False, | |
| priority=None, callback=None, identifier=None, custom={}): | |
| """Send a GNTP notifications | |
| .. warning:: | |
| Must have registered with growl beforehand or messages will be ignored | |
| :param string noteType: One of the notification names registered earlier | |
| :param string title: Notification title (usually displayed on the notification) | |
| :param string description: The main content of the notification | |
| :param string icon: Icon URL path | |
| :param boolean sticky: Sticky notification | |
| :param integer priority: Message priority level from -2 to 2 | |
| :param string callback: URL callback | |
| :param dict custom: Custom attributes. Key names should be prefixed with X- | |
| according to the spec but this is not enforced by this class | |
| .. warning:: | |
| For now, only URL callbacks are supported. In the future, the | |
| callback argument will also support a function | |
| """ | |
| logger.info('Sending notification [%s] to %s:%s', noteType, self.hostname, self.port) | |
| assert noteType in self.notifications | |
| notice = gntp.core.GNTPNotice() | |
| notice.add_header('Application-Name', self.applicationName) | |
| notice.add_header('Notification-Name', noteType) | |
| notice.add_header('Notification-Title', title) | |
| if self.password: | |
| notice.set_password(self.password, self.passwordHash) | |
| if sticky: | |
| notice.add_header('Notification-Sticky', sticky) | |
| if priority: | |
| notice.add_header('Notification-Priority', priority) | |
| if icon: | |
| if self._checkIcon(icon): | |
| notice.add_header('Notification-Icon', icon) | |
| else: | |
| resource = notice.add_resource(icon) | |
| notice.add_header('Notification-Icon', resource) | |
| if description: | |
| notice.add_header('Notification-Text', description) | |
| if callback: | |
| notice.add_header('Notification-Callback-Target', callback) | |
| if identifier: | |
| notice.add_header('Notification-Coalescing-ID', identifier) | |
| for key in custom: | |
| notice.add_header(key, custom[key]) | |
| self.add_origin_info(notice) | |
| self.notify_hook(notice) | |
| return self._send('notify', notice) | |
| def subscribe(self, id, name, port): | |
| """Send a Subscribe request to a remote machine""" | |
| sub = gntp.core.GNTPSubscribe() | |
| sub.add_header('Subscriber-ID', id) | |
| sub.add_header('Subscriber-Name', name) | |
| sub.add_header('Subscriber-Port', port) | |
| if self.password: | |
| sub.set_password(self.password, self.passwordHash) | |
| self.add_origin_info(sub) | |
| self.subscribe_hook(sub) | |
| return self._send('subscribe', sub) | |
| def add_origin_info(self, packet): | |
| """Add optional Origin headers to message""" | |
| packet.add_header('Origin-Machine-Name', platform.node()) | |
| packet.add_header('Origin-Software-Name', 'gntp.py') | |
| packet.add_header('Origin-Software-Version', __version__) | |
| packet.add_header('Origin-Platform-Name', platform.system()) | |
| packet.add_header('Origin-Platform-Version', platform.platform()) | |
| def register_hook(self, packet): | |
| pass | |
| def notify_hook(self, packet): | |
| pass | |
| def subscribe_hook(self, packet): | |
| pass | |
| def _send(self, messagetype, packet): | |
| """Send the GNTP Packet""" | |
| packet.validate() | |
| data = packet.encode() | |
| logger.debug('To : %s:%s <%s>\n%s', self.hostname, self.port, packet.__class__, data) | |
| s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
| s.settimeout(self.socketTimeout) | |
| try: | |
| s.connect((self.hostname, self.port)) | |
| s.send(data) | |
| recv_data = s.recv(1024) | |
| while not recv_data.endswith(gntp.shim.b("\r\n\r\n")): | |
| recv_data += s.recv(1024) | |
| except socket.error: | |
| # Python2.5 and Python3 compatibile exception | |
| exc = sys.exc_info()[1] | |
| raise errors.NetworkError(exc) | |
| response = gntp.core.parse_gntp(recv_data) | |
| s.close() | |
| logger.debug('From : %s:%s <%s>\n%s', self.hostname, self.port, response.__class__, response) | |
| if type(response) == gntp.core.GNTPOK: | |
| return True | |
| logger.error('Invalid response: %s', response.error()) | |
| return response.error() | |
| def mini(description, applicationName='PythonMini', noteType="Message", | |
| title="Mini Message", applicationIcon=None, hostname='localhost', | |
| password=None, port=23053, sticky=False, priority=None, | |
| callback=None, notificationIcon=None, identifier=None, | |
| notifierFactory=GrowlNotifier): | |
| """Single notification function | |
| Simple notification function in one line. Has only one required parameter | |
| and attempts to use reasonable defaults for everything else | |
| :param string description: Notification message | |
| .. warning:: | |
| For now, only URL callbacks are supported. In the future, the | |
| callback argument will also support a function | |
| """ | |
| try: | |
| growl = notifierFactory( | |
| applicationName=applicationName, | |
| notifications=[noteType], | |
| defaultNotifications=[noteType], | |
| applicationIcon=applicationIcon, | |
| hostname=hostname, | |
| password=password, | |
| port=port, | |
| ) | |
| result = growl.register() | |
| if result is not True: | |
| return result | |
| return growl.notify( | |
| noteType=noteType, | |
| title=title, | |
| description=description, | |
| icon=notificationIcon, | |
| sticky=sticky, | |
| priority=priority, | |
| callback=callback, | |
| identifier=identifier, | |
| ) | |
| except Exception: | |
| # We want the "mini" function to be simple and swallow Exceptions | |
| # in order to be less invasive | |
| logger.exception("Growl error") | |
| if __name__ == '__main__': | |
| # If we're running this module directly we're likely running it as a test | |
| # so extra debugging is useful | |
| logging.basicConfig(level=logging.INFO) | |
| mini('Testing mini notification') |