def process_face_search()

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

• 60 lines of code
• 15 McCabe index (conditional complexity)

def process_face_search(asset, workflow, results):
    metadata = json.loads(results)
    es = connect_es(es_endpoint)

    extracted_items = []
    if isinstance(metadata, list):
        for page in metadata:
            if "Persons" in page:
                for item in page["Persons"]:
                    item["Operator"] = "face_search"
                    item["Workflow"] = workflow
                    # flatten person key
                    item["PersonIndex"] = item["Person"]["Index"]
                    if "BoundingBox" in item["Person"]:
                        item["PersonBoundingBox"] = item["Person"]["BoundingBox"]
                    # flatten face key
                    if "Face" in item["Person"]:
                        item["FaceBoundingBox"] = item["Person"]["Face"]["BoundingBox"]
                        item["FaceLandmarks"] = item["Person"]["Face"]["Landmarks"]
                        item["FacePose"] = item["Person"]["Face"]["Pose"]
                        item["FaceQuality"] = item["Person"]["Face"]["Quality"]
                        confidence = item["Person"]["Face"]["Confidence"]
                        item["Confidence"] = confidence

                    if "FaceMatches" in item:
                        item["ContainsKnownFace"] = True
                        # flatten face matches key
                        for face in item["FaceMatches"]:
                            item["KnownFaceSimilarity"] = face["Similarity"]
                            item["MatchingKnownFaceId"] = face["Face"]["FaceId"]
                            item["KnownFaceBoundingBox"] = face["Face"]["BoundingBox"]
                            item["ImageId"] = face["Face"]["ImageId"]
                        del item["FaceMatches"]
                    else:
                        item["ContainsKnownFace"] = False
                    del item["Person"]

                    extracted_items.append(item)

    else:
        if "Persons" in metadata:
            for item in metadata["Persons"]:
                item["Operator"] = "face_search"
                item["Workflow"] = workflow
                # flatten person key
                item["PersonIndex"] = item["Person"]["Index"]
                if "BoundingBox" in item["Person"]:
                    item["PersonBoundingBox"] = item["Person"]["BoundingBox"]
                #flatten face key
                if "Face" in item["Person"]:
                    item["FaceBoundingBox"] = item["Person"]["Face"]["BoundingBox"]
                    item["FaceLandmarks"] = item["Person"]["Face"]["Landmarks"]
                    item["FacePose"] = item["Person"]["Face"]["Pose"]
                    item["FaceQuality"] = item["Person"]["Face"]["Quality"]
                    confidence = item["Person"]["Face"]["Confidence"]
                    item["Confidence"] = confidence

                if "FaceMatches" in item:
                    item["ContainsKnownFace"] = True
                    # flatten face matches key
                    for face in item["FaceMatches"]:
                        item["KnownFaceSimilarity"] = face["Similarity"]
                        item["MatchingKnownFaceId"] = face["Face"]["FaceId"]
                        item["KnownFaceBoundingBox"] = face["Face"]["BoundingBox"]
                        item["ImageId"] = face["Face"]["ImageId"]
                    del item["FaceMatches"]
                else:
                    item["ContainsKnownFace"] = False
                del item["Person"]

                extracted_items.append(item)

    bulk_index(es, asset, "face_search", extracted_items)