canto-curses notification plugin

canto-curses-notify-plugin.py

'''
canto-curses-notify v0.3 | Copyright (c) 2015, Pablo Cabeza

MIT Licese (http://opensource.org/licenses/mit-license.php)


Notification plugin for
[canto-curses](https://github.com/themoken/canto-curses) that
integrates with desktop notifications. See [Canto
webpage](http://codezen.org/canto-ng/) for more info.

Requires:
=========
- `PyGObject` (`PyGI`)


Changeslist:
============

v0.4:
-----
- [ ] command to close all opened notification
- [ ] command to control whether enable or disable notifications
- [ ] when a feed is added, dont show notifications for it
- [ ] merge notifications into one where there are many new stories

v0.3:
-----
- [X] reverted to `gi.repository.Notify` due to bug #2
- [X] action for *mark as read*
- [X] when reading or marking as read a story on canto, close its
      notification bubble

v0.2:
-----
- [X] change to `notfy2` library
- [X] fixed notify2 signal, adding glib mainloop in a new thread
- [X] close all notifications on canto close (hook on curses_exit_)
- [X] each new story in a separate notification bubble
- [X] added `read` action to notification that syncs with canto

v0.1:
-----
- [X] hook on stories_added and parse unread stories
- [X] notify using python-gtk3 Notify library

Bugs list:
==========
1. [X] On initializing with `curses_stories_added` signal and
       accessing "canto-state", sometimes is empty and raises
       exception.
       - **Fix**: wrap code in try-except to avoid exception for now

2. [X] Exception `dbus.exceptions.DBusException:
       org.freedesktop.DBus.Error.ServiceUnknown: The name :1.100 as
       not provided by any .service files` on `noti.show()`
       - **Fix**: wrap show() in try-except
       - **Fix 2**: Revert to `gi.respository.Notify`, seems fixed
'''

import logging
import traceback
import threading
import webbrowser

from canto_next.plugins import check_program
from canto_next.hooks import on_hook
from canto_curses.tagcore import tag_updater

import glib
import dbus
import gi
gi.require_version('Notify', '0.7'); from gi.repository import Notify


# Uncomment for use this extra check with canto-curses
# check_program("canto-curses")

log = logging.getLogger("NOTIFY")
log.setLevel(logging.DEBUG)


# Notification clases and factory
# ===============================

class Notification(Notify.Notification):
    """
    Notification wrapper to add needed functionality
    """

    def __init__(self, title, text, file_path_to_icon="",
                 timeout=Notify.EXPIRES_NEVER):
        self.title = title
        self.text = text
        self.file_path_to_icon = file_path_to_icon

        super(Notification, self).__init__(
            summary=title,
            body=text,
            icon_name=file_path_to_icon
        )

        self.set_timeout(timeout)

    @property
    def id(self): return self.get_property("id")

    def on_close(self, callback, *args):
        "connects callback to `closed` signal"
        return self.connect('closed', callback, *args)


class StoryNotification(Notification):
    '''
    Specific notifications for canto-curses
    '''

    def LINK(l, t): return '%(title)s' % {"link": l, "title": t}
    STORAGE = None

    def __init__(self, tag, story):
        '''
        Contructs Notification from a story and a tag.

        - `tag` (`canto_curses.tag.Tag`): a feed tag
        - `story` (`canto_curses.story.Story`): a story from `tag`
        '''

        self.tag = tag
        self.story = story

        self.tagname = str(tag)
        self.link = story.content['link']
        self.storytitle = story.content['title']

        log.debug(
            '[creating] notificaton %s [%s]' %
            (self.storytitle, self.tagname)
        )

        super(StoryNotification, self).__init__(
            self.tagname,
            StoryNotification.LINK(self.link, self.storytitle)
        )


