in source/consumer/lambda_handler.py [0:0]
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)