def create_wellboretop_manifest_from_path()

in src/loading_manifest/osdu_wellboretop_manifest.py [0:0]

• 54 lines of code
• 10 McCabe index (conditional complexity)

def create_wellboretop_manifest_from_path(
        input_path, output_path,
        preload_file_path, file_source,
        schema_ns_name, schema_ns_value,
        acl, legal, schema_version, dict_schemas):

    # check supported schema version
    if schema_version not in SUPPORTED_SCHEMA_VERSIONS:
        logging.error("Schema version %s is not in the supported list: %s", schema_version,
                      SUPPORTED_SCHEMA_VERSIONS)
        logging.info("Generated 0 wellboretop load manifests.")
        return

    # list top filenames
    files = os.listdir(input_path)
    files_wellboretop = fnmatch.filter(files, "*.csv")

    valid_files = []
    logging.info("Checking {} files".format(len(files_wellboretop)))
    for file_wellboretop in sorted(files_wellboretop):
        file_wellboretop = file_wellboretop.strip()
        # minimum check: file and size
        if cm.is_nonzero_file(input_path, file_wellboretop):
            valid_files.append(file_wellboretop)

    processed_files = []
    logging.info("Processing {} files".format(len(valid_files)))
    for valid_file in valid_files:
        full_valid_file_path = os.path.join(input_path, valid_file)

        # retrieve wellbore id
        # osdu
        wellbore_id = valid_file[:-4]
        # try:
        #     int(wellbore_id)
        # except ValueError:
        #     logging.warning("Skip Invalid File:" + valid_file)
        #     continue

        wellboretop_file = valid_file
        output_file = os.path.join(output_path,
                                   "load_top_" + schema_version + "_"
                                   + wellboretop_file.replace(".", "_") + ".json")
        try:
            # retrieve markers
            top_markers = read_markers_from_csv_file(full_valid_file_path)
            if top_markers is None or len(top_markers) == 0:
                logging.warning("Markers not found for: " + full_valid_file_path)

            with open(output_file, "w") as f:
                json.dump(
                    obj=create_wellboretop_manifest(
                        wellboretop_name=wellboretop_file,
                        preload_file_path=preload_file_path,
                        file_source=file_source,
                        wellbore_id=wellbore_id,
                        top_markers=top_markers,
                        schema_ns_name=schema_ns_name,
                        schema_ns_value=schema_ns_value,
                        acl=acl,
                        legal=legal,
                        schema_version=schema_version,
                        dict_schemas=dict_schemas
                    ),
                    fp=f,
                    indent=4
                )
                processed_files.append(wellboretop_file)
        except Exception:
            logging.exception("Unable to process wellboretop file: {}".format(wellboretop_file))
            os.remove(output_file)

    logging.info("Generated {} wellboretop load manifests.".format(len(processed_files)))