def process_celebrity_detection()

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

• 79 lines of code
• 33 McCabe index (conditional complexity)

def process_celebrity_detection(asset, workflow, results):
    metadata = json.loads(results)
    es = connect_es(es_endpoint)
    extracted_items = []
    if isinstance(metadata, list):
        for page in metadata:
            # Parse schema for videos:
            # https://docs.aws.amazon.com/rekognition/latest/dg/celebrities-video-sqs.html
            if "Celebrities" in page:
                for item in page["Celebrities"]:
                    try:
                        item["Operator"] = "celebrity_detection"
                        item["Workflow"] = workflow
                        if "Celebrity" in item:
                            # flatten the inner Celebrity array
                            item["Name"] = item["Celebrity"]["Name"]
                            item["Confidence"] = item["Celebrity"]["Confidence"]
                            # Bounding box can be around body or face. Prefer body.
                            bounding_box = ''
                            if 'BoundingBox' in item["Celebrity"]:
                                bounding_box = item["Celebrity"]["BoundingBox"]
                            elif 'BoundingBox' in item["Celebrity"]["Face"]:
                                bounding_box = item["Celebrity"]["Face"]["BoundingBox"]
                            item["BoundingBox"] = bounding_box
                            # Set IMDB URL if it exists.
                            url=''
                            if item["Celebrity"]["Urls"]:
                                url = item["Celebrity"]["Urls"][0]
                            item['URL'] = url
                            # delete flattened array
                            del item["Celebrity"]
                        extracted_items.append(item)
                    except KeyError as e:
                        print("KeyError: " + str(e))
                        print("Item: " + json.dumps(item))
            # Parse schema for images:
            # https://docs.aws.amazon.com/rekognition/latest/dg/API_RecognizeCelebrities.html
            if "CelebrityFaces" in page:
                for item in page["CelebrityFaces"]:
                    try:
                        item["Operator"] = "celebrity_detection"
                        item["Workflow"] = workflow
                        if "Face" in item:
                            # flatten the inner Face array
                            item["Confidence"] = item["Face"]["Confidence"]
                            item["BoundingBox"] = item["Face"]["BoundingBox"]
                            # delete flattened array
                            del item["Face"]
                        extracted_items.append(item)
                    except KeyError as e:
                        print("KeyError: " + str(e))
                        print("Item: " + json.dumps(item))
    else:
        # Parse schema for videos:
        if "Celebrities" in metadata:
            for item in metadata["Celebrities"]:
                try:
                    item["Operator"] = "celebrity_detection"
                    item["Workflow"] = workflow
                    if "Celebrity" in item:
                        # flatten the inner Celebrity array
                        item["Name"] = item["Celebrity"]["Name"]
                        item["Confidence"] = item["Celebrity"]["Confidence"]
                        # Bounding box can be around body or face. Prefer body.
                        item["BoundingBox"] = ''
                        if 'BoundingBox' in item["Celebrity"]:
                            item["BoundingBox"] = item["Celebrity"]["BoundingBox"]
                        elif 'BoundingBox' in item["Celebrity"]["Face"]:
                            item["BoundingBox"] = item["Celebrity"]["Face"]["BoundingBox"]
                        # Set IMDB URL if it exists.
                        url=''
                        if item["Celebrity"]["Urls"]:
                            url = item["Celebrity"]["Urls"][0]
                        item['URL'] = url
                        # delete flattened array
                        del item["Celebrity"]
                    extracted_items.append(item)
                except KeyError as e:
                    print("KeyError: " + str(e))
                    print("Item: " + json.dumps(item))
        # Parse schema for images:
        if "CelebrityFaces" in metadata:
            for item in metadata["CelebrityFaces"]:
                try:
                    item["Operator"] = "celebrity_detection"
                    item["Workflow"] = workflow
                    if "Face" in item:
                        # flatten the inner Face array
                        item["Confidence"] = item["Face"]["Confidence"]
                        item["BoundingBox"] = item["Face"]["BoundingBox"]
                        # delete flattened array
                        del item["Face"]
                    extracted_items.append(item)
                except KeyError as e:
                    print("KeyError: " + str(e))
                    print("Item: " + json.dumps(item))
    bulk_index(es, asset, "celebrity_detection", extracted_items)