From e9737ee8ea92d1e3512c743fae6f41a096e908de Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 11:50:25 +0300 Subject: [PATCH 01/12] Transition work to bring it to work with cmd2 (latest version) --- pbapclient.py | 309 ++++++++++++++++++++++++++++++++++------------- pbapcommon.py | 98 +++++++++------ pbapheaders.py | 15 ++- pbapresponses.py | 12 +- pbapserver.py | 199 +++++++++++++++++++++--------- setup.py | 9 +- vcard_helper.py | 71 +++++++---- vfolder.py | 25 +++- 8 files changed, 512 insertions(+), 226 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 1f124e4..23fe03a 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -32,52 +32,82 @@ def __init__(self, address, port): client.Client.__init__(self, address, port) self.current_dir = "/" - def pull_phonebook(self, name, filter_=0, format_=0, max_list_count=65535, list_startoffset=0): + def pull_phonebook( + self, name, filter_=0, format_=0, max_list_count=65535, list_startoffset=0 + ): """Retrieves entire phonebook object from current folder""" - logger.info("Requesting pull_phonebook for pbobject '%s' with appl parameters %s", name, str(locals())) - data = {"Filter": headers.Filter(filter_), - "Format": headers.Format(format_), - "MaxListCount": headers.MaxListCount(max_list_count), - "ListStartOffset": headers.ListStartOffset(list_startoffset)} + logger.info( + "Requesting pull_phonebook for pbobject '%s' with appl parameters %s", + name, + str(locals()), + ) + data = { + "Filter": headers.Filter(filter_), + "Format": headers.Format(format_), + "MaxListCount": headers.MaxListCount(max_list_count), + "ListStartOffset": headers.ListStartOffset(list_startoffset), + } application_parameters = headers.App_Parameters(data, encoded=False) header_list = [headers.PBAPType("x-bt/phonebook")] if application_parameters.data: header_list.append(application_parameters) response = self.get(name, header_list) - if not isinstance(response, tuple) and isinstance(response, responses.FailureResponse): - logger.error("pull_phonebook failed for pbobject '%s'. reason = %s", name, response) + if not isinstance(response, tuple) and isinstance( + response, responses.FailureResponse + ): + logger.error( + "pull_phonebook failed for pbobject '%s'. reason = %s", name, response + ) return return response - def pull_vcard_listing(self, name, order=0, search_value=None, - search_attribute="N", max_list_count=65535, list_startoffset=0): + def pull_vcard_listing( + self, + name, + order=0, + search_value=None, + search_attribute="N", + max_list_count=65535, + list_startoffset=0, + ): """Retrieves phonebook listing object from current folder""" logger.info("Requesting pull_vcard_listing with parameters %s", str(locals())) - data = {"Order": headers.Order(order), - "MaxListCount": headers.MaxListCount(max_list_count), - "ListStartOffset": headers.ListStartOffset(list_startoffset)} + data = { + "Order": headers.Order(order), + "MaxListCount": headers.MaxListCount(max_list_count), + "ListStartOffset": headers.ListStartOffset(list_startoffset), + } if search_value is not None and search_attribute is not None: - data.update({ - "SearchValue": headers.SearchValue(search_value), - "SearchAttribute": headers.SearchAttribute(search_attribute), - }) + data.update( + { + "SearchValue": headers.SearchValue(search_value), + "SearchAttribute": headers.SearchAttribute(search_attribute), + } + ) application_parameters = headers.App_Parameters(data, encoded=False) header_list = [headers.PBAPType("x-bt/vcard-listing")] if application_parameters.data: header_list.append(application_parameters) response = self.get(name, header_list) - if not isinstance(response, tuple) and isinstance(response, responses.FailureResponse): - logger.error("pull_vcard_listing failed for pbobject '%s'. reason = %s", name, response) + if not isinstance(response, tuple) and isinstance( + response, responses.FailureResponse + ): + logger.error( + "pull_vcard_listing failed for pbobject '%s'. reason = %s", + name, + response, + ) return return response def pull_vcard_entry(self, name, filter_=0, format_=0): """Retrieves specific vcard from pbap server""" - logger.info("Requesting pull_vcard_entry for pbobject with parameters %s", str(locals())) - data = {"Filter": headers.Filter(filter_), - "Format": headers.Format(format_)} + logger.info( + "Requesting pull_vcard_entry for pbobject with parameters %s", str(locals()) + ) + data = {"Filter": headers.Filter(filter_), "Format": headers.Format(format_)} application_parameters = headers.App_Parameters(data, encoded=False) header_list = [headers.PBAPType("x-bt/vcard")] @@ -85,8 +115,12 @@ def pull_vcard_entry(self, name, filter_=0, format_=0): header_list.append(application_parameters) response = self.get(name, header_list) - if not isinstance(response, tuple) and isinstance(response, responses.FailureResponse): - logger.error("pull_vcard_entry failed for pbobject '%s'. reason = %s", name, response) + if not isinstance(response, tuple) and isinstance( + response, responses.FailureResponse + ): + logger.error( + "pull_vcard_entry failed for pbobject '%s'. reason = %s", name, response + ) return return response @@ -94,8 +128,10 @@ def set_phonebook(self, name="", to_root=False, to_parent=False): """Sets the current folder in the virtual folder architecture""" logger.info("Setting current folder with params '%s'", str(locals())) if name == "" and not to_parent and not to_root: - logger.error("Not a valid action, " - "either name should be not empty or to_parent/to_root should be True") + logger.error( + "Not a valid action, " + "either name should be not empty or to_parent/to_root should be True" + ) return # TODO: not exactly as per spec, limited by pyobex setpath. need to refine further if to_root: @@ -113,7 +149,9 @@ def set_phonebook(self, name="", to_root=False, to_parent=False): else: response = self.setpath(name) - if not isinstance(response, tuple) and isinstance(response, responses.FailureResponse): + if not isinstance(response, tuple) and isinstance( + response, responses.FailureResponse + ): logger.error("set_phonebook failed. reason = %s", name, response) return @@ -131,25 +169,31 @@ class REPL(cmd2.Cmd): def __init__(self): cmd2.Cmd.__init__(self) - self.prompt = self.colorize("pbap> ", "yellow") - self.intro = self.colorize("Welcome to the PhoneBook Access Profile!", "green") + # self.prompt = self.colorize("pbap> ", "yellow") + self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.YELLOW) + # self.intro = self.colorize("Welcome to the PhoneBook Access Profile!", "green") + self.intro = cmd2.ansi.style( + "Welcome to the PhoneBook Access Profile!", fg=cmd2.ansi.Fg.GREEN + ) self.client = None self._store_history() - cmd2.set_use_arg_list(False) + # cmd2.set_use_arg_list(False) @staticmethod def _store_history(): - history_file = os.path.expanduser('~/.pbapclient_history') + history_file = os.path.expanduser("~/.pbapclient_history") if not os.path.exists(history_file): - with open(history_file, "w") as fobj: + with open(history_file, "w", encoding="latin1") as fobj: fobj.write("") readline.read_history_file(history_file) atexit.register(readline.write_history_file, history_file) - @cmd2.options([], arg_desc="server_address") + # @cmd2.options([], arg_desc="server_address") def do_connect(self, line, opts): + self.add_settable(cmd2.Settable(name="server_address", val_type=str)) + profile_id = "1130" # profile id of PBAP - service_id = b'\x79\x61\x35\xf0\xf0\xc5\x11\xd8\x09\x66\x08\x00\x20\x0c\x9a\x66' + service_id = b"\x79\x61\x35\xf0\xf0\xc5\x11\xd8\x09\x66\x08\x00\x20\x0c\x9a\x66" server_address = line if not server_address: raise ValueError("server_address should not be empty") @@ -170,80 +214,154 @@ def do_connect(self, line, opts): logger.error("Connect Failed, Terminating the Pbap client..") sys.exit(2) logger.info("Connect success") - self.prompt = self.colorize("pbap> ", "green") + # self.prompt = self.colorize("pbap> ", "green") + self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.GREEN) - @cmd2.options([], arg_desc="") + # @cmd2.options([], arg_desc="") def do_disconnect(self, line, opts): if self.client is None: - logger.error("PBAPClient is not even connected.. Connect and then try disconnect") + logger.error( + "PBAPClient is not even connected.. Connect and then try disconnect" + ) sys.exit(2) logger.debug("Disconnecting pbap client with pbap server") self.client.disconnect() self.client = None - self.prompt = self.colorize("pbap> ", "yellow") - - @cmd2.options([make_option('-f', '--filter', default=0x00000000, type=int, help="Attributes filter mask"), - make_option('-t', '--format', default=0, type=int, help="vcard format"), - make_option('-c', '--max-count', default=65535, type=int, - help="maximum number of contacts to be returned"), - make_option('-o', '--start-offset', default=0, type=int, - help="offset of first entry to be returned"), - ], - arg_desc="phonebook_name") + # self.prompt = self.colorize("pbap> ", "yellow") + self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.YELLOW) + + # @cmd2.options( + # [ + # make_option( + # "-f", + # "--filter", + # default=0x00000000, + # type=int, + # help="Attributes filter mask", + # ), + # make_option("-t", "--format", default=0, type=int, help="vcard format"), + # make_option( + # "-c", + # "--max-count", + # default=65535, + # type=int, + # help="maximum number of contacts to be returned", + # ), + # make_option( + # "-o", + # "--start-offset", + # default=0, + # type=int, + # help="offset of first entry to be returned", + # ), + # ], + # arg_desc="phonebook_name", + # ) def do_pull_phonebook(self, line, opts): """Returns phonebook as per requested options""" - result = self.client.pull_phonebook(name=line, filter_=opts.filter, format_=opts.format, - max_list_count=opts.max_count, list_startoffset=opts.start_offset) + result = self.client.pull_phonebook( + name=line, + filter_=opts.filter, + format_=opts.format, + max_list_count=opts.max_count, + list_startoffset=opts.start_offset, + ) if result is not None: - header, data = result + _, data = result logger.info("Result of pull_phonebook:\n%s", data) - @cmd2.options([make_option('-r', '--order', default=0, type=int, - help="Ordering { Alphabetical | Indexed | Phonetical}"), - make_option('--search-attribute', default=0, type=int, - help="SearchAttribute {Name | Number | Sound }"), - make_option('--search-value', default=None, help="SearchValue {}"), - make_option('-c', '--max-count', default=65535, type=int, - help="maximum number of contacts to be returned"), - make_option('-o', '--start-offset', default=0, type=int, - help="offset of first entry to be returned"), - ], - arg_desc="vcard_folder") + pull_vcard_listing_parser = cmd2.Cmd2ArgumentParser() + pull_vcard_listing_parser.add_argument( + "-r", + "--order", + default=0, + type=int, + help="Ordering { Alphabetical | Indexed | Phonetical}", + ) + pull_vcard_listing_parser.add_argument( + "--search-attribute", + default=0, + type=int, + help="SearchAttribute {Name | Number | Sound }", + ) + pull_vcard_listing_parser.add_argument( + "-c", + "--max-count", + default=65535, + type=int, + help="Maximum number of contacts to be returned", + ) + pull_vcard_listing_parser.add_argument( + "-o", + "--start-offset", + default=0, + type=int, + help="offset of first entry to be returned", + ) + + @cmd2.with_argparser(pull_vcard_listing_parser) def do_pull_vcard_listing(self, line, opts): """Returns vcardlisting as per requested options""" - result = self.client.pull_vcard_listing(name=line, order=opts.order, - search_value=opts.search_value, - search_attribute=opts.search_attribute, - max_list_count=opts.max_count, - list_startoffset=opts.start_offset) + result = self.client.pull_vcard_listing( + name=line, + order=opts.order, + search_value=opts.search_value, + search_attribute=opts.search_attribute, + max_list_count=opts.max_count, + list_startoffset=opts.start_offset, + ) if result is not None: header, data = result logger.info("Result of pull_vcard_listing:\n%s", data) - @cmd2.options([make_option('-f', '--filter', default=0x00000000, type=int, help="Attributes filter mask"), - make_option('-t', '--format', default=0, type=int, help="vcard format") - ], - arg_desc="vcard_handle") + # @cmd2.options( + # [ + # make_option( + # "-f", + # "--filter", + # default=0x00000000, + # type=int, + # help="Attributes filter mask", + # ), + # make_option("-t", "--format", default=0, type=int, help="vcard format"), + # ], + # arg_desc="vcard_handle", + # ) def do_pull_vcard_entry(self, line, opts): """Returns a single vcardentry as per requested options""" - result = self.client.pull_vcard_entry(name=line, filter_=opts.filter, format_=opts.format) + result = self.client.pull_vcard_entry( + name=line, filter_=opts.filter, format_=opts.format + ) if result is not None: header, data = result logger.info("Result of pull_vcard_entry:\n%s", data) - @cmd2.options([make_option('--to-parent', action="store_true", default=False, - help="navigate to parent dir"), - make_option('--to-root', action="store_true", default=False, - help="navigate to root dir") - ], - arg_desc="[folder_name]") + # @cmd2.options( + # [ + # make_option( + # "--to-parent", + # action="store_true", + # default=False, + # help="navigate to parent dir", + # ), + # make_option( + # "--to-root", + # action="store_true", + # default=False, + # help="navigate to root dir", + # ), + # ], + # arg_desc="[folder_name]", + # ) def do_set_phonebook(self, line, opts): """Set current folder path of pbapserver virtual folder""" - result = self.client.set_phonebook(name=line, to_parent=opts.to_parent, to_root=opts.to_root) + result = self.client.set_phonebook( + name=line, to_parent=opts.to_parent, to_root=opts.to_root + ) if result is not None: logger.info("Result of set_phonebook:\n%s", result) - @cmd2.options([], arg_desc="server_address [folder_name]") + # @cmd2.options([], arg_desc="server_address [folder_name]") def do_mirror_vfolder(self, line, opts): """Downloads phonebook from pbapserver and save it in virtual folder architecture in FS""" args = line.split() @@ -262,9 +380,14 @@ def do_mirror_vfolder(self, line, opts): os.makedirs(current_dir) # Access the list of vcards in the phone's internal phone book. response = self.client.pull_vcard_listing( - "{prefix}telecom/{pbobject}".format(prefix=prefix, pbobject=pbobject)) + "{prefix}telecom/{pbobject}".format( + prefix=prefix, pbobject=pbobject + ) + ) if response is None: - logger.error("vcard-listing get is failed for pbobject '%s'", pbobject) + logger.error( + "vcard-listing get is failed for pbobject '%s'", pbobject + ) continue hdrs, cards = response # Parse the XML response to the previous request. @@ -280,7 +403,11 @@ def do_mirror_vfolder(self, line, opts): logger.info("\nCards in %stelecom/%s\n", prefix, pbobject) # Request all the file names obtained earlier. - self.client.set_phonebook("{prefix}telecom/{pbobject}".format(prefix=prefix, pbobject=pbobject)) + self.client.set_phonebook( + "{prefix}telecom/{pbobject}".format( + prefix=prefix, pbobject=pbobject + ) + ) for name in names: response = self.client.pull_vcard_entry(name) if response is None: @@ -303,15 +430,23 @@ def do_mirror_vfolder(self, line, opts): current_dir = os.path.normpath(os.path.join(current_dir, "..")) logger.debug("current_dir = %s", current_dir) - logger.info("\nThe phonebook in %s/telecom/%s as one vcard\n", prefix, pbobject) + logger.info( + "\nThe phonebook in %s/telecom/%s as one vcard\n", prefix, pbobject + ) response = self.client.pull_phonebook( - "{prefix}telecom/{pbobject}.vcf".format(prefix=prefix, pbobject=pbobject)) + "{prefix}telecom/{pbobject}.vcf".format( + prefix=prefix, pbobject=pbobject + ) + ) if response is None: logger.error("phonebook get is failed for pbobject '%s'", pbobject) continue hdrs, phonebook = response logger.info(phonebook) - with open(os.path.join(current_dir, prefix, "telecom", pbobject + ".vcf"), "w+") as f: + with open( + os.path.join(current_dir, prefix, "telecom", pbobject + ".vcf"), + "w+", + ) as f: f.write(phonebook) logger.info(hdrs) @@ -321,6 +456,8 @@ def do_mirror_vfolder(self, line, opts): if __name__ == "__main__": - logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(name)s %(levelname)-8s %(message)s') + logging.basicConfig( + level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)-8s %(message)s" + ) repl = REPL() repl.cmdloop() diff --git a/pbapcommon.py b/pbapcommon.py index 54cb8ca..a95fa1a 100644 --- a/pbapcommon.py +++ b/pbapcommon.py @@ -4,35 +4,35 @@ """Common tools and attributes for pbap client and server""" FILTER_ATTR_DICT = { - 0: ('VERSION', 'vCard Version'), - 1: ('FN', 'Formatted Name'), - 2: ('N', 'Structured Presentation of Name'), - 3: ('PHOTO', 'Associated Image or Photo'), - 4: ('BDAY', 'Birthday'), - 5: ('ADR', 'Delivery Address'), - 6: ('LABEL', 'Delivery'), - 7: ('TEL', 'Telephone Number'), - 8: ('EMAIL', 'Electronic Mail Address'), - 9: ('MAILER', 'Electronic Mail'), - 10: ('TZ', 'Time Zone'), - 11: ('GEO', 'Geographic Position'), - 12: ('TITLE', 'Job'), - 13: ('ROLE', 'Role within the Organization'), - 14: ('LOGO', 'Organization Logo'), - 15: ('AGENT', 'vCard of Person Representing'), - 16: ('ORG', 'Name of Organization'), - 17: ('NOTE', 'Comments'), - 18: ('REV', 'Revision'), - 19: ('SOUND', 'Pronunciation of Name'), - 20: ('URL', 'Uniform Resource Locator'), - 21: ('UID', 'Unique ID'), - 22: ('KEY', 'Public Encryption Key'), - 23: ('NICKNAME', 'Nickname'), - 24: ('CATEGORIES', 'Categories'), - 25: ('PROID', 'Product ID'), - 26: ('CLASS', 'Class information'), - 27: ('SORT', 'STRING String used for sorting operations'), - 28: ('X-IRMC-CALL-DATETIME', 'Time stamp'), + 0: ("VERSION", "vCard Version"), + 1: ("FN", "Formatted Name"), + 2: ("N", "Structured Presentation of Name"), + 3: ("PHOTO", "Associated Image or Photo"), + 4: ("BDAY", "Birthday"), + 5: ("ADR", "Delivery Address"), + 6: ("LABEL", "Delivery"), + 7: ("TEL", "Telephone Number"), + 8: ("EMAIL", "Electronic Mail Address"), + 9: ("MAILER", "Electronic Mail"), + 10: ("TZ", "Time Zone"), + 11: ("GEO", "Geographic Position"), + 12: ("TITLE", "Job"), + 13: ("ROLE", "Role within the Organization"), + 14: ("LOGO", "Organization Logo"), + 15: ("AGENT", "vCard of Person Representing"), + 16: ("ORG", "Name of Organization"), + 17: ("NOTE", "Comments"), + 18: ("REV", "Revision"), + 19: ("SOUND", "Pronunciation of Name"), + 20: ("URL", "Uniform Resource Locator"), + 21: ("UID", "Unique ID"), + 22: ("KEY", "Public Encryption Key"), + 23: ("NICKNAME", "Nickname"), + 24: ("CATEGORIES", "Categories"), + 25: ("PROID", "Product ID"), + 26: ("CLASS", "Class information"), + 27: ("SORT", "STRING String used for sorting operations"), + 28: ("X-IRMC-CALL-DATETIME", "Time stamp"), # 29: ('RESERVED_FUTURE_USE', 'Reserved for future use'), # 30: ('RESERVED_FUTURE_USE', 'Reserved for future use'), # 31: ('RESERVED_FUTURE_USE', 'Reserved for future use'), @@ -70,15 +70,39 @@ # 63: ('RESERVED_PROPRIETARY_FILTER', 'Reserved for proprietary filter usage') } -POSSIBLE_VCARD_ATTR = ["VERSION", "FN", "N", "PHOTO", "BDAY", "ADR", "LABEL", "TEL", - "EMAIL", "MAILER", "TZ", "GEO", "TITLE", "ROLE", "LOGO", - "AGENT", "ORG", "NOTE", "REV", "SOUND", "URL", "UID", "KEY", - "NICKNAME", "CATEGORIES", "PROID", "CLASS", "SORT", "X-IRMC-CALL-DATETIME"] +POSSIBLE_VCARD_ATTR = [ + "VERSION", + "FN", + "N", + "PHOTO", + "BDAY", + "ADR", + "LABEL", + "TEL", + "EMAIL", + "MAILER", + "TZ", + "GEO", + "TITLE", + "ROLE", + "LOGO", + "AGENT", + "ORG", + "NOTE", + "REV", + "SOUND", + "URL", + "UID", + "KEY", + "NICKNAME", + "CATEGORIES", + "PROID", + "CLASS", + "SORT", + "X-IRMC-CALL-DATETIME", +] # vcard_format: bitmask for mandatory attributes # vcard 2.1 are VERSION ,N and TEL # vcard 3.0 are VERSION, N, FN and TEL -MANDATORY_ATTR_BITMASK = { - "2.1": int("10000101", 2), - "3.0": int("10000111", 2) -} +MANDATORY_ATTR_BITMASK = {"2.1": int("10000101", 2), "3.0": int("10000111", 2)} diff --git a/pbapheaders.py b/pbapheaders.py index 52227b7..88ceee9 100644 --- a/pbapheaders.py +++ b/pbapheaders.py @@ -64,7 +64,9 @@ class VariableLengthProperty(AppParamProperty): fmt = "{len}s" def encode(self, data): - return super(VariableLengthProperty, self).encode(struct.pack(self.fmt.format(len=len(data)), data)) + return super(VariableLengthProperty, self).encode( + struct.pack(self.fmt.format(len=len(data)), data) + ) def decode(self): headers, data = super(VariableLengthProperty, self).decode() @@ -122,7 +124,7 @@ class NewMissedCalls(OneByteProperty): 0x06: Filter, 0x07: Format, 0x08: PhonebookSize, - 0x09: NewMissedCalls + 0x09: NewMissedCalls, } @@ -130,6 +132,7 @@ class NewMissedCalls(OneByteProperty): # code | length | data # 4c | 00 18 | 06 08 00 00 00 3f d0 00 00 80 07 01 00 04 02 00 00 05 02 00 00 + def extended_decode(self): # assumption: # size of tagid = 1 byte @@ -140,13 +143,15 @@ def extended_decode(self): tagid = ord(data[0]) length = ord(data[1]) app_param_class = app_parameters_dict[tagid] - res_dict[app_param_class.__name__] = app_param_class(data[:length + 2], encoded=True) - data = data[length + 2:] + res_dict[app_param_class.__name__] = app_param_class( + data[: length + 2], encoded=True + ) + data = data[length + 2 :] return res_dict def extended_encode(self, data_dict): - data = b'' + data = b"" for item in data_dict.values(): if item is None: continue diff --git a/pbapresponses.py b/pbapresponses.py index 130b98d..10585ba 100644 --- a/pbapresponses.py +++ b/pbapresponses.py @@ -19,8 +19,10 @@ class Service_Unavailable(FailureResponse): code = OBEX_Service_Unavailable = 0xD3 -ResponseHandler.message_dict.update({ - Not_Acceptable.code: Not_Acceptable, - Not_Implemented.code: Not_Implemented, - Service_Unavailable.code: Service_Unavailable -}) +ResponseHandler.message_dict.update( + { + Not_Acceptable.code: Not_Acceptable, + Not_Implemented.code: Not_Implemented, + Service_Unavailable.code: Service_Unavailable, + } +) diff --git a/pbapserver.py b/pbapserver.py index 87c8ac7..2dbcadc 100644 --- a/pbapserver.py +++ b/pbapserver.py @@ -23,10 +23,11 @@ class PbapServer(server.Server): - def __init__(self, address, rootdir="/", use_fs=True): server.Server.__init__(self, address) - self.vfolder = VFolderPhoneBook_FS(rootdir) if use_fs else VFolderPhoneBook_DB(rootdir) + self.vfolder = ( + VFolderPhoneBook_FS(rootdir) if use_fs else VFolderPhoneBook_DB(rootdir) + ) def process_request(self, connection, request): """Processes the request from the connection.""" @@ -63,7 +64,9 @@ def setpath(self, socket, request): # This is just a overloaded version of obex setpath if toparent: if self.vfolder.curdir == self.vfolder.rootdir: - logger.error("Current directory is same as Root dir, so can't go to parent") + logger.error( + "Current directory is same as Root dir, so can't go to parent" + ) self.send_response(socket, responses.Forbidden()) return else: @@ -111,7 +114,9 @@ def get(self, socket, request): elif decoded_header["Type"] == "x-bt/phonebook": self._pull_phonebook(socket, request, decoded_header) else: - logger.error("Requested type = %s is not supported yet.", decoded_header["Type"]) + logger.error( + "Requested type = %s is not supported yet.", decoded_header["Type"] + ) self.send_response(socket, responses.Bad_Request()) def _pull_vcard_listing(self, socket, request, decoded_header): @@ -124,7 +129,9 @@ def _pull_vcard_listing(self, socket, request, decoded_header): self.send_response(socket, responses.Not_Found()) else: phonebook_size = self.vfolder.count(abs_name) - search_query = self._get_search_query(app_params["SearchAttribute"], app_params["SearchValue"]) + search_query = self._get_search_query( + app_params["SearchAttribute"], app_params["SearchValue"] + ) # TODO: sorting based on order not works, since handle is not part of db record. it is not proper # make the "handle" as part of db record sort_key = self._get_sort_key(app_params["Order"]) @@ -136,36 +143,50 @@ def _pull_vcard_listing(self, socket, request, decoded_header): return data = "" - res_vcard_list = self._limit_phonebook(vcard_list, app_params["MaxListCount"], - app_params["ListStartOffset"]) - res_vcard_list_range = range(app_params["ListStartOffset"], - min((app_params["ListStartOffset"] + app_params["MaxListCount"]), 65535)) + res_vcard_list = self._limit_phonebook( + vcard_list, app_params["MaxListCount"], app_params["ListStartOffset"] + ) + res_vcard_list_range = range( + app_params["ListStartOffset"], + min( + (app_params["ListStartOffset"] + app_params["MaxListCount"]), 65535 + ), + ) # "NewMissedCalls": This application parameter shall be used in the response when and only when the # phone book object is mch. It indicates the number of missed calls that have been # received on the PSE since the last PullPhoneBook request on the mch folder, at the # point of the request. if "mch" in abs_name: - response_dict = {'NewMissedCalls': headers.NewMissedCalls(phonebook_size - mch_size)} + response_dict = { + "NewMissedCalls": headers.NewMissedCalls(phonebook_size - mch_size) + } mch_size = phonebook_size else: response_dict = {} - vcard_listing_object_tmpl = ('\r\n' - '\r\n' - '\r\n' - '{cards}' - '\r\n') + vcard_listing_object_tmpl = ( + '\r\n' + '\r\n' + '\r\n' + "{cards}" + "\r\n" + ) card_tag_tmpl = '\r\n' cards = "" # TODO: As per spec the handles should be hex?? for index, vcard in zip(res_vcard_list_range, res_vcard_list): - cards += card_tag_tmpl.format(handle="{}.vcf".format(index), - name=self._get_param_values(vcard, "N")) + cards += card_tag_tmpl.format( + handle="{}.vcf".format(index), + name=self._get_param_values(vcard, "N"), + ) data = vcard_listing_object_tmpl.format(cards=cards) logger.debug("Sending response success with following data") logger.debug("vcard-listing data: \r\n%s", data) - self.send_response(socket, responses.Success(), - [headers.End_Of_Body(data), headers.App_Parameters(response_dict)]) + self.send_response( + socket, + responses.Success(), + [headers.End_Of_Body(data), headers.App_Parameters(response_dict)], + ) def _get_param_values(self, vcard, param_name): for param in vcard["vcard"]: @@ -180,14 +201,13 @@ def _pull_vcard_entry(self, socket, request, decoded_header): logger.error("Requested vcard file doesn't exists") self.send_response(socket, responses.Not_Found()) else: - filtered_data = self._filter_attributes(app_params["Filter"], - self.vfolder.read(abs_name), - app_params["Format"]) + filtered_data = self._filter_attributes( + app_params["Filter"], self.vfolder.read(abs_name), app_params["Format"] + ) data = VCard(filtered_data, parsed=True).serialize(app_params["Format"]) logger.debug("Sending response success with following data") logger.debug("vcard data: \r\n%s", data) - self.send_response(socket, responses.Success(), [ - headers.End_Of_Body(data)]) + self.send_response(socket, responses.Success(), [headers.End_Of_Body(data)]) def _pull_phonebook(self, socket, request, decoded_header): mch_size = 0 # mch_size for phonebook folder (Don't optimize) @@ -206,21 +226,28 @@ def _pull_phonebook(self, socket, request, decoded_header): return data = "" - res_vcard_list = self._limit_phonebook(vcard_list, app_params["MaxListCount"], - app_params["ListStartOffset"]) + res_vcard_list = self._limit_phonebook( + vcard_list, app_params["MaxListCount"], app_params["ListStartOffset"] + ) # "NewMissedCalls": This application parameter shall be used in the response when and only when the # phone book object is mch. It indicates the number of missed calls that have been # received on the PSE since the last PullPhoneBook request on the mch folder, at the # point of the request. if "mch" in abs_name: - response_dict = {'NewMissedCalls': headers.NewMissedCalls(phonebook_size - mch_size)} + response_dict = { + "NewMissedCalls": headers.NewMissedCalls(phonebook_size - mch_size) + } mch_size = phonebook_size else: response_dict = {} for item in res_vcard_list: - filtered_data = self._filter_attributes(app_params["Filter"], item, app_params["Format"]) - data += VCard(filtered_data, parsed=True).serialize(app_params["Format"]) + filtered_data = self._filter_attributes( + app_params["Filter"], item, app_params["Format"] + ) + data += VCard(filtered_data, parsed=True).serialize( + app_params["Format"] + ) logger.debug("Sending response success with following data") logger.debug("phonebook data: \r\n%s", data) @@ -238,8 +265,13 @@ def _pull_phonebook(self, socket, request, decoded_header): data_last_chunk = data else: while bytes_transferred < datasize: - data_chunk = data[bytes_transferred: (bytes_transferred + max_datalen)] - header_list = [headers.App_Parameters(response_dict), headers.Body(data_chunk)] + data_chunk = data[ + bytes_transferred : (bytes_transferred + max_datalen) + ] + header_list = [ + headers.App_Parameters(response_dict), + headers.Body(data_chunk), + ] # 'continue' response and process the subsequent requests self.send_response(socket, responses.Continue(), header_list) while True: @@ -252,7 +284,10 @@ def _pull_phonebook(self, socket, request, decoded_header): bytes_transferred += max_datalen data_last_chunk = "" - header_list = [headers.App_Parameters(response_dict), headers.End_Of_Body(data_last_chunk)] + header_list = [ + headers.App_Parameters(response_dict), + headers.End_Of_Body(data_last_chunk), + ] self.send_response(socket, responses.Success(), header_list) def _get_search_query(self, searchattribute, searchvalue): @@ -265,7 +300,14 @@ def _get_search_query(self, searchattribute, searchvalue): else: logger.error("Unsupported value for SearchAttribute=%s", searchattribute) return {} - query = {"vcard": {"$elemMatch": {'type': searchattribute, 'values': {"$in": [searchvalue]}}}} + query = { + "vcard": { + "$elemMatch": { + "type": searchattribute, + "values": {"$in": [searchvalue]}, + } + } + } return query if searchvalue else {} def _get_sort_key(self, order): @@ -282,6 +324,7 @@ def _key_func(item): for param in item["vcard"]: if param["type"] == sort_key[0]: return ";".join(param["values"]) + return sorted(vcard_list, key=_key_func) def _respond_phonebook_size(self, socket, phonebook_size): @@ -290,11 +333,15 @@ def _respond_phonebook_size(self, socket, phonebook_size): # When MaxListCount = 0, the PSE shall ignore all other application parameters that may # be present in the request. The response shall include the PhonebookSize application # parameter (see Section 5.1.4.5). The response shall not contain any Body header - logger.debug("MaxListCount is 0, so responding with PhonebookSize = {}".format( - phonebook_size)) - response_dict = {'PhonebookSize': headers.PhonebookSize(phonebook_size)} - self.send_response(socket, responses.Success(), [ - headers.App_Parameters(response_dict)]) + logger.debug( + "MaxListCount is 0, so responding with PhonebookSize = {}".format( + phonebook_size + ) + ) + response_dict = {"PhonebookSize": headers.PhonebookSize(phonebook_size)} + self.send_response( + socket, responses.Success(), [headers.App_Parameters(response_dict)] + ) def _decode_header_data(self, request): """Decodes all headers in given request and return the decoded values in dict""" @@ -316,17 +363,23 @@ def _decode_header_data(self, request): header_dict["App_Parameters"] = header.decode() logger.info("App Parameters are :") for param, value in header_dict["App_Parameters"].items(): - logger.info("{param}: {value}".format(param=param, value=value.decode())) + logger.info( + "{param}: {value}".format(param=param, value=value.decode()) + ) else: logger.error("Some Header data is not yet added in _decode_header_data") - raise NotImplementedError("Some Header data is not yet added in _decode_header_data") + raise NotImplementedError( + "Some Header data is not yet added in _decode_header_data" + ) return header_dict def _limit_phonebook(self, vcard_list, max_listcount, list_startoffset=0): """limit the phonebook size based on max listcount and list startoffset and update the index of phonebook accordingly.""" vcard_list = vcard_list[list_startoffset:] - if max_listcount != 65535: # Range: 0 <= max_listcount <= 65535 (65535 => Unrestricted) + if ( + max_listcount != 65535 + ): # Range: 0 <= max_listcount <= 65535 (65535 => Unrestricted) vcard_list = vcard_list[:max_listcount] return vcard_list @@ -342,7 +395,9 @@ def _decode_app_params(self, app_params): else: decoded_app_params["SearchValue"] = "" if "SearchAttribute" in app_params: - decoded_app_params["SearchAttribute"] = app_params["SearchAttribute"].decode() + decoded_app_params["SearchAttribute"] = app_params[ + "SearchAttribute" + ].decode() else: decoded_app_params["SearchAttribute"] = 0 # Default: Name attribute if "MaxListCount" in app_params: @@ -350,15 +405,21 @@ def _decode_app_params(self, app_params): else: decoded_app_params["MaxListCount"] = 65535 if "ListStartOffset" in app_params: - decoded_app_params["ListStartOffset"] = app_params["ListStartOffset"].decode() + decoded_app_params["ListStartOffset"] = app_params[ + "ListStartOffset" + ].decode() else: decoded_app_params["ListStartOffset"] = 0 # Default: 0 if "Filter" in app_params: decoded_app_params["Filter"] = app_params["Filter"].decode() else: - decoded_app_params["Filter"] = 0 # Default: 0 [means should return all the attributes] + decoded_app_params[ + "Filter" + ] = 0 # Default: 0 [means should return all the attributes] if "Format" in app_params: - decoded_app_params["Format"] = "3.0" if app_params["Format"].decode() else "2.1" + decoded_app_params["Format"] = ( + "3.0" if app_params["Format"].decode() else "2.1" + ) else: decoded_app_params["Format"] = "2.1" # Default: v2.1 if "PhonebookSize" in app_params: @@ -369,8 +430,11 @@ def _decode_app_params(self, app_params): def _filter_attributes(self, filter_bitmask, data, vcard_version="2.1"): """receives filter bitmask and vcard data as dict then returns the filtered dict""" - logger.debug("Filtering attributes for bitmask: {bitmask} data: {data}".format( - bitmask=filter_bitmask, data=data)) + logger.debug( + "Filtering attributes for bitmask: {bitmask} data: {data}".format( + bitmask=filter_bitmask, data=data + ) + ) # if filter is 0, return all the attributes if filter_bitmask == 0: return data @@ -381,7 +445,9 @@ def _filter_attributes(self, filter_bitmask, data, vcard_version="2.1"): bit = 1 << bitmarker if bit & filter_bitmask == bit: unfiltered_attrs.add(attr_tuple[0]) - logger.debug("Necessary attributes: {unfiltered}".format(unfiltered=unfiltered_attrs)) + logger.debug( + "Necessary attributes: {unfiltered}".format(unfiltered=unfiltered_attrs) + ) for param in data["vcard"][:]: attr = param["type"] if attr in unfiltered_attrs: @@ -431,8 +497,15 @@ def start_service(self, port=PORT_ANY): # protocols = [L2CAP_UUID, RFCOMM_UUID, OBEX_UUID] return server.Server.start_service( - self, port, name, uuid, service_classes, service_profiles, - provider, description, [] + self, + port, + name, + uuid, + service_classes, + service_profiles, + provider, + description, + [], ) @@ -453,17 +526,25 @@ def run_server(device_address, rootdir, use_fs): def main(): - logging.basicConfig(level=logging.DEBUG, - format='%(asctime)s %(name)s %(levelname)-8s %(message)s') + logging.basicConfig( + level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)-8s %(message)s" + ) parser = argparse.ArgumentParser(description="Phonebook Access Profile") - parser.add_argument("--address", required=True, - help="bluetooth address to start the server") - parser.add_argument("--use-fs", action="store_true", - help="Use the phonebook virtual folder stored in filesystem." - "(if not given will use the phonebook from mongodb)") - parser.add_argument("--rootdir", help="rootdir of phonebook virtual folder, " - "required while using filesystem as storage") + parser.add_argument( + "--address", required=True, help="bluetooth address to start the server" + ) + parser.add_argument( + "--use-fs", + action="store_true", + help="Use the phonebook virtual folder stored in filesystem." + "(if not given will use the phonebook from mongodb)", + ) + parser.add_argument( + "--rootdir", + help="rootdir of phonebook virtual folder, " + "required while using filesystem as storage", + ) args = parser.parse_args() if args.use_fs and args.rootdir is None: diff --git a/setup.py b/setup.py index ba24cb4..e11c2df 100644 --- a/setup.py +++ b/setup.py @@ -19,11 +19,8 @@ keywords="pbap", zip_safe=False, packages=find_packages(), - dependency_links=["https://bitbucket.org/dboddie/pyobex/get/tip.zip#egg=pyobex-0.26"], - install_requires=[ - "pybluez==0.22", - "pyobex>=0.26", - "pymongo", - "cmd2==0.8.8" + dependency_links=[ + "https://bitbucket.org/dboddie/pyobex/get/tip.zip#egg=pyobex-0.26" ], + install_requires=["pybluez==0.22", "pyobex>=0.26", "pymongo", "cmd2==0.8.8"], ) diff --git a/vcard_helper.py b/vcard_helper.py index 7c329b3..cce11b3 100644 --- a/vcard_helper.py +++ b/vcard_helper.py @@ -33,27 +33,33 @@ def serialize(self, version="2.1"): vcf_str = "" for vcard_prop_dict in self.denormalize(self.to_dict(), version)["vcard"]: vcard_property_class = VCardProperties[vcard_prop_dict["type"]] - vcf_str += vcard_property_class(vcard_prop_dict, parsed=True).serialize(version) + vcf_str += vcard_property_class(vcard_prop_dict, parsed=True).serialize( + version + ) return vcf_str def normalize(self, vcf_dict): """converts version dependent vcard dict to version independent one""" # removes begin, version, end properties which will be added on denormalization vcf_dict_copy = copy.deepcopy(vcf_dict) - vcf_dict_copy['vcard'].pop(0) - vcf_dict_copy['vcard'].pop(0) - vcf_dict_copy['vcard'].pop() + vcf_dict_copy["vcard"].pop(0) + vcf_dict_copy["vcard"].pop(0) + vcf_dict_copy["vcard"].pop() return vcf_dict_copy def denormalize(self, vcf_dict, version="2.1"): """converts version independent vcard dict to version dependent one""" vcf_dict_copy = copy.deepcopy(vcf_dict) - begin_dict = {'values': ['VCARD'], 'type': 'BEGIN', 'parameters': []} - version_dict = {'values': ['{version}'.format(version=version)], 'type': 'VERSION', 'parameters': []} - end_dict = {'values': ['VCARD'], 'type': 'END', 'parameters': []} - vcf_dict_copy['vcard'].insert(0, begin_dict) - vcf_dict_copy['vcard'].insert(1, version_dict) - vcf_dict_copy['vcard'].append(end_dict) + begin_dict = {"values": ["VCARD"], "type": "BEGIN", "parameters": []} + version_dict = { + "values": ["{version}".format(version=version)], + "type": "VERSION", + "parameters": [], + } + end_dict = {"values": ["VCARD"], "type": "END", "parameters": []} + vcf_dict_copy["vcard"].insert(0, begin_dict) + vcf_dict_copy["vcard"].insert(1, version_dict) + vcf_dict_copy["vcard"].append(end_dict) return vcf_dict_copy def to_dict(self): @@ -94,6 +100,7 @@ def parse(self, vcf_str): class VCardProperty(object): """Represents property of VCard""" + STR_TEMPLATE = "{type}{sep}{param}:{value}\r\n" def __init__(self, vcard, parsed=False): @@ -103,10 +110,12 @@ def serialize(self, version="2.1"): """Serializes the current vcard property into string""" vcf_dict = self.denormalize(self.vcf_dict, version) params = self._tuples_to_params(vcf_dict["parameters"]) - vcf_result = self.STR_TEMPLATE.format(type=vcf_dict["type"], - sep=";" if params else "", - param=params, - value=";".join([item.encode('utf-8') for item in vcf_dict["values"]])) + vcf_result = self.STR_TEMPLATE.format( + type=vcf_dict["type"], + sep=";" if params else "", + param=params, + value=";".join([item.encode("utf-8") for item in vcf_dict["values"]]), + ) return vcf_result def to_dict(self): @@ -156,7 +165,7 @@ def parse(self, vcf_str): vcf_dict = { "type": type_, "values": rhs.split(";"), - "parameters": self._params_to_tuple(params) + "parameters": self._params_to_tuple(params), } return self.normalize(vcf_dict) @@ -166,7 +175,9 @@ class VCardProperty_CharsetEncodingParamNormalized(VCardProperty): def normalize(self, vcf_dict): """converts version dependent vcard dict to version independent one""" - vcf_dict = super(VCardProperty_CharsetEncodingParamNormalized, self).normalize(vcf_dict) + vcf_dict = super(VCardProperty_CharsetEncodingParamNormalized, self).normalize( + vcf_dict + ) vcf_dict = copy.deepcopy(vcf_dict) charset = None encoding = None @@ -184,14 +195,18 @@ def normalize(self, vcf_dict): value, _ = codecs.lookup(encoding).decode(";".join(vcf_dict["values"])) vcf_dict["values"] = value.split(";") if charset: - value, _ = codecs.lookup(charset).decode(";".join(vcf_dict["values"]), errors="replace") + value, _ = codecs.lookup(charset).decode( + ";".join(vcf_dict["values"]), errors="replace" + ) vcf_dict["values"] = value.split(";") vcf_dict["values"] = ";".join(vcf_dict["values"]).encode("utf8").split(";") return vcf_dict def denormalize(self, vcf_dict, version="2.1"): """converts version independent vcard dict to version dependent one""" - vcf_dict = super(VCardProperty_CharsetEncodingParamNormalized, self).denormalize(vcf_dict, version) + vcf_dict = super( + VCardProperty_CharsetEncodingParamNormalized, self + ).denormalize(vcf_dict, version) vcf_dict = copy.deepcopy(vcf_dict) encoding = "QUOTED-PRINTABLE" charset = "UTF-8" @@ -213,12 +228,16 @@ class VCardProperty_BASE64EncodingParamNormalized(VCardProperty): def normalize(self, vcf_dict): """converts version dependent vcard dict to version independent one""" - vcf_dict = super(VCardProperty_BASE64EncodingParamNormalized, self).normalize(vcf_dict) + vcf_dict = super(VCardProperty_BASE64EncodingParamNormalized, self).normalize( + vcf_dict + ) vcf_dict = copy.deepcopy(vcf_dict) normalized_params = [] for param in vcf_dict["parameters"]: if param[0].upper() == "ENCODING": - normalized_params.append(("ENCODING", "b")) # key property only have base64 encoding + normalized_params.append( + ("ENCODING", "b") + ) # key property only have base64 encoding elif param[0].upper() == "" or param[0].upper() == "TYPE": normalized_params.append(("TYPE", param[1])) else: @@ -228,15 +247,21 @@ def normalize(self, vcf_dict): def denormalize(self, vcf_dict, version="2.1"): """converts version independent vcard dict to version dependent one""" - vcf_dict = super(VCardProperty_BASE64EncodingParamNormalized, self).denormalize(vcf_dict, version) + vcf_dict = super(VCardProperty_BASE64EncodingParamNormalized, self).denormalize( + vcf_dict, version + ) vcf_dict = copy.deepcopy(vcf_dict) denormalized_params = [] for param in vcf_dict["parameters"]: if param[0].upper() == "ENCODING": if version == "2.1": - denormalized_params.append(("ENCODING", "BASE64")) # key property only have base64 encoding + denormalized_params.append( + ("ENCODING", "BASE64") + ) # key property only have base64 encoding elif version == "3.0": - denormalized_params.append(("ENCODING", "b")) # key property only have base64 encoding + denormalized_params.append( + ("ENCODING", "b") + ) # key property only have base64 encoding else: raise TypeError("Unsupported version") elif param[0].upper() == "TYPE": diff --git a/vfolder.py b/vfolder.py index ad89000..dc40e3e 100644 --- a/vfolder.py +++ b/vfolder.py @@ -99,7 +99,9 @@ def makedirs(self, path): def listdir(self, path, query={}, projection={"_id": False}, sort=("_id", 1)): if not self.isdir(path): - raise RuntimeError("Specified path {path} is not a directory.".format(path=path)) + raise RuntimeError( + "Specified path {path} is not a directory.".format(path=path) + ) else: db, coll = self._path_to_db_elements(path) return list(db[coll].find(query, projection).sort(*sort)) @@ -110,7 +112,14 @@ def read(self, path, query={}, projection={"_id": False}, sort=("_id", 1)): else: vcard_index = int(os.path.splitext(os.path.basename(path))[0]) db, coll = self._path_to_db_elements(path) - return db[coll].find(query, projection).sort(*sort).skip(vcard_index).limit(1).next() + return ( + db[coll] + .find(query, projection) + .sort(*sort) + .skip(vcard_index) + .limit(1) + .next() + ) def chdir(self, path): if not self.exists(path): @@ -120,7 +129,9 @@ def chdir(self, path): def count(self, path): if not self.isdir(path): - raise RuntimeError("Specified path {path} is not a directory.".format(path=path)) + raise RuntimeError( + "Specified path {path} is not a directory.".format(path=path) + ) else: db, coll = self._path_to_db_elements(path) return db[coll].count() @@ -178,8 +189,12 @@ def listdir(self, path, query={}, projection={"_id": False}, sort=("_id", 1)): abspath = os.path.abspath(os.path.join(self.curdir, path)) dir_contents = [] # TODO: Need to add sort, project, query functionalities - for pb_object in sorted(os.listdir(abspath), key=lambda x: int(os.path.splitext(x)[0])): - vcard_dict = VCard(open(os.path.join(abspath, pb_object)).read(), parsed=False).to_dict() + for pb_object in sorted( + os.listdir(abspath), key=lambda x: int(os.path.splitext(x)[0]) + ): + vcard_dict = VCard( + open(os.path.join(abspath, pb_object)).read(), parsed=False + ).to_dict() dir_contents.append(vcard_dict) return dir_contents From ddf1ded8edf0529118abf90836af65d7871ccd8f Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 11:51:45 +0300 Subject: [PATCH 02/12] More refactoring --- pbapclient.py | 32 ++++++++++++++++++-------------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 23fe03a..14fd7ed 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -311,22 +311,26 @@ def do_pull_vcard_listing(self, line, opts): list_startoffset=opts.start_offset, ) if result is not None: - header, data = result + _, data = result logger.info("Result of pull_vcard_listing:\n%s", data) - # @cmd2.options( - # [ - # make_option( - # "-f", - # "--filter", - # default=0x00000000, - # type=int, - # help="Attributes filter mask", - # ), - # make_option("-t", "--format", default=0, type=int, help="vcard format"), - # ], - # arg_desc="vcard_handle", - # ) + pull_vcard_entry_parser = cmd2.Cmd2ArgumentParser() + pull_vcard_entry_parser.add_argument( + "-f", + "--filter", + default=0x00000000, + type=int, + help="Attributes filter mask", + ) + pull_vcard_entry_parser.add_argument( + "-t", + "--format", + default=0, + type=int, + help="vcard format", + ) + + @cmd2.with_argparser(pull_vcard_entry_parser) def do_pull_vcard_entry(self, line, opts): """Returns a single vcardentry as per requested options""" result = self.client.pull_vcard_entry( From e748df89d3227d6f5454c40485ba0085d837bd02 Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 11:53:18 +0300 Subject: [PATCH 03/12] More transition --- pbapclient.py | 34 ++++++++++++++++------------------ 1 file changed, 16 insertions(+), 18 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 14fd7ed..cbd4f78 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -337,26 +337,24 @@ def do_pull_vcard_entry(self, line, opts): name=line, filter_=opts.filter, format_=opts.format ) if result is not None: - header, data = result + _, data = result logger.info("Result of pull_vcard_entry:\n%s", data) - # @cmd2.options( - # [ - # make_option( - # "--to-parent", - # action="store_true", - # default=False, - # help="navigate to parent dir", - # ), - # make_option( - # "--to-root", - # action="store_true", - # default=False, - # help="navigate to root dir", - # ), - # ], - # arg_desc="[folder_name]", - # ) + set_phonebook_parser = cmd2.Cmd2ArgumentParser() + set_phonebook_parser.add_argument( + "--to-parent", + action="store_true", + default=False, + help="navigate to parent dir", + ) + set_phonebook_parser.add_argument( + "--to-root", + action="store_true", + default=False, + help="navigate to root dir", + ) + + @cmd2.with_argparser(set_phonebook_parser) def do_set_phonebook(self, line, opts): """Set current folder path of pbapserver virtual folder""" result = self.client.set_phonebook( From a742e7930f694f203a1e540c04a873a29467e276 Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 11:56:23 +0300 Subject: [PATCH 04/12] More transition --- pbapclient.py | 76 +++++++++++++++++++++++++++------------------------ 1 file changed, 41 insertions(+), 35 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index cbd4f78..ecfdbf1 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -189,9 +189,8 @@ def _store_history(): atexit.register(readline.write_history_file, history_file) # @cmd2.options([], arg_desc="server_address") - def do_connect(self, line, opts): - self.add_settable(cmd2.Settable(name="server_address", val_type=str)) - + def do_connect(self, line): + """Connect to PBABClient""" profile_id = "1130" # profile id of PBAP service_id = b"\x79\x61\x35\xf0\xf0\xc5\x11\xd8\x09\x66\x08\x00\x20\x0c\x9a\x66" server_address = line @@ -217,8 +216,8 @@ def do_connect(self, line, opts): # self.prompt = self.colorize("pbap> ", "green") self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.GREEN) - # @cmd2.options([], arg_desc="") - def do_disconnect(self, line, opts): + def do_disconnect(self): + """Disconnect from PBABClient""" if self.client is None: logger.error( "PBAPClient is not even connected.. Connect and then try disconnect" @@ -230,33 +229,37 @@ def do_disconnect(self, line, opts): # self.prompt = self.colorize("pbap> ", "yellow") self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.YELLOW) - # @cmd2.options( - # [ - # make_option( - # "-f", - # "--filter", - # default=0x00000000, - # type=int, - # help="Attributes filter mask", - # ), - # make_option("-t", "--format", default=0, type=int, help="vcard format"), - # make_option( - # "-c", - # "--max-count", - # default=65535, - # type=int, - # help="maximum number of contacts to be returned", - # ), - # make_option( - # "-o", - # "--start-offset", - # default=0, - # type=int, - # help="offset of first entry to be returned", - # ), - # ], - # arg_desc="phonebook_name", - # ) + pull_phonebook_parser = cmd2.Cmd2ArgumentParser() + pull_phonebook_parser.add_argument( + "-f", + "--filter", + default=0x00000000, + type=int, + help="Attributes filter mask", + ) + pull_phonebook_parser.add_argument( + "-t", + "--format", + default=0, + type=int, + help="vcard format", + ) + pull_phonebook_parser.add_argument( + "-c", + "--max-count", + default=65535, + type=int, + help="Maximum number of contacts to be returned", + ) + pull_phonebook_parser.add_argument( + "-o", + "--start-offset", + default=0, + type=int, + help="offset of first entry to be returned", + ) + + @cmd2.with_argparser(pull_phonebook_parser) def do_pull_phonebook(self, line, opts): """Returns phonebook as per requested options""" result = self.client.pull_phonebook( @@ -364,7 +367,7 @@ def do_set_phonebook(self, line, opts): logger.info("Result of set_phonebook:\n%s", result) # @cmd2.options([], arg_desc="server_address [folder_name]") - def do_mirror_vfolder(self, line, opts): + def do_mirror_vfolder(self, line, _): """Downloads phonebook from pbapserver and save it in virtual folder architecture in FS""" args = line.split() self.do_connect(args[0] if len(args) else "") @@ -417,7 +420,9 @@ def do_mirror_vfolder(self, line, opts): continue hdrs, card = response logger.info(card) - with open(os.path.join(current_dir, name), "w+") as f: + with open( + os.path.join(current_dir, name), "w+", encoding="latin1" + ) as f: f.write(card) logger.debug("current_dir = %s", current_dir) @@ -448,11 +453,12 @@ def do_mirror_vfolder(self, line, opts): with open( os.path.join(current_dir, prefix, "telecom", pbobject + ".vcf"), "w+", + encoding="latin1", ) as f: f.write(phonebook) logger.info(hdrs) - self.do_disconnect("") + self.do_disconnect() do_q = cmd2.Cmd.do_quit From 6565271ae9f2fe5632fc416e6be2c716c06c958a Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 12:12:35 +0300 Subject: [PATCH 05/12] More transition work --- pbapclient.py | 32 +++++++++++++++++++++++--------- 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index ecfdbf1..2b75be7 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -188,7 +188,11 @@ def _store_history(): readline.read_history_file(history_file) atexit.register(readline.write_history_file, history_file) - # @cmd2.options([], arg_desc="server_address") + # connect_parser + connect_parser = cmd2.Cmd2ArgumentParser() + connect_parser.add_argument("line", help="server_address") + + @cmd2.with_argparser(connect_parser) def do_connect(self, line): """Connect to PBABClient""" profile_id = "1130" # profile id of PBAP @@ -229,6 +233,7 @@ def do_disconnect(self): # self.prompt = self.colorize("pbap> ", "yellow") self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.YELLOW) + # pull_phonebook_parser pull_phonebook_parser = cmd2.Cmd2ArgumentParser() pull_phonebook_parser.add_argument( "-f", @@ -258,6 +263,7 @@ def do_disconnect(self): type=int, help="offset of first entry to be returned", ) + pull_phonebook_parser.add_argument("line", help="phonebook_name") @cmd2.with_argparser(pull_phonebook_parser) def do_pull_phonebook(self, line, opts): @@ -273,6 +279,7 @@ def do_pull_phonebook(self, line, opts): _, data = result logger.info("Result of pull_phonebook:\n%s", data) + # pull_vcard_listing_parser pull_vcard_listing_parser = cmd2.Cmd2ArgumentParser() pull_vcard_listing_parser.add_argument( "-r", @@ -301,12 +308,13 @@ def do_pull_phonebook(self, line, opts): type=int, help="offset of first entry to be returned", ) + pull_vcard_listing_parser.add_argument("line", help="vcard_folder") @cmd2.with_argparser(pull_vcard_listing_parser) - def do_pull_vcard_listing(self, line, opts): + def do_pull_vcard_listing(self, opts): """Returns vcardlisting as per requested options""" result = self.client.pull_vcard_listing( - name=line, + name=opts.line, order=opts.order, search_value=opts.search_value, search_attribute=opts.search_attribute, @@ -332,12 +340,13 @@ def do_pull_vcard_listing(self, line, opts): type=int, help="vcard format", ) + pull_vcard_entry_parser.add_argument("line", help="vcard_handle") @cmd2.with_argparser(pull_vcard_entry_parser) - def do_pull_vcard_entry(self, line, opts): + def do_pull_vcard_entry(self, opts): """Returns a single vcardentry as per requested options""" result = self.client.pull_vcard_entry( - name=line, filter_=opts.filter, format_=opts.format + name=opts.line, filter_=opts.filter, format_=opts.format ) if result is not None: _, data = result @@ -356,20 +365,25 @@ def do_pull_vcard_entry(self, line, opts): default=False, help="navigate to root dir", ) + set_phonebook_parser.add_argument("line", help="[folder_name]") @cmd2.with_argparser(set_phonebook_parser) - def do_set_phonebook(self, line, opts): + def do_set_phonebook(self, opts): """Set current folder path of pbapserver virtual folder""" result = self.client.set_phonebook( - name=line, to_parent=opts.to_parent, to_root=opts.to_root + name=opts.line, to_parent=opts.to_parent, to_root=opts.to_root ) if result is not None: logger.info("Result of set_phonebook:\n%s", result) # @cmd2.options([], arg_desc="server_address [folder_name]") - def do_mirror_vfolder(self, line, _): + mirror_vfolder_parser = cmd2.Cmd2ArgumentParser() + mirror_vfolder_parser.add_argument("line", help="server_address [folder_name]") + + @cmd2.with_argparser(mirror_vfolder_parser) + def do_mirror_vfolder(self, opts): """Downloads phonebook from pbapserver and save it in virtual folder architecture in FS""" - args = line.split() + args = opts.line.split() self.do_connect(args[0] if len(args) else "") rootdir = args[1] if len(args) >= 2 else "phonebook_vfolder" # TODO: need to handle multiple SIM contacts From dfa5395456c5bc545beac452169aa860a5d389a7 Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 12:14:44 +0300 Subject: [PATCH 06/12] More transition --- pbapclient.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 2b75be7..78aa26a 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -193,11 +193,11 @@ def _store_history(): connect_parser.add_argument("line", help="server_address") @cmd2.with_argparser(connect_parser) - def do_connect(self, line): + def do_connect(self, opts): """Connect to PBABClient""" profile_id = "1130" # profile id of PBAP service_id = b"\x79\x61\x35\xf0\xf0\xc5\x11\xd8\x09\x66\x08\x00\x20\x0c\x9a\x66" - server_address = line + server_address = opts.line if not server_address: raise ValueError("server_address should not be empty") logger.info("Finding PBAP service ...") From 780bc225c10dc780fde542405f0e48800772eb6c Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 12:18:31 +0300 Subject: [PATCH 07/12] Make sense in opts --- pbapclient.py | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 78aa26a..02f5fa8 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -190,14 +190,14 @@ def _store_history(): # connect_parser connect_parser = cmd2.Cmd2ArgumentParser() - connect_parser.add_argument("line", help="server_address") + connect_parser.add_argument("server_address", help="server_address") @cmd2.with_argparser(connect_parser) def do_connect(self, opts): """Connect to PBABClient""" profile_id = "1130" # profile id of PBAP service_id = b"\x79\x61\x35\xf0\xf0\xc5\x11\xd8\x09\x66\x08\x00\x20\x0c\x9a\x66" - server_address = opts.line + server_address = opts.server_address if not server_address: raise ValueError("server_address should not be empty") logger.info("Finding PBAP service ...") @@ -263,13 +263,13 @@ def do_disconnect(self): type=int, help="offset of first entry to be returned", ) - pull_phonebook_parser.add_argument("line", help="phonebook_name") + pull_phonebook_parser.add_argument("phonebook_name", help="phonebook_name") @cmd2.with_argparser(pull_phonebook_parser) - def do_pull_phonebook(self, line, opts): + def do_pull_phonebook(self, opts): """Returns phonebook as per requested options""" result = self.client.pull_phonebook( - name=line, + name=opts.phonebook_name, filter_=opts.filter, format_=opts.format, max_list_count=opts.max_count, @@ -294,6 +294,11 @@ def do_pull_phonebook(self, line, opts): type=int, help="SearchAttribute {Name | Number | Sound }", ) + pull_vcard_listing_parser.add_argument( + "--search-value", + default=None, + help="SearchValue {}", + ) pull_vcard_listing_parser.add_argument( "-c", "--max-count", @@ -308,13 +313,13 @@ def do_pull_phonebook(self, line, opts): type=int, help="offset of first entry to be returned", ) - pull_vcard_listing_parser.add_argument("line", help="vcard_folder") + pull_vcard_listing_parser.add_argument("vcard_folder", help="vcard_folder") @cmd2.with_argparser(pull_vcard_listing_parser) def do_pull_vcard_listing(self, opts): """Returns vcardlisting as per requested options""" result = self.client.pull_vcard_listing( - name=opts.line, + name=opts.vcard_folder, order=opts.order, search_value=opts.search_value, search_attribute=opts.search_attribute, @@ -340,13 +345,13 @@ def do_pull_vcard_listing(self, opts): type=int, help="vcard format", ) - pull_vcard_entry_parser.add_argument("line", help="vcard_handle") + pull_vcard_entry_parser.add_argument("vcard_handle", help="vcard_handle") @cmd2.with_argparser(pull_vcard_entry_parser) def do_pull_vcard_entry(self, opts): """Returns a single vcardentry as per requested options""" result = self.client.pull_vcard_entry( - name=opts.line, filter_=opts.filter, format_=opts.format + name=opts.vcard_handle, filter_=opts.filter, format_=opts.format ) if result is not None: _, data = result @@ -365,13 +370,13 @@ def do_pull_vcard_entry(self, opts): default=False, help="navigate to root dir", ) - set_phonebook_parser.add_argument("line", help="[folder_name]") + set_phonebook_parser.add_argument("folder_name", help="[folder_name]") @cmd2.with_argparser(set_phonebook_parser) def do_set_phonebook(self, opts): """Set current folder path of pbapserver virtual folder""" result = self.client.set_phonebook( - name=opts.line, to_parent=opts.to_parent, to_root=opts.to_root + name=opts.folder_name, to_parent=opts.to_parent, to_root=opts.to_root ) if result is not None: logger.info("Result of set_phonebook:\n%s", result) From 0676874d7ec7355f1a954b1b303b1d157fc6301b Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 12:29:36 +0300 Subject: [PATCH 08/12] More code styling --- pbapclient.py | 29 +++++++---------------------- 1 file changed, 7 insertions(+), 22 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 02f5fa8..324c6bf 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -152,7 +152,8 @@ def set_phonebook(self, name="", to_root=False, to_parent=False): if not isinstance(response, tuple) and isinstance( response, responses.FailureResponse ): - logger.error("set_phonebook failed. reason = %s", name, response) + err = f"set_phonebook failed. reason = {response}" + logger.error(err) return if to_root: @@ -169,9 +170,7 @@ class REPL(cmd2.Cmd): def __init__(self): cmd2.Cmd.__init__(self) - # self.prompt = self.colorize("pbap> ", "yellow") self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.YELLOW) - # self.intro = self.colorize("Welcome to the PhoneBook Access Profile!", "green") self.intro = cmd2.ansi.style( "Welcome to the PhoneBook Access Profile!", fg=cmd2.ansi.Fg.GREEN ) @@ -217,7 +216,6 @@ def do_connect(self, opts): logger.error("Connect Failed, Terminating the Pbap client..") sys.exit(2) logger.info("Connect success") - # self.prompt = self.colorize("pbap> ", "green") self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.GREEN) def do_disconnect(self): @@ -230,7 +228,6 @@ def do_disconnect(self): logger.debug("Disconnecting pbap client with pbap server") self.client.disconnect() self.client = None - # self.prompt = self.colorize("pbap> ", "yellow") self.prompt = cmd2.ansi.style("pbap> ", fg=cmd2.ansi.Fg.YELLOW) # pull_phonebook_parser @@ -403,11 +400,7 @@ def do_mirror_vfolder(self, opts): current_dir = os.path.join(telecom_dir, pbobject) os.makedirs(current_dir) # Access the list of vcards in the phone's internal phone book. - response = self.client.pull_vcard_listing( - "{prefix}telecom/{pbobject}".format( - prefix=prefix, pbobject=pbobject - ) - ) + response = self.client.pull_vcard_listing(f"{prefix}telecom/{pbobject}") if response is None: logger.error( "vcard-listing get is failed for pbobject '%s'", pbobject @@ -427,11 +420,7 @@ def do_mirror_vfolder(self, opts): logger.info("\nCards in %stelecom/%s\n", prefix, pbobject) # Request all the file names obtained earlier. - self.client.set_phonebook( - "{prefix}telecom/{pbobject}".format( - prefix=prefix, pbobject=pbobject - ) - ) + self.client.set_phonebook(f"{prefix}telecom/{pbobject}") for name in names: response = self.client.pull_vcard_entry(name) if response is None: @@ -459,11 +448,7 @@ def do_mirror_vfolder(self, opts): logger.info( "\nThe phonebook in %s/telecom/%s as one vcard\n", prefix, pbobject ) - response = self.client.pull_phonebook( - "{prefix}telecom/{pbobject}.vcf".format( - prefix=prefix, pbobject=pbobject - ) - ) + response = self.client.pull_phonebook(f"{prefix}telecom/{pbobject}.vcf") if response is None: logger.error("phonebook get is failed for pbobject '%s'", pbobject) continue @@ -473,8 +458,8 @@ def do_mirror_vfolder(self, opts): os.path.join(current_dir, prefix, "telecom", pbobject + ".vcf"), "w+", encoding="latin1", - ) as f: - f.write(phonebook) + ) as file_handle: + file_handle.write(phonebook) logger.info(hdrs) self.do_disconnect() From c9e22a6121454294680c9f99f9546e476acf00a7 Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 13:05:02 +0300 Subject: [PATCH 09/12] Implement fix for https://github.com/bmwcarit/pypbap/issues/5 --- pbapclient.py | 2 +- pbapheaders.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 324c6bf..263c421 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -48,7 +48,7 @@ def pull_phonebook( "ListStartOffset": headers.ListStartOffset(list_startoffset), } application_parameters = headers.App_Parameters(data, encoded=False) - header_list = [headers.PBAPType("x-bt/phonebook")] + header_list = [headers.PBAPType(b"x-bt/phonebook")] if application_parameters.data: header_list.append(application_parameters) diff --git a/pbapheaders.py b/pbapheaders.py index 88ceee9..29dcc9f 100644 --- a/pbapheaders.py +++ b/pbapheaders.py @@ -70,11 +70,11 @@ def encode(self, data): def decode(self): headers, data = super(VariableLengthProperty, self).decode() - tagid, length = headers + _, length = headers return struct.unpack(self.fmt.format(len=length), data)[0] -class PBAPType(UnicodeHeader): +class PBAPType(Type): code = 0x42 From 769be90f5603154fc5bc4d53b93bdc4ee61631f2 Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 13:11:56 +0300 Subject: [PATCH 10/12] Code cleanup --- pbapheaders.py | 40 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 37 insertions(+), 3 deletions(-) diff --git a/pbapheaders.py b/pbapheaders.py index 29dcc9f..790fa4d 100644 --- a/pbapheaders.py +++ b/pbapheaders.py @@ -3,12 +3,18 @@ # -*- coding: utf-8 -*- """Phone Book Access Profile headers""" +import struct + from PyOBEX.headers import * from pbapcommon import FILTER_ATTR_DICT # Application Parameters Header Properties class AppParamProperty(object): + """AppParamProperty Class""" + + tagid = None + def __init__(self, data, encoded=False): if encoded: self.data = data @@ -25,6 +31,8 @@ def decode(self): class OneByteProperty(AppParamProperty): + """OneByteProperty Class""" + length = 1 # byte fmt = ">B" @@ -32,11 +40,13 @@ def encode(self, data): return super(OneByteProperty, self).encode(struct.pack(self.fmt, data)) def decode(self): - headers, data = super(OneByteProperty, self).decode() + _, data = super(OneByteProperty, self).decode() return struct.unpack(self.fmt, data)[0] class TwoByteProperty(AppParamProperty): + """TwoByteProperty Class""" + length = 2 # bytes fmt = ">H" @@ -44,11 +54,13 @@ def encode(self, data): return super(TwoByteProperty, self).encode(struct.pack(self.fmt, data)) def decode(self): - headers, data = super(TwoByteProperty, self).decode() + _, data = super(TwoByteProperty, self).decode() return struct.unpack(self.fmt, data)[0] class EightByteProperty(AppParamProperty): + """EightByteProperty Class""" + length = 8 # bytes fmt = ">Q" @@ -56,7 +68,7 @@ def encode(self, data): return super(EightByteProperty, self).encode(struct.pack(self.fmt, data)) def decode(self): - headers, data = super(EightByteProperty, self).decode() + _, data = super(EightByteProperty, self).decode() return struct.unpack(self.fmt, data)[0] @@ -75,43 +87,63 @@ def decode(self): class PBAPType(Type): + """PBAPType Class""" + code = 0x42 class Order(OneByteProperty): + """Order Class""" + tagid = 0x01 class SearchValue(VariableLengthProperty): + """SearchValue Class""" + tagid = 0x02 class SearchAttribute(OneByteProperty): + """SearchAttribute Class""" + tagid = 0x03 class MaxListCount(TwoByteProperty): + """MaxListCount Class""" + tagid = 0x04 class ListStartOffset(TwoByteProperty): + """ListStartOffset Class""" + tagid = 0x05 class Filter(EightByteProperty): + """Filter Class""" + tagid = 0x06 attr_dict = FILTER_ATTR_DICT class Format(OneByteProperty): + """Format Class""" + tagid = 0x07 class PhonebookSize(TwoByteProperty): + """PhonebookSize Class""" + tagid = 0x08 class NewMissedCalls(OneByteProperty): + """NewMissedCalls Class""" + tagid = 0x09 @@ -134,6 +166,7 @@ class NewMissedCalls(OneByteProperty): def extended_decode(self): + """Extended version of the 'decode'""" # assumption: # size of tagid = 1 byte # size of length = 1 byte (This is just the data length) @@ -151,6 +184,7 @@ def extended_decode(self): def extended_encode(self, data_dict): + """Extended version of the 'encode'""" data = b"" for item in data_dict.values(): if item is None: From 9ba592a0fee518f928467db00ed1a1cde930a580 Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 13:21:27 +0300 Subject: [PATCH 11/12] "bytes" not str --- pbapclient.py | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/pbapclient.py b/pbapclient.py index 263c421..038968e 100644 --- a/pbapclient.py +++ b/pbapclient.py @@ -72,7 +72,8 @@ def pull_vcard_listing( list_startoffset=0, ): """Retrieves phonebook listing object from current folder""" - logger.info("Requesting pull_vcard_listing with parameters %s", str(locals())) + info = f"Requesting pull_vcard_listing with parameters {str(locals())}" + logger.info(info) data = { "Order": headers.Order(order), "MaxListCount": headers.MaxListCount(max_list_count), @@ -86,7 +87,7 @@ def pull_vcard_listing( } ) application_parameters = headers.App_Parameters(data, encoded=False) - header_list = [headers.PBAPType("x-bt/vcard-listing")] + header_list = [headers.PBAPType(b"x-bt/vcard-listing")] if application_parameters.data: header_list.append(application_parameters) @@ -94,11 +95,8 @@ def pull_vcard_listing( if not isinstance(response, tuple) and isinstance( response, responses.FailureResponse ): - logger.error( - "pull_vcard_listing failed for pbobject '%s'. reason = %s", - name, - response, - ) + err = f"pull_vcard_listing failed for pbobject '{name}'. reason = {response}" + logger.error(err) return return response @@ -110,7 +108,7 @@ def pull_vcard_entry(self, name, filter_=0, format_=0): data = {"Filter": headers.Filter(filter_), "Format": headers.Format(format_)} application_parameters = headers.App_Parameters(data, encoded=False) - header_list = [headers.PBAPType("x-bt/vcard")] + header_list = [headers.PBAPType(b"x-bt/vcard")] if application_parameters.data: header_list.append(application_parameters) From 1830ed2508ddd6a13025fb0549f1e962a6cd462a Mon Sep 17 00:00:00 2001 From: Noam Rathaus Date: Sun, 9 Apr 2023 18:18:19 +0300 Subject: [PATCH 12/12] Add requirements file --- requirements.txt | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 requirements.txt diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e8def7b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +git+https://github.com/pybluez/pybluez +cmd2 +git+https://github.com/BlackLight/PyOBEX.git