131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
from utils.compreface_util import ComprefaceUtil
|
||
from config.env import ComprefaceConfig
|
||
|
||
class ComprefaceService(ComprefaceUtil):
|
||
|
||
"""compreface服务类"""
|
||
|
||
# 人脸检测
|
||
@classmethod
|
||
async def face_detection_service(cls, image, options: dict = {}):
|
||
""" 人脸检测
|
||
|
||
Args:
|
||
image_path (str): 图片路径
|
||
options (dict, optional): 检测参数. Defaults to {}.
|
||
|
||
Returns:
|
||
[type]: [description]
|
||
"""
|
||
options={
|
||
"limit": 0,
|
||
"det_prob_threshold": 0.8,
|
||
"prediction_count": 1,
|
||
# 可选参数 age,gender,landmarks,calculator
|
||
"face_plugins": "pose",
|
||
"status": "false",
|
||
}
|
||
result = await ComprefaceUtil.face_detection(image, options)
|
||
return result["result"]
|
||
|
||
# 人脸识别
|
||
@classmethod
|
||
async def face_recognition_service(cls, image, options: dict = {}):
|
||
""" 人脸识别
|
||
|
||
Args:
|
||
image_path (str): 图片路径
|
||
options (dict, optional): 参数. Defaults to {}.
|
||
|
||
Returns:
|
||
[type]: [description]
|
||
"""
|
||
result = await ComprefaceUtil.face_recognition(image, options)
|
||
subject = result.get('result')[0].get('subjects')[0]
|
||
name = subject.get('subject')
|
||
similarity = subject.get('similarity')
|
||
if similarity < ComprefaceConfig.COMPREFACE_SIMILARITY_THRESHOLD:
|
||
return {
|
||
"name": "未知",
|
||
"role": "陌生人"
|
||
}
|
||
else:
|
||
t = name.split("_")
|
||
return {
|
||
"name": t[0],
|
||
"role": t[1] if len(t) > 1 else "员工"
|
||
}
|
||
|
||
# 添加图像到人脸库
|
||
@classmethod
|
||
async def face_addition_service(cls, image, name):
|
||
result = await ComprefaceUtil.face_addition(image, name)
|
||
return result
|
||
|
||
# 批量添加图像到人脸库
|
||
@classmethod
|
||
async def face_addition_batch_service(cls, t):
|
||
result_list = []
|
||
for image, name in t:
|
||
result = await ComprefaceUtil.face_addition(image, name)
|
||
result_list.append(result)
|
||
return result_list
|
||
|
||
# 删除过期的访客人脸
|
||
@classmethod
|
||
async def delete_expired_visitor_faces_service(cls, active_visitor_names: list):
|
||
"""删除CompreFace中已过期的访客人脸数据
|
||
|
||
Args:
|
||
active_visitor_names (list): 当前有效的访客名字列表
|
||
|
||
Returns:
|
||
dict: 包含删除成功和失败的subject列表
|
||
"""
|
||
try:
|
||
# 获取CompreFace中所有的subjects
|
||
all_subjects_result = await ComprefaceUtil.get_all_subjects()
|
||
all_subjects = all_subjects_result.get('subjects', [])
|
||
|
||
deleted_subjects = []
|
||
failed_subjects = []
|
||
|
||
# 遍历所有subjects，删除不在active_visitor_names中的访客
|
||
for subject in all_subjects:
|
||
# 假设访客的subject格式为 "姓名_访客" 或 "姓名"
|
||
# 只删除访客类型的人脸，员工人脸不删除
|
||
if subject not in active_visitor_names:
|
||
# 如果subject不在活跃访客列表中，且不是员工（员工一般包含"_员工"等标识）
|
||
# 这里假设访客没有特殊后缀，或者可以根据实际情况调整判断逻辑
|
||
# 如果subject包含"_员工"等标识，则跳过
|
||
if "_员工" in subject or "_职员" in subject or "_工作人员" in subject:
|
||
continue
|
||
|
||
try:
|
||
# 删除该subject
|
||
await ComprefaceUtil.delete_subject(subject)
|
||
deleted_subjects.append(subject)
|
||
except Exception as e:
|
||
failed_subjects.append({"subject": subject, "error": str(e)})
|
||
|
||
return {
|
||
"deleted_count": len(deleted_subjects),
|
||
"deleted_subjects": deleted_subjects,
|
||
"failed_count": len(failed_subjects),
|
||
"failed_subjects": failed_subjects
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"error": f"删除过期访客失败: {str(e)}",
|
||
"deleted_count": 0,
|
||
"deleted_subjects": [],
|
||
"failed_count": 0,
|
||
"failed_subjects": []
|
||
}
|
||
|
||
|
||
|
||
|
||
|
||
|