def process_generic_data()

in source/consumer/lambda_handler.py [0:0]

• 58 lines of code
• 19 McCabe index (conditional complexity)

def process_generic_data(asset, workflow, results):
    # This function puts generic data in Elasticsearch.
    metadata = json.loads(results)
    es = connect_es(es_endpoint)
    extracted_items = []
    # We can tell if json results are paged by checking to see if the json results are an instance of the list type.
    if isinstance(metadata, list):
        # handle paged results
        for page in metadata:
            if "Labels" in page:
                for item in page["Labels"]:
                    try:
                        item["Operator"] = "generic_data_lookup"
                        item["Workflow"] = workflow
                        if "Label" in item:
                            # Flatten the inner Label array
                            item["Confidence"] = float(item["Label"]["Confidence"])*100
                            item["Name"] = item["Label"]["Name"]
                            item["Instances"] = ''
                            if 'Instances' in item["Label"]:
                                for box in item["Label"]["Instances"]:
                                    box["BoundingBox"]["Height"] = float(box["BoundingBox"]["Height"]) / 720
                                    box["BoundingBox"]["Top"] = float(box["BoundingBox"]["Top"]) / 720
                                    box["BoundingBox"]["Left"] = float(box["BoundingBox"]["Left"]) / 1280
                                    box["BoundingBox"]["Width"] = float(box["BoundingBox"]["Width"]) / 1280
                                    box["Confidence"] = float(box["Confidence"])*100
                                item["Instances"] = item["Label"]["Instances"]
                            item["Parents"] = ''
                            if 'Parents' in item["Label"]:
                                item["Parents"] = item["Label"]["Parents"]
                            # Delete the flattened array
                            del item["Label"]
                        extracted_items.append(item)
                    except KeyError as e:
                        print("KeyError: " + str(e))
                        print("Item: " + json.dumps(item))
    else:
        # these results are not paged
        if "Labels" in metadata:
            for item in metadata["Labels"]:
                try:
                    item["Operator"] = "generic_data_lookup"
                    item["Workflow"] = workflow
                    if "Label" in item:
                        # Flatten the inner Label array
                        item["Confidence"] = float(item["Label"]["Confidence"])*100
                        item["Name"] = item["Label"]["Name"]
                        item["Instances"] = ''
                        if 'Instances' in item["Label"]:
                            for box in item["Label"]["Instances"]:
                                box["BoundingBox"]["Height"] = float(box["BoundingBox"]["Height"]) / 720
                                box["BoundingBox"]["Top"] = float(box["BoundingBox"]["Top"]) / 720
                                box["BoundingBox"]["Left"] = float(box["BoundingBox"]["Left"]) / 1280
                                box["BoundingBox"]["Width"] = float(box["BoundingBox"]["Width"]) / 1280
                                box["Confidence"] = float(box["Confidence"])*100
                            item["Instances"] = item["Label"]["Instances"]
                        item["Parents"] = ''
                        if 'Parents' in item["Label"]:
                            item["Parents"] = item["Label"]["Parents"]
                        # Delete the flattened array
                        del item["Label"]
                    extracted_items.append(item)
                except KeyError as e:
                    print("KeyError: " + str(e))
                    print("Item: " + json.dumps(item))
    bulk_index(es, asset, "labels", extracted_items)