/usr/share/doc/python-telepathy/examples/roster.py is in python-telepathy 0.15.19-2.1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | import dbus
import dbus.glib
import gobject
import sys
from account import connection_from_file
from telepathy.client.channel import Channel
from telepathy.constants import (
CONNECTION_HANDLE_TYPE_CONTACT, CONNECTION_HANDLE_TYPE_LIST,
CONNECTION_STATUS_CONNECTED, CONNECTION_STATUS_DISCONNECTED)
from telepathy.errors import NotAvailable
from telepathy.interfaces import (
CHANNEL_INTERFACE_GROUP, CHANNEL_TYPE_CONTACT_LIST, CONN_INTERFACE)
def print_members(conn, chan):
current, local_pending, remote_pending = (
chan[CHANNEL_INTERFACE_GROUP].GetAllMembers())
for member in current:
print ' - %s' % (
conn[CONN_INTERFACE].InspectHandles(
CONNECTION_HANDLE_TYPE_CONTACT, [member])[0])
if not current:
print ' (none)'
class RosterClient:
def __init__(self, conn):
self.conn = conn
conn[CONN_INTERFACE].connect_to_signal(
'StatusChanged', self.status_changed_cb)
def _request_list_channel(self, name):
handle = self.conn[CONN_INTERFACE].RequestHandles(
CONNECTION_HANDLE_TYPE_LIST, [name])[0]
return self.conn.request_channel(
CHANNEL_TYPE_CONTACT_LIST, CONNECTION_HANDLE_TYPE_LIST,
handle, True)
def status_changed_cb(self, state, reason):
if state == CONNECTION_STATUS_DISCONNECTED:
print 'disconnected: %s' % reason
self.quit()
return
if state != CONNECTION_STATUS_CONNECTED:
return
print 'connected'
for name in ('subscribe', 'publish', 'hide', 'allow', 'deny', 'known'):
try:
chan = self._request_list_channel(name)
except dbus.DBusException:
print "'%s' channel is not available" % name
continue
print '%s: members' % name
print_members(self.conn, chan)
chan[CHANNEL_INTERFACE_GROUP].connect_to_signal('MembersChanged',
lambda *args: self.members_changed_cb(name, *args))
print 'waiting for changes'
def members_changed_cb(self, name, message, added, removed, local_pending,
remote_pending, actor, reason):
if added:
for handle in added:
print '%s: added: %d' % (name, added)
if removed:
for handle in removed:
print '%s: removed: %d' % (name, added)
def run(self):
self.loop = gobject.MainLoop()
try:
self.loop.run()
except KeyboardInterrupt:
print 'interrupted'
def quit(self):
self.loop.quit()
if __name__ == '__main__':
assert len(sys.argv) == 2
conn = connection_from_file(sys.argv[1])
client = RosterClient(conn)
print "connecting"
conn[CONN_INTERFACE].Connect()
client.run()
print "disconnecting"
try:
conn[CONN_INTERFACE].Disconnect()
except dbus.DBusException:
pass
|