# vim:fileencoding=utf-8:sw=4
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
# Copyright (c) 2009 by Paweł Tomak
# Copyright (c) 2009 by Paweł Zuzelski
import ekg
import time
import pynotify
import re
import sys
import glib
TIMEOUT_STATUS=3500
TIMEOUT_MSG=3500
def removeTextFormatting(text):
"""
Removes advance text formatting from notifications.
Converts html reserved characters (&, ", < and >) to xml entities.
Removes ANSII color strings.
"""
# Remove html tags
reg = re.compile("&")
text = reg.sub("&", text)
reg = re.compile('"')
text = reg.sub(""", text)
reg = re.compile("<")
text = reg.sub("<", text)
reg = re.compile(">")
text = reg.sub(">", text)
# Remove terminal color codes
reg = re.compile("\[[0-9]+;[0-9]+m")
text = reg.sub("", text)
reg = re.compile("\[[0-9]+m")
text = reg.sub("", text)
return text
def parseMeCommand(text, user):
"""
Interpretes /me command as defined in JEP-0245
If text begins with '/me ', substitute it with '* user '
"""
if (text[:4] == "/me "):
text = "* " + user + text[3:]
return text
def catchURL(text):
"""
Converts URLs to html "a href" tags, so they will be clickable if
notification daemon supports basic html.
"""
reg = re.compile("((news|telnet|nttp|file|http|ftp|https)://[^ ]+|www.[^ ]+)")
if len(reg.findall(text)):
text = reg.sub(r'\1', text)
return [1, text]
else:
return [0, text]
def transStatus(status):
return {
'avail': 'dostepny',
'away': 'zaraz wracam',
'blocking': 'BLOKUJE',
'error': 'BLAD STATUSU!!',
'ffc': 'chetny do rozmowy',
'chat': 'chetny do rozmowy',
'dnd': 'nie przeszkadzac',
'xa': 'bardzo zajety',
'notavail': 'niedostepny',
'unknown': 'nieznany',
}[status]
def getUrgency(title, text):
"""
Performs notify:urgency_critical_regexp and notify:urgency_normal_regexp
matching on "title&text". "&" character was chosen as separator, because
acording to RFC3920, Apendix A, use of this character is forbiden i JID,
so it should not apear in title.
If notify:urgency_critical_regexp matches, returns urgency CRITICAL. If
not, tries to match notify:urgency_normal_regexp, on success return
urgency NORMAL. Finally return urgency LOW.
Default values are "^$" wich are never matched, as string always contain
at least single "&" character.
"""
message=title+"&"+text
urgencyCritical = re.compile(ekg.config["notify:urgency_critical_regexp"])
if (urgencyCritical.match(message)):
return pynotify.URGENCY_CRITICAL
urgencyNormal = re.compile(ekg.config["notify:urgency_normal_regexp"])
if (urgencyNormal.match(message)):
return pynotify.URGENCY_NORMAL
return pynotify.URGENCY_LOW
def displayNotify(title, text, timeout, type):
"""
Sends notification to dbus org.freedesktop.Notifications service using
pynotify python library.
"""
if not pynotify.init("EkgNotif"):
ekg.echo("you don't seem to have pynotify installed")
return 0
if ekg.config["notify:catch_url"] != "0":
l = catchURL(text)
if l[0]:
text = l[1]
timeout = int(ekg.config["notify:catch_url_timeout"])
n = pynotify.Notification(title, text, type)
n.set_timeout(timeout)
n.set_urgency(getUrgency(title, text))
# Most probably glib.GError is:
# The name org.freedesktop.Notifications was not provided by any
# .service files
# Catch this exception and print information in debug window.
# Sometimes I
# do not have org.freedesktop.Notifications registered and
# I do not want
# error messages in chat window.
# Or logs buffer has overflowed ;/
try:
n.show()
except glib.GError as e:
ekg.debug("pynotif: " + str(e))
return 1
def notifyStatus(session, uid, status, descr):
"""
Display status change notifications, but first check if status change
notifications are enabled, then check if session and uids match
ignore_{sessions,uids}_regexp.
This function is bound to protocol-status handler
"""
if ekg.config["notify:status_notify"] == "0":
return 1
if (ekg.config["notify:ignore_sessions_regexp"]):
regexp = re.compile(ekg.config["notify:ignore_sessions_regexp"])
if regexp.match(session):
return 1
if (ekg.config["notify:ignore_uids_regexp"]):
regexp = re.compile(ekg.config["notify:ignore_uids_regexp"])
if regexp.match(uid):
return 1
regexp = re.compile('.*' + session + '.*')
if regexp.match(uid):
ekg.debug("Zmienil sie status sesji: %s. Nie zostal on zmieniony przez ten program. Sprawdz to, jesli nie zmieniales statusu jakims innym programem" % session)
return 1
sesja = ekg.session_get(session)
regexp = re.compile('([a-z]{2,4}:[^/]+)')
regexp = regexp.match(uid)
regexp = regexp.group()
try:
user = sesja.user_get(regexp)
except KeyError:
ekg.debug("Nie znalazlem uzytkownika %s." % uid)
user = "Empty"
status = transStatus(status)
if user == "Empty":
nick = regexp
else:
nick = user.nickname or user.uid or "Empty"
s = status or "Empty"
s = removeTextFormatting(s)
text = "" + nick + " zmienil status na " + s + ""
if descr:
descr = removeTextFormatting(descr)
text = text + ":\n" + descr + "\n"
return displayNotify(session, text, TIMEOUT_STATUS, ekg.config["notify:icon_status"])
def notifyMessage(session, uid, type, text, stime, ignore_level):
"""
Display message notifications, but first check if message notifications
are enabled, then check if session and uids match
ignore_{sessions,uids}_regexp.
This function is bound to protocol-message handler
"""
if ekg.config["notify:message_notify"] == "0":
return 1
if (ekg.config["notify:ignore_sessions_regexp"]):
regexp = re.compile(ekg.config["notify:ignore_sessions_regexp"])
if regexp.match(session):
return 1
if (ekg.config["notify:ignore_uids_regexp"]):
regexp = re.compile(ekg.config["notify:ignore_uids_regexp"])
if regexp.match(uid):
return 1
sesja = ekg.session_get(session)
try:
user = sesja.user_get(uid)
except KeyError:
ekg.debug("Nie znalazlem uzytkownika %s." % uid)
user = "Empty"
if user == None:
user = "Empty"
if user == "Empty" and ekg.config["notify:message_notify_unknown"] == "0":
return 1
if user == "Empty":
user = uid
else:
user = user.nickname
try:
title = user
except KeyError:
title = uid
if (ekg.config["notify:show_timestamps"] == "1"):
title = time.strftime("%H:%M:%S", time.localtime(stime)) + " " + title
text = removeTextFormatting(text)
text = parseMeCommand(text, user)
if len(text) > 200:
text = text[0:199] + "... >>>\n\n"
return displayNotify(title, text, TIMEOUT_MSG, ekg.config["notify:icon_msg"])
def timeCheck(name, args):
global TIMEOUT_MSG
global TIMEOUT_STATUS
rexp = re.compile('^[0-9]{4,4}')
rexp = rexp.findall(args)
if len(rexp) == 1:
if name == "notify:message_timeout":
TIMEOUT_MSG = int(rexp[0])
return 1
if name == "notify:status_timeout":
TIMEOUT_STATUS = int(rexp[0])
return 1
if name == "notify:catch_url_timeout":
return 1
if name == "notify:message_timeout":
ekg.echo("Zmienna %s bedzie pomijana do czasu, az zostanie ustawiona wartosc z zakresu od 1000ms do 9999ms. Jej obecna wartosc to: %i" % (name,TIMEOUT_MSG))
elif name == "notify:status_timeout":
ekg.echo("Zmienna %s bedzie pomijana do czasu, az zostanie ustawiona wartosc z zakresu od 1000ms do 9999ms. Jej obecna wartosc to: %i" % (name,TIMEOUT_STATUS))
elif name == "notify:catch_url_timeout":
ekg.echo("Zmienna %s bedzie pomijana do czasu, az zostanie ustawiona wartosc z zakresu od 1000ms do 9999ms. Jej obecna wartosc to: %i" % (name,TIMEOUT_STATUS))
return 0
def notifyTest(name, args):
"""
Sends test notification.
This function is bound to notify:send command
"""
args = args.split(None, 1)
if (len(args) == 0):
title="Test"
else:
title=args[0]
if (len(args) <= 1):
text="Pięćdziesiąt trzy"
else:
text = args[1]
return displayNotify(title, text, TIMEOUT_MSG, ekg.config["notify:icon_msg"])
ekg.handler_bind('protocol-status', notifyStatus)
ekg.handler_bind("protocol-message-received", notifyMessage)
ekg.variable_add("notify:ignore_sessions_regexp", "^irc:")
ekg.variable_add("notify:ignore_uids_regexp", "^xmpp:.*@conference\.")
ekg.variable_add("notify:urgency_critical_regexp", "^$")
ekg.variable_add("notify:urgency_normal_regexp", "^$")
ekg.variable_add("notify:icon_status", "dialog-warning")
ekg.variable_add("notify:icon_msg", "dialog-warning")
ekg.variable_add("notify:message_timeout", "3500", timeCheck)
ekg.variable_add("notify:message_notify", "1")
ekg.variable_add("notify:message_notify_unknown", "1")
ekg.variable_add("notify:show_timestamps", "1")
ekg.variable_add("notify:status_timeout", "3500", timeCheck)
ekg.variable_add("notify:status_notify", "1")
ekg.variable_add("notify:catch_url", "1")
ekg.variable_add("notify:catch_url_timeout", "5000", timeCheck)
ekg.command_bind("notify:send", notifyTest)
if int(ekg.config["notify:message_timeout"]) < 1000 or int(ekg.config["notify:message_timeout"]) > 9999:
timeCheck("notify:message_timeout", ekg.config["notify:message_timeout"])
if int(ekg.config["notify:status_timeout"]) < 1000 or int(ekg.config["notify:status_timeout"]) > 9999:
timeCheck("notify:status_timeout", ekg.config["notify:status_timeout"])