kangda-robot-backend/ruoyi-fastapi-backend/module_admin/service/compreface_service.py

from typing import Dict, List, Optional

from utils.compreface_util import ComprefaceUtil
from config.env import ComprefaceConfig

POSE_MAX_ANGLE = getattr(ComprefaceConfig, "COMPREFACE_POSE_MAX_ANGLE", 10)
MIN_FACE_WIDTH = getattr(ComprefaceConfig, "COMPREFACE_MIN_FACE_WIDTH", 200)
MIN_FACE_HEIGHT = getattr(ComprefaceConfig, "COMPREFACE_MIN_FACE_HEIGHT", 200)
MIN_DETECTION_PROBABILITY = getattr(ComprefaceConfig, "COMPREFACE_MIN_DETECTION_PROBABILITY", 0.9)

class ComprefaceService(ComprefaceUtil):
    
    """compreface服务类"""

    @classmethod
    def _is_pose_front(cls, pose: Dict | None) -> bool:
        if not pose:
            return False
        return all(abs(float(pose.get(axis, 0))) <= POSE_MAX_ANGLE for axis in ("pitch", "roll", "yaw"))

    @classmethod
    def _is_face_close(cls, box: Dict | None) -> bool:
        if not box:
            return False
        width = float(box.get("x_max", 0)) - float(box.get("x_min", 0))
        height = float(box.get("y_max", 0)) - float(box.get("y_min", 0))
        probability = float(box.get("probability", 0))
        return (
            width >= MIN_FACE_WIDTH
            and height >= MIN_FACE_HEIGHT
            and probability >= MIN_DETECTION_PROBABILITY
        )

    @classmethod
    def _select_valid_detection(cls, detections: List[Dict]) -> Optional[Dict]:
        for detection in detections:
            pose = detection.get("pose")
            box = detection.get("box")
            if cls._is_pose_front(pose) and cls._is_face_close(box):
                return detection
        return None

    @staticmethod
    def _pending_response() -> Dict[str, str]:
        return {
            "name": "未识别",
            "role": "等待正脸"
        }
    
    # 人脸检测
    @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.get("result", [])
    
    # 人脸识别
    @classmethod
    async def face_recognition_service(cls, image, options: dict = {}):
        """ 人脸识别

        Args:
            image_path (str): 图片路径
            options (dict, optional): 参数. Defaults to {}.

        Returns:
            [type]: [description]
        """
        detections = await cls.face_detection_service(image)
        detection = cls._select_valid_detection(detections)
        if not detection:
            return cls._pending_response()

        result = await ComprefaceUtil.face_recognition(image, options)
        recognition_list = result.get('result') or []
        if not recognition_list:
            return cls._pending_response()

        subjects = recognition_list[0].get('subjects') or []
        if not subjects:
            return cls._pending_response()

        subject = subjects[0]
        name = subject.get('subject', '')
        similarity = subject.get('similarity', 0)
        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": []
            }