class Notification_factory(object):
    '''
    Factory for notifications (meant for gobject instrocpection)
    '''

    def __init__(self):
        super(Notification_factory, self).__init__()

        # lets initialise with the application name
        Notify.init("canto-reader-notify-plugin")

    def send_notification(self, title, text, file_path_to_icon=""):
        return Notification(title, text, file_path_to_icon)

    def send_storynotif(self, tag, story):
        return StoryNotification(tag, story)


# Manager for notification signals and hooks
# ==========================================

class StoryManager(object):
    '''
    Manage stories notifications and hook to canto-curses and
    notificatins signals
    '''

    def __init__(self, mainloop=None):
        self.factory = Notification_factory()

        self.notifications = []
        StoryNotification.STORAGE = self.notifications

        on_hook("curses_stories_added", self.on_stories_added)
        on_hook("curses_exit", self.on_curses_exit)
        on_hook("curses_attributes", self.on_attributes)

        # start glib mainloop if no mainloop is passed
        if mainloop is None:
            glib.threads_init()
            mainloop = glib.MainLoop()
            t = threading.Thread(target=mainloop.run)
            t.daemon = True
            t.start()

        self.mainloop = mainloop

    @staticmethod
    def _get_read_state(story):
        "Auxiliary method to fetch 'read' state from a story"
        story.sync()  # Update content of a story to avoid bug #1
        return "read" in story.content["canto-state"]

    def on_stories_added(self, tag, stories):
        '''
        Callback for `curses_stories_added` signal, shows
        notifications and adds signal and actions

        - `tag` (`canto_curses.tag.Tag`): a feed tag
        - `stories` (list of `canto_curses.story.Story`): a list
           of stories from `tag`
        '''

        for s in stories:
            try:
                # filter read stories
                if not self._get_read_state(s):
                    noti = self.factory.send_storynotif(tag, s)

                    # signals and actions
                    noti.on_close(self.on_close)
                    noti.add_action("read", "Read", self.on_read, s)
                    noti.add_action("mark-as-read", "Mark as read",
                                    self.on_mark_as_read, s)

                    ret = False
                    try: ret = noti.show()
                    except dbus.exceptions.DBusException as e:
                        log.debug(traceback.format_exc())
                        log.warning(
                            "[exception] on notification.show(): %s" %
                            str(e)
                        )

                    if ret: self.notifications.append(noti)

            except KeyError as e:
                # solve a bug #1, on startup stories are empty
                log.debug(traceback.format_exc())
                log.debug(
                    "[exception] `on_stories_added`: %s" %
                    str(e)
                )

    def on_attributes(self, attributes):
        """
        Callback for `curses_attributes`. When state is changed, if a
        story is read, close it's notification bubble
        """

        for noti in [s for s in self.notifications
                       if s.story.id in attributes]:
            if self._get_read_state(noti.story): noti.close()

    def on_close(self, gtknot):
        log.debug('[close] notification [%d]' % gtknot.id)

        gtknot.close()
        self.notifications.remove(gtknot)

    def on_read(self, notification, action, story):
        if action != "read": return

        log.debug(
            "[read] notification [%d]: story %s" %
            (notification.id, notification.storytitle, )
        )

        # open link in browser
        open_in_tab = 2
        webbrowser.open(notification.link, new=open_in_tab)

        self.on_mark_as_read(notification, 'mark-as-read', story)

    def on_mark_as_read(self, notification, action, story):
        if action != "mark-as-read": return

        log.debug(
            "[mark as read] notification [%d]: story %s" %
            (notification.id, notification.storytitle, )
        )

        # change element state
        state = "read"
        attributes = {}
        if story.handle_state(state):
            attributes[story.id] = {
                "canto-state": story.content["canto-state"]
            }

        if attributes: tag_updater.set_attributes(attributes)

    def on_curses_exit(self, *args):
        log.debug("closing all notifications")

        # copy list before iterating because close() modifies the list
        for n in list(self.notifications): n.close()
        self.mainloop.quit()


manager = StoryManager()
"`StoryManager` that sends notifications on new stories"