def __call__()

in obfuscator-cli/obfuscation_manager.py [0:0]

• 55 lines of code
• 9 McCabe index (conditional complexity)

    def __call__(self, dump_directory: Path, to: Optional[Path],
                 keep_all_fields: bool, dictionary_out: Path,
                 dictionary_path: Optional[Path]):
        # output dump directory validation
        if not to:
            if query_yes_no('Parameter --to was not provided. Patched files '
                            'will override the dump'):
                to = dump_directory
            else:
                _LOG.error('Aborting')
                sys.exit(1)
        _LOG.info(f'Obfuscated dump will be places to: "{to}"')

        # Loading desired values dictionary
        if dictionary_path:
            _LOG.info('Loading dictionary')
            try:
                with open(dictionary_path, 'r') as file:
                    dictionary = json.load(file)
                if not isinstance(dictionary, dict):
                    raise ValueError('The content must be a dict')
            except Exception as e:
                _LOG.error(f'Could not load {dictionary_path}: {e}')
                sys.exit(1)
        else:
            _LOG.info('Dictionary was not provided. All the '
                      'aliases will be randomly generated')
            dictionary = {}

        # Logging whether all the fields will be kept
        if keep_all_fields:
            _LOG.warning('All the fields will be kept')
        else:
            _LOG.info('Only id fields will be kept')

        # obfuscating
        out = {}  # here we will put real names to our randomly generated
        for path, content in self.yield_jsons(dump_directory):
            if self.is_findings(content):
                _LOG.info(f'Findings found by path: {path}. Pathing')
                content = cast(dict, content)  # is_findings ensures it's dict
                self.patch_findings(content, keep_all_fields, dictionary, out)
            elif self.is_list_of_shard_parts(content):
                _LOG.info(f'List of shard parts found by path: {path}. '
                          f'Pathing')
                content = cast(list, content)
                self.patch_list_of_shard_parts(content, keep_all_fields,
                                               dictionary, out)
            elif self.is_list_of_resources(content):
                _LOG.info(f'List of resources found by path: {path}. Pathing')
                content = cast(list, content)
                self.patch_list_of_resources(content, keep_all_fields,
                                             dictionary, out)
            else:
                _LOG.warning(f'Unknown file format: {path}. Skipping')
            self.dump_json(to / path, content)

        # dumping output dict
        _LOG.info(f'Output dictionary will be dumped to {dictionary_out}')
        dictionary_out.parent.mkdir(parents=True, exist_ok=True)
        flip_dict(out)
        with open(dictionary_out, 'w') as file:
            json.dump(out, file, indent=2)
        _LOG.info('Finished!')