#!/usr/bin/python # SPDX-License-Identifier: LGPL-2.1-or-later from __future__ import absolute_import, print_function, unicode_literals from optparse import OptionParser import os from pprint import pformat import sys import dbus import dbus.mainloop.glib try: from gi.repository import GObject except ImportError: import gobject as GObject BUS_NAME='org.bluez.obex' PATH = '/org/bluez/obex' CLIENT_INTERFACE = 'org.bluez.obex.Client1' SESSION_INTERFACE = 'org.bluez.obex.Session1' MESSAGE_ACCESS_INTERFACE = 'org.bluez.obex.MessageAccess1' MESSAGE_INTERFACE = 'org.bluez.obex.Message1' TRANSFER_INTERFACE = 'org.bluez.obex.Transfer1' def unwrap(x): """Hack to unwrap D-Bus values, so that they're easier to read when printed. Taken from d-feet """ if isinstance(x, list): return map(unwrap, x) if isinstance(x, tuple): return tuple(map(unwrap, x)) if isinstance(x, dict): return dict([(unwrap(k), unwrap(v)) for k, v in x.items()]) for t in [unicode, str, long, int, float, bool]: if isinstance(x, t): return t(x) return x def parse_options(): parser.add_option("-d", "--device", dest="device", help="Device to connect", metavar="DEVICE") parser.add_option("-c", "--chdir", dest="new_dir", help="Change current directory to DIR", metavar="DIR") parser.add_option("-l", "--lsdir", action="store_true", dest="ls_dir", help="List folders in current directory") parser.add_option("-v", "--verbose", action="store_true", dest="verbose") parser.add_option("-L", "--lsmsg", action="store", dest="ls_msg", help="List messages in supplied CWD subdir") parser.add_option("-g", "--get", action="store", dest="get_msg", help="Get message contents") parser.add_option("-p", "--push", action="store", dest="push_msg", help="Push message") parser.add_option("--get-properties", action="store", dest="get_msg_properties", help="Get message properties") parser.add_option("--mark-read", action="store", dest="mark_msg_read", help="Marks the messages as read") parser.add_option("--mark-unread", action="store", dest="mark_msg_unread", help="Marks the messages as unread") parser.add_option("--mark-deleted", action="store", dest="mark_msg_deleted", help="Deletes the message from the folder") parser.add_option("--mark-undeleted", action="store", dest="mark_msg_undeleted", help="Undeletes the message") parser.add_option("-u", "--update-inbox", action="store_true", dest="update_inbox", help="Checks for new mails") return parser.parse_args() def set_folder(session, new_dir): session.SetFolder(new_dir) class MapClient: def __init__(self, session_path, verbose=False): self.progress = 0 self.transfer_path = None self.props = dict() self.verbose = verbose self.path = session_path bus = dbus.SessionBus() obj = bus.get_object(BUS_NAME, session_path) self.session = dbus.Interface(obj, SESSION_INTERFACE) self.map = dbus.Interface(obj, MESSAGE_ACCESS_INTERFACE) bus.add_signal_receiver(self.properties_changed, dbus_interface="org.freedesktop.DBus.Properties", signal_name="PropertiesChanged", path_keyword="path") def create_transfer_reply(self, path, properties): self.transfer_path = path self.props[path] = properties if self.verbose: print("Transfer created: %s (file %s)" % (path, properties["Filename"])) def generic_reply(self): if self.verbose: print("Operation succeeded") def error(self, err): print(err) mainloop.quit() def transfer_complete(self, path): if self.verbose: print("Transfer finished") properties = self.props.get(path) if properties == None: return f = open(properties["Filename"], "r") os.remove(properties["Filename"]) print(f.readlines()) def transfer_error(self, path): print("Transfer %s error" % path) mainloop.quit() def properties_changed(self, interface, properties, invalidated, path): req = self.props.get(path) if req == None: return if properties['Status'] == 'complete': self.transfer_complete(path) return if properties['Status'] == 'error': self.transfer_error(path) return def set_folder(self, new_dir): self.map.SetFolder(new_dir) def list_folders(self): for i in self.map.ListFolders(dict()): print("%s/" % (i["Name"])) def list_messages(self, folder): ret = self.map.ListMessages(folder, dict()) print(pformat(unwrap(ret))) def get_message(self, handle): self.map.ListMessages("", dict()) path = self.path + "/message" + handle obj = bus.get_object(BUS_NAME, path) msg = dbus.Interface(obj, MESSAGE_INTERFACE) msg.Get("", True, reply_handler=self.create_transfer_reply, error_handler=self.error) def push_message(self, filename): self.map.PushMessage(filename, "telecom/msg/outbox", dict(), reply_handler=self.create_transfer_reply, error_handler=self.error) def get_message_properties(self, handle): self.map.ListMessages("", dict()) path = self.path + "/message" + handle obj = bus.get_object(BUS_NAME, path) msg = dbus.Interface(obj, "org.freedesktop.DBus.Properties") ret = msg.GetAll(MESSAGE_INTERFACE) print(pformat(unwrap(ret))) def set_message_property(self, handle, prop, flag): self.map.ListMessages("", dict()) path = self.path + "/message" + handle obj = bus.get_object(BUS_NAME, path) msg = dbus.Interface(obj, MESSAGE_INTERFACE) msg.SetProperty (prop, flag); def update_inbox(self): self.map.UpdateInbox() if __name__ == '__main__': dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) parser = OptionParser() (options, args) = parse_options() if not options.device: parser.print_help() exit(0) bus = dbus.SessionBus() mainloop = GObject.MainLoop() client = dbus.Interface(bus.get_object(BUS_NAME, PATH), CLIENT_INTERFACE) print("Creating Session") path = client.CreateSession(options.device, { "Target": "map" }) map_client = MapClient(path, options.verbose) if options.new_dir: map_client.set_folder(options.new_dir) if options.ls_dir: map_client.list_folders() if options.ls_msg is not None: map_client.list_messages(options.ls_msg) if options.get_msg is not None: map_client.get_message(options.get_msg) if options.push_msg is not None: map_client.push_message(options.push_msg) if options.get_msg_properties is not None: map_client.get_message_properties(options.get_msg_properties) if options.mark_msg_read is not None: map_client.set_message_property(options.mark_msg_read, "Read", True) if options.mark_msg_unread is not None: map_client.set_message_property(options.mark_msg_unread, "Read", False) if options.mark_msg_deleted is not None: map_client.set_message_property(options.mark_msg_deleted, "Deleted", True) if options.mark_msg_undeleted is not None: map_client.set_message_property(options.mark_msg_undeleted, "Deleted", False) if options.update_inbox: map_client.update_inbox() mainloop.run()