Skip to content

Instantly share code, notes, and snippets.

@pblocz

pblocz/canto-curses-notify-plugin.py

Last active June 18, 2016 13:31
Show Gist options
  • Save pblocz/91ae284455e3227b1dbf to your computer and use it in GitHub Desktop.
Save pblocz/91ae284455e3227b1dbf to your computer and use it in GitHub Desktop.
Download ZIP
canto-curses notification plugin
Raw
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"
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment