picpocket / face_analyzer.py
chawin.chen
fix: 五官辅助分析失败不影响主评分并兼容旧版 App
3558500
Raw History Blame Contribute Delete
60.1 kB
import os
import random
import time
from typing import List, Dict, Any, Optional
import cv2
import numpy as np
import config
from config import logger, MODELS_PATH, OUTPUT_DIR, DEEPFACE_AVAILABLE, \
YOLO_AVAILABLE
from facial_analyzer import FacialFeatureAnalyzer
from models import ModelType
from utils import save_image_high_quality, get_date_subfolder
try:
from tensorflow.keras import backend as keras_backend
except ImportError:
try:
from tf_keras import backend as keras_backend # type: ignore
except ImportError:
keras_backend = None
if DEEPFACE_AVAILABLE:
from deepface import DeepFace
def _is_deepface_tensor_error(exc: Exception) -> bool:
"""判断 DeepFace 是否命中 Keras 3 / SymbolicTensor 兼容性问题。"""
message = str(exc)
return "SymbolicTensor" in message or "'numpy'" in message or ".numpy" in message
def _is_deepface_cache_error(exc: Exception) -> bool:
"""判断 DeepFace 模型缓存结构是否被破坏。"""
message = str(exc)
return "facial_attribute" in message or "facial_recognition" in message
def _ensure_deepface_modeling_cache() -> None:
"""恢复新版 DeepFace modeling.cached_models 的顶层分组结构。"""
try:
from deepface.modules import modeling
except Exception:
return
cached_models = getattr(modeling, "cached_models", None)
if not isinstance(cached_models, dict):
return
available_models = getattr(modeling, "AVAILABLE_MODELS", None)
group_names = (
tuple(available_models.keys())
if isinstance(available_models, dict)
else ("facial_recognition", "spoofing", "facial_attribute", "face_detector")
)
for group_name in group_names:
group_cache = cached_models.get(group_name)
if not isinstance(group_cache, dict):
cached_models[group_name] = {}
def _clear_deepface_cache(cache: Dict[str, Any], cache_name: str) -> None:
"""清理 DeepFace 缓存内容,同时保留新版缓存的顶层分组 key。"""
try:
if cache_name.endswith("cached_models"):
group_names = tuple(cache.keys()) or (
"facial_recognition",
"spoofing",
"facial_attribute",
"face_detector",
)
for group_name in group_names:
group_cache = cache.setdefault(group_name, {})
if isinstance(group_cache, dict):
group_cache.clear()
return
cache.clear()
except Exception as cache_exc:
logger.warning(f"清除DeepFace缓存失败: {cache_name}: {cache_exc}")
def _recover_deepface_state() -> None:
"""清理 DeepFace/TensorFlow 状态,让下次 analyze 重新走可用路径。"""
if keras_backend is not None:
try:
keras_backend.clear_session()
except Exception as clear_exc:
logger.warning(f"清理Keras会话失败: {clear_exc}")
for module_name in ("deepface.commons.functions", "deepface.modules.modeling"):
try:
module = __import__(module_name, fromlist=["*"])
except Exception:
continue
for attr_name in ("models", "model_cache", "built_models", "cached_models"):
cache = getattr(module, attr_name, None)
if isinstance(cache, dict):
cache_name = f"{module_name}.{attr_name}"
_clear_deepface_cache(cache, cache_name)
logger.info(f"已清除DeepFace缓存: {cache_name}")
_ensure_deepface_modeling_cache()
def _patch_deepface_build_model() -> None:
"""包住 DeepFace build_model,避免 cached_models 顶层分组缺失导致 KeyError。"""
try:
from deepface.modules import modeling
except Exception:
return
original_build_model = getattr(modeling, "build_model", None)
if original_build_model is None or getattr(original_build_model, "_picpocket_patched", False):
return
def _patched_build_model(task: str, model_name: str) -> Any:
_ensure_deepface_modeling_cache()
try:
return original_build_model(task, model_name)
except KeyError as key_err:
if str(key_err).strip("'\"") != task:
raise
logger.warning(f"DeepFace build_model 缓存缺少 {task},恢复缓存结构后重试")
_ensure_deepface_modeling_cache()
return original_build_model(task, model_name)
_patched_build_model._picpocket_patched = True # type: ignore[attr-defined]
modeling.build_model = _patched_build_model
def _patch_deepface_demography_predict_internal() -> None:
"""补 DeepFace Demography._predict_internal 的 Keras 3 SymbolicTensor 兼容。"""
try:
from deepface.models import Demography as df_demography
except Exception:
return
demography_base = getattr(df_demography, "Demography", None)
original_predict_internal = getattr(demography_base, "_predict_internal", None)
if (
demography_base is None
or original_predict_internal is None
or getattr(original_predict_internal, "_picpocket_patched", False)
):
return
def _patched_predict_internal(self, img_batch: np.ndarray) -> np.ndarray:
try:
return original_predict_internal(self, img_batch)
except Exception as exc:
if not _is_deepface_tensor_error(exc):
raise
logger.warning(f"DeepFace demography _predict_internal 触发 Tensor 兼容异常,改用 predict: {exc}")
if img_batch.ndim != 4:
raise AssertionError(f"expected 4-dimensional tensor input, got {img_batch.shape}")
predict_fn = getattr(self.model, "predict", None)
if predict_fn is not None:
predictions = predict_fn(img_batch, verbose=0)
else:
predictions = self.model.predict_on_batch(img_batch)
predictions = np.asarray(predictions)
if predictions.ndim == 0:
raise ValueError("DeepFace demography prediction output is empty.")
if img_batch.shape[0] == 1 and predictions.ndim > 1:
return predictions[0, :]
return predictions
_patched_predict_internal._picpocket_patched = True # type: ignore[attr-defined]
demography_base._predict_internal = _patched_predict_internal
def _patch_deepface_runtime_compat() -> None:
"""应用 DeepFace 运行期兼容补丁。"""
if not DEEPFACE_AVAILABLE:
return
_ensure_deepface_modeling_cache()
_patch_deepface_build_model()
_patch_deepface_demography_predict_internal()
def _deepface_analyze_with_recovery(**kwargs: Any) -> Any:
"""执行 DeepFace.analyze;遇到 SymbolicTensor/.numpy 兼容错误时恢复后重试一次。"""
_patch_deepface_runtime_compat()
try:
return DeepFace.analyze(**kwargs)
except Exception as exc:
if not (_is_deepface_tensor_error(exc) or _is_deepface_cache_error(exc)):
raise
logger.warning(f"DeepFace analyze 遇到 Tensor/缓存兼容异常,恢复后重试: {exc}")
_recover_deepface_state()
_patch_deepface_runtime_compat()
return DeepFace.analyze(**kwargs)
_patch_deepface_runtime_compat()
# 可选导入 YOLO
if YOLO_AVAILABLE:
try:
from ultralytics import YOLO
YOLO_AVAILABLE = True
except ImportError:
YOLO_AVAILABLE = False
YOLO = None
print("Warning: ENABLE_YOLO=true but ultralytics not available")
class EnhancedFaceAnalyzer:
"""增强版人脸分析器 - 支持混合模型"""
def __init__(self, models_dir: str = MODELS_PATH):
"""
初始化人脸分析器
:param models_dir: 模型文件目录
"""
start_time = time.perf_counter()
self.models_dir = models_dir
self.MODEL_MEAN_VALUES = (104, 117, 123)
self.age_list = [
"(0-2)",
"(4-6)",
"(8-12)",
"(15-20)",
"(25-32)",
"(38-43)",
"(48-53)",
"(60-100)",
]
self.gender_list = ["Male", "Female"]
# 性别对应的颜色 (BGR格式)
self.gender_colors = {
"Male": (255, 165, 0), # 橙色 Orange
"Female": (255, 0, 255), # 洋红 Magenta / Fuchsia
}
# 初始化五官分析器
self.facial_analyzer = FacialFeatureAnalyzer()
# 加载HowCuteAmI模型
self._load_howcuteami_models()
# 加载YOLOv人脸检测模型
self._load_yolo_model()
# 预热模型(可选,通过配置开关)
if getattr(config, "ENABLE_WARMUP", False):
self._warmup_models()
init_time = time.perf_counter() - start_time
logger.info(f"EnhancedFaceAnalyzer initialized successfully, time: {init_time:.3f}s")
def _cap_conf(self, value: float) -> float:
"""将置信度限制在 [0, 0.9999] 并保留4位小数。"""
try:
v = float(value if value is not None else 0.0)
except Exception:
v = 0.0
if v >= 1.0:
v = 0.9999
if v < 0.0:
v = 0.0
return round(v, 4)
def _adjust_beauty_score(self, score: float, scale: float = 10.0) -> float:
try:
if not config.BEAUTY_ADJUST_ENABLED:
return score
# 读取提分区间与力度
low = float(getattr(config, "BEAUTY_ADJUST_MIN", 6.0))
high = float(getattr(config, "BEAUTY_ADJUST_MAX", getattr(config, "BEAUTY_ADJUST_THRESHOLD", 8.0)))
# 如果是百分制,则区间也需要相应放大
if scale == 100.0:
low *= 10
high *= 10
gamma = float(getattr(config, "BEAUTY_ADJUST_GAMMA", 0.3))
gamma = max(0.0001, min(1.0, gamma))
# 区间有效性保护
if not (0.0 <= low < high <= scale):
return score
# 低于下限不提分,区间内提向上限,高于上限不变
if score < low:
return score
if score < high:
# 向上限 high 进行温和靠拢:adjusted = high - gamma * (high - score)
adjusted = high - gamma * (high - score)
# 十分制保留1位小数,百分制保留2位小数
precision = 1 if scale == 10.0 else 2
adjusted = round(min(scale, max(0.0, adjusted)), precision)
try:
logger.info(
f"beauty_score ({scale}) adjusted: original={score:.2f} -> adjusted={adjusted:.2f} "
f"(range=[{low:.2f},{high:.2f}], gamma={gamma:.3f})"
)
except Exception:
pass
return adjusted
return score
except Exception:
return score
def _load_yolo_model(self):
"""加载YOLOv人脸检测模型"""
self.yolo_model = None
if config.YOLO_AVAILABLE:
try:
# 尝试加载本地YOLOv人脸模型
yolo_face_path = os.path.join(self.models_dir, config.YOLO_MODEL)
if os.path.exists(yolo_face_path):
self.yolo_model = YOLO(yolo_face_path)
logger.info(f"Local YOLO face model loaded successfully: {yolo_face_path}")
else:
# 如果本地没有,尝试在线下载(第一次使用时)
logger.info("Local YOLO face model does not exist, attempting to download...")
try:
# 检查是否是yolov8,使用相应的模型
model_name = "yolov11n-face.pt" # 默认使用yolov8n
self.yolo_model = YOLO(model_name)
logger.info(
f"YOLOv8 general model loaded successfully (detecting 'person' class as face regions)"
)
except Exception as e:
logger.warning(f"YOLOv model download failed: {e}")
except Exception as e:
logger.error(f"YOLOv model loading failed: {e}")
else:
logger.warning("ultralytics not installed, cannot use YOLOv")
def _load_howcuteami_models(self):
"""加载HowCuteAmI深度学习模型"""
try:
# 人脸检测模型
face_proto = os.path.join(self.models_dir, "opencv_face_detector.pbtxt")
face_model = os.path.join(self.models_dir, "opencv_face_detector_uint8.pb")
self.face_net = cv2.dnn.readNet(face_model, face_proto)
# 年龄预测模型
age_proto = os.path.join(self.models_dir, "age_googlenet.prototxt")
age_model = os.path.join(self.models_dir, "age_googlenet.caffemodel")
self.age_net = cv2.dnn.readNet(age_model, age_proto)
# 性别预测模型
gender_proto = os.path.join(self.models_dir, "gender_googlenet.prototxt")
gender_model = os.path.join(self.models_dir, "gender_googlenet.caffemodel")
self.gender_net = cv2.dnn.readNet(gender_model, gender_proto)
# 颜值预测模型
beauty_proto = os.path.join(self.models_dir, "beauty_resnet.prototxt")
beauty_model = os.path.join(self.models_dir, "beauty_resnet.caffemodel")
self.beauty_net = cv2.dnn.readNet(beauty_model, beauty_proto)
logger.info("HowCuteAmI model loaded successfully!")
except Exception as e:
logger.error(f"HowCuteAmI model loading failed: {e}")
raise e
# 人脸检测方法
def _detect_faces(
self, frame: np.ndarray, conf_threshold: float = config.FACE_CONFIDENCE
) -> List[List[int]]:
"""
使用YOLO进行人脸检测,如果失败则回退到OpenCV DNN
"""
# 优先使用YOLO
face_boxes = []
if self.yolo_model is not None:
try:
results = self.yolo_model(frame, conf=conf_threshold, verbose=False)
for result in results:
boxes = result.boxes
if boxes is not None:
for box in boxes:
# 检查类别ID (如果是专门的人脸模型,通常是0;如果是通用模型,person类别通常是0)
class_id = int(box.cls[0])
# 获取边界框坐标 (xyxy格式)
x1, y1, x2, y2 = box.xyxy[0].cpu().numpy().astype(int)
confidence = float(box.conf[0])
logger.info(
f"detect class_id={class_id}, confidence={confidence}"
)
# 基本边界检查
frame_height, frame_width = frame.shape[:2]
x1 = max(0, int(x1))
y1 = max(0, int(y1))
x2 = min(frame_width, int(x2))
y2 = min(frame_height, int(y2))
# 过滤太小的检测框
width, height = x2 - x1, y2 - y1
if (
width > 30 and height > 30
): # YOLO通常检测精度更高,可以稍微提高最小尺寸
# 如果使用通用模型检测person,需要进一步过滤头部区域
if self._is_likely_face_region(x1, y1, x2, y2, frame):
face_boxes.append(self._scale_box([x1, y1, x2, y2]))
logger.info(
f"YOLO detected {len(face_boxes)} faces, conf_threshold={conf_threshold}"
)
if face_boxes: # 如果YOLO检测到了人脸,直接返回
return face_boxes
except Exception as e:
logger.warning(f"YOLO detection failed, falling back to OpenCV DNN: {e}")
return self._detect_faces_opencv_fallback(frame, conf_threshold)
return face_boxes
def _is_likely_face_region(
self, x1: int, y1: int, x2: int, y2: int, frame: np.ndarray
) -> bool:
"""
判断检测区域是否可能是人脸区域(当使用通用YOLO模型时)
"""
width, height = x2 - x1, y2 - y1
# 长宽比检查 - 人脸/头部通常接近正方形
aspect_ratio = width / height
if not (0.6 <= aspect_ratio <= 1.6):
return False
# 位置检查 - 人脸通常在图像上半部分(简单启发式)
frame_height = frame.shape[0]
center_y = (y1 + y2) / 2
if center_y > frame_height * 0.8: # 如果中心点在图像下方80%以下,可能不是人脸
return False
# 尺寸检查 - 不应该占据整个图像
frame_width, frame_height = frame.shape[1], frame.shape[0]
if width > frame_width * 0.8 or height > frame_height * 0.8:
return False
return True
def _detect_faces_opencv_fallback(
self, frame: np.ndarray, conf_threshold: float = 0.5
) -> List[List[int]]:
"""
优化版人脸检测 - 支持多尺度检测和小人脸识别
"""
frame_height, frame_width = frame.shape[:2]
all_boxes = []
# 多尺度检测配置 - 从小到大,更好地检测不同大小的人脸
detection_configs = [
{"size": (300, 300), "threshold": conf_threshold},
{
"size": (416, 416),
"threshold": max(0.3, conf_threshold - 0.2),
}, # 对大尺度降低阈值
{
"size": (512, 512),
"threshold": max(0.25, conf_threshold - 0.25),
}, # 进一步降低阈值检测小脸
]
logger.info(f"Detecting faces using opencv, conf_threshold={conf_threshold}")
for config in detection_configs:
try:
# 图像预处理 - 增强对比度有助于小人脸检测
processed_frame = cv2.convertScaleAbs(frame, alpha=1.1, beta=10)
blob = cv2.dnn.blobFromImage(
processed_frame, 1.0, config["size"], [104, 117, 123], True, False
)
self.face_net.setInput(blob)
detections = self.face_net.forward()
# 提取检测结果
for i in range(detections.shape[2]):
confidence = detections[0, 0, i, 2]
if confidence > config["threshold"]:
x1 = int(detections[0, 0, i, 3] * frame_width)
y1 = int(detections[0, 0, i, 4] * frame_height)
x2 = int(detections[0, 0, i, 5] * frame_width)
y2 = int(detections[0, 0, i, 6] * frame_height)
# 基本边界检查
x1, y1 = max(0, x1), max(0, y1)
x2, y2 = min(frame_width, x2), min(frame_height, y2)
# 过滤太小或不合理的检测框
width, height = x2 - x1, y2 - y1
if (
width > 20
and height > 20
and width < frame_width * 0.8
and height < frame_height * 0.8
):
# 长宽比检查 - 人脸通常接近正方形
aspect_ratio = width / height
if 0.6 <= aspect_ratio <= 1.8: # 允许一定的椭圆形变
all_boxes.append(
{
"box": [x1, y1, x2, y2],
"confidence": confidence,
"area": width * height,
}
)
except Exception as e:
logger.warning(f"Scale {config['size']} detection failed: {e}")
continue
# 如果没有检测到任何人脸,尝试更宽松的条件
if not all_boxes:
logger.info("No faces detected, trying more relaxed detection conditions...")
try:
# 最后一次尝试:最低阈值 + 图像增强
enhanced_frame = cv2.equalizeHist(
cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
)
enhanced_frame = cv2.cvtColor(enhanced_frame, cv2.COLOR_GRAY2BGR)
blob = cv2.dnn.blobFromImage(
enhanced_frame, 1.0, (300, 300), [104, 117, 123], True, False
)
self.face_net.setInput(blob)
detections = self.face_net.forward()
for i in range(detections.shape[2]):
confidence = detections[0, 0, i, 2]
if confidence > 0.15: # 非常低的阈值
x1 = int(detections[0, 0, i, 3] * frame_width)
y1 = int(detections[0, 0, i, 4] * frame_height)
x2 = int(detections[0, 0, i, 5] * frame_width)
y2 = int(detections[0, 0, i, 6] * frame_height)
x1, y1 = max(0, x1), max(0, y1)
x2, y2 = min(frame_width, x2), min(frame_height, y2)
width, height = x2 - x1, y2 - y1
if width > 15 and height > 15: # 更小的最小尺寸
aspect_ratio = width / height
if 0.5 <= aspect_ratio <= 2.0: # 更宽松的长宽比
all_boxes.append(
{
"box": [x1, y1, x2, y2],
"confidence": confidence,
"area": width * height,
}
)
except Exception as e:
logger.warning(f"Relaxed condition detection also failed: {e}")
# NMS (非极大值抑制) 去除重复检测
if all_boxes:
final_boxes = self._apply_nms(all_boxes, overlap_threshold=0.4)
return [self._scale_box(box["box"]) for box in final_boxes]
return []
def _apply_nms(
self, detections: List[Dict], overlap_threshold: float = 0.4
) -> List[Dict]:
"""
非极大值抑制,去除重复的检测框
"""
if not detections:
return []
# 按置信度排序
detections.sort(key=lambda x: x["confidence"], reverse=True)
keep = []
while detections:
# 保留置信度最高的
best = detections.pop(0)
keep.append(best)
# 移除与最佳检测重叠度高的其他检测
remaining = []
for det in detections:
if self._calculate_iou(best["box"], det["box"]) < overlap_threshold:
remaining.append(det)
detections = remaining
return keep
def _calculate_iou(self, box1: List[int], box2: List[int]) -> float:
"""
计算两个边界框的IoU (交并比)
"""
x1_1, y1_1, x2_1, y2_1 = box1
x1_2, y1_2, x2_2, y2_2 = box2
# 计算交集
x1_i = max(x1_1, x1_2)
y1_i = max(y1_1, y1_2)
x2_i = min(x2_1, x2_2)
y2_i = min(y2_1, y2_2)
if x2_i <= x1_i or y2_i <= y1_i:
return 0.0
intersection = (x2_i - x1_i) * (y2_i - y1_i)
# 计算并集
area1 = (x2_1 - x1_1) * (y2_1 - y1_1)
area2 = (x2_2 - x1_2) * (y2_2 - y1_2)
union = area1 + area2 - intersection
return intersection / union if union > 0 else 0.0
def _scale_box(self, box: List[int]) -> List[int]:
"""将矩形框缩放为正方形"""
width = box[2] - box[0]
height = box[3] - box[1]
maximum = max(width, height)
dx = int((maximum - width) / 2)
dy = int((maximum - height) / 2)
return [box[0] - dx, box[1] - dy, box[2] + dx, box[3] + dy]
def _crop_face(self, image: np.ndarray, box: List[int]) -> np.ndarray:
"""裁剪人脸区域"""
x1, y1, x2, y2 = box
h, w = image.shape[:2]
x1 = max(0, x1)
y1 = max(0, y1)
x2 = min(w, x2)
y2 = min(h, y2)
return image[y1:y2, x1:x2]
def _predict_beauty_gender_with_howcuteami(
self, face: np.ndarray
) -> Dict[str, Any]:
"""使用HowCuteAmI模型预测颜值和性别"""
try:
blob = cv2.dnn.blobFromImage(
face, 1.0, (224, 224), self.MODEL_MEAN_VALUES, swapRB=False
)
# 性别预测
self.gender_net.setInput(blob)
gender_preds = self.gender_net.forward()
gender = self.gender_list[gender_preds[0].argmax()]
gender_confidence = float(np.max(gender_preds[0]))
gender_confidence = self._cap_conf(gender_confidence)
# 年龄预测
self.age_net.setInput(blob)
age_preds = self.age_net.forward()
age = self.age_list[age_preds[0].argmax()]
age_confidence = float(np.max(age_preds[0]))
# 颜值预测
blob_beauty = cv2.dnn.blobFromImage(
face, 1.0 / 255, (224, 224), self.MODEL_MEAN_VALUES, swapRB=False
)
self.beauty_net.setInput(blob_beauty)
beauty_preds = self.beauty_net.forward()
raw_beauty_sum = float(np.sum(beauty_preds[0]))
# 十分制计算
beauty_score = round(float(2.0 * raw_beauty_sum), 1)
beauty_score = min(10.0, max(0.0, beauty_score))
beauty_score = self._adjust_beauty_score(beauty_score)
# 百分制计算
beauty_score_100 = float(20.0 * raw_beauty_sum)
beauty_score_100 = min(100.0, max(0.0, beauty_score_100))
beauty_score_100 = self._adjust_beauty_score(beauty_score_100, scale=100.0)
raw_score = raw_beauty_sum
return {
"age": age,
"age_confidence": round(age_confidence, 4),
"gender": gender,
"gender_confidence": gender_confidence,
"beauty_score": beauty_score_100,
"beauty_score_100": beauty_score_100,
"beauty_raw_score": round(raw_score, 4),
"age_model_used": "HowCuteAmI",
"gender_model_used": "HowCuteAmI",
"beauty_model_used": "HowCuteAmI",
}
except Exception as e:
logger.error(f"HowCuteAmI beauty gender prediction failed: {e}")
raise e
def _predict_age_emotion_with_deepface(
self, face_image: np.ndarray, include_emotion: bool = True,
initial_age_confidence: Optional[float] = None
) -> Dict[str, Any]:
"""使用DeepFace预测年龄、情绪(并返回可用的性别信息用于回退)"""
if not DEEPFACE_AVAILABLE:
# 如果DeepFace不可用,使用HowCuteAmI的年龄预测作为回退
return self._predict_age_with_howcuteami_fallback(face_image)
if face_image is None or face_image.size == 0:
raise ValueError("无效的人脸图像")
try:
actions = ["age", "gender", "emotion"] if include_emotion else ["age", "gender"]
# DeepFace分析 - 禁用进度条和详细输出
result = _deepface_analyze_with_recovery(
img_path=face_image,
actions=actions,
enforce_detection=False,
detector_backend="skip",
silent=True # 禁用进度条输出
)
# 处理结果 (DeepFace返回的结果格式可能是list或dict)
if isinstance(result, list):
result = result[0]
# 提取信息
age = result.get("age", 25)
if include_emotion:
emotion = result.get("dominant_emotion", "neutral")
emotion_scores = result.get("emotion", {}) or {}
else:
emotion = "neutral"
emotion_scores = {"neutral": 100.0}
# 性别信息(用于在HowCuteAmI置信度低时回退)
deep_gender = result.get("dominant_gender", "Woman")
deep_gender_conf = result.get("gender", {}).get(deep_gender, 50.0) / 100.0
deep_gender_conf = self._cap_conf(deep_gender_conf)
if str(deep_gender).lower() in ["woman", "female"]:
deep_gender = "Female"
else:
deep_gender = "Male"
# 使用传入的置信度,如果没有则使用固定参考值 0.85
age_conf = initial_age_confidence if initial_age_confidence is not None else 0.85
return {
"age": str(int(age)),
"age_confidence": round(age_conf, 4),
"emotion": emotion,
"emotion_analysis": emotion_scores,
"gender": deep_gender,
"gender_confidence": deep_gender_conf,
}
except Exception as e:
logger.error(f"DeepFace age emotion prediction failed, falling back to HowCuteAmI: {e}")
return self._predict_age_with_howcuteami_fallback(face_image)
def _predict_age_with_howcuteami_fallback(
self, face_image: np.ndarray
) -> Dict[str, Any]:
"""HowCuteAmI年龄预测回退方案"""
try:
if face_image is None or face_image.size == 0:
raise ValueError("无法读取人脸图像")
face_resized = cv2.resize(face_image, (224, 224))
blob = cv2.dnn.blobFromImage(
face_resized, 1.0, (224, 224), self.MODEL_MEAN_VALUES, swapRB=False
)
# 年龄预测
self.age_net.setInput(blob)
age_preds = self.age_net.forward()
age = self.age_list[age_preds[0].argmax()]
age_confidence = float(np.max(age_preds[0]))
return {
"age": age[1:-1], # 去掉括号
"age_confidence": round(age_confidence, 4),
"emotion": "neutral", # 默认情绪
"emotion_analysis": {"neutral": 100.0}, # 默认情绪分析
}
except Exception as e:
logger.error(f"HowCuteAmI age prediction fallback failed: {e}")
return {
"age": "25-32",
"age_confidence": 0.5,
"emotion": "neutral",
"emotion_analysis": {"neutral": 100.0},
}
def _predict_with_hybrid_model(
self, face: np.ndarray, face_image: np.ndarray
) -> Dict[str, Any]:
"""混合模型预测:HowCuteAmI(颜值+性别)+ DeepFace(年龄+情绪,年龄置信度低时优先使用)"""
# 使用HowCuteAmI预测颜值和性别
beauty_gender_result = self._predict_beauty_gender_with_howcuteami(face)
# Hybrid 模式下可配置是否启用 DeepFace 情绪识别(默认启用)。
deepface_emotion_enabled = bool(getattr(config, "DEEPFACE_EMOTION_ENABLED", True))
age_emotion_result: Optional[Dict[str, Any]] = None
# 首先获取HowCuteAmI的年龄/性别预测置信度
howcuteami_age_confidence = beauty_gender_result.get("age_confidence", 0)
gender_confidence = beauty_gender_result.get("gender_confidence", 0)
if gender_confidence >= 1:
gender_confidence = 0.9999
age = beauty_gender_result["age"]
# 如果HowCuteAmI的年龄置信度低于阈值,则使用DeepFace的年龄
agec = config.AGE_CONFIDENCE
if howcuteami_age_confidence < agec:
# 需要 DeepFace 年龄时,仍调用 DeepFace(但可按开关选择是否同时跑 emotion)
age_emotion_result = self._predict_age_emotion_with_deepface(
face_image, include_emotion=deepface_emotion_enabled,
initial_age_confidence=howcuteami_age_confidence
)
deep_age = age_emotion_result["age"]
logger.info(
f"HowCuteAmI age confidence ({howcuteami_age_confidence}) below {agec}, value=({age}); using DeepFace for age prediction, value={deep_age}"
)
# 合并结果,使用DeepFace的年龄预测
result = {
"gender": beauty_gender_result["gender"], # 先用HowCuteAmI,后面可能回退
"gender_confidence": self._cap_conf(gender_confidence),
"beauty_score": beauty_gender_result["beauty_score"],
"beauty_score_100": beauty_gender_result["beauty_score_100"],
"beauty_raw_score": beauty_gender_result["beauty_raw_score"],
"age": deep_age,
"age_confidence": age_emotion_result["age_confidence"],
"emotion": age_emotion_result.get("emotion") or "neutral",
"emotion_analysis": age_emotion_result.get("emotion_analysis") or {"neutral": 100.0},
"model_used": "hybrid_deepface_age",
"age_model_used": "DeepFace",
"gender_model_used": "HowCuteAmI",
}
else:
# HowCuteAmI年龄置信度足够高,使用原有逻辑
logger.info(
f"HowCuteAmI age confidence ({howcuteami_age_confidence}) is high enough, value={age}; using HowCuteAmI for age prediction"
)
# 情绪识别完全可选:关闭时直接返回默认值,避免多一次 DeepFace 推理。
if deepface_emotion_enabled:
age_emotion_result = self._predict_age_emotion_with_deepface(
face_image, include_emotion=True,
initial_age_confidence=howcuteami_age_confidence
)
emotion = age_emotion_result.get("emotion") or "neutral"
emotion_analysis = age_emotion_result.get("emotion_analysis") or {"neutral": 100.0}
else:
emotion = "neutral"
emotion_analysis = {"neutral": 100.0}
# 合并结果,保留HowCuteAmI的年龄预测
result = {
"gender": beauty_gender_result["gender"], # 先用HowCuteAmI,后面可能回退
"gender_confidence": self._cap_conf(gender_confidence),
"beauty_score": beauty_gender_result["beauty_score"],
"beauty_score_100": beauty_gender_result["beauty_score_100"],
"beauty_raw_score": beauty_gender_result["beauty_raw_score"],
"age": beauty_gender_result["age"],
"age_confidence": beauty_gender_result["age_confidence"],
"emotion": emotion,
"emotion_analysis": emotion_analysis,
"model_used": "hybrid",
"age_model_used": "HowCuteAmI",
"gender_model_used": "HowCuteAmI",
}
# 统一性别判定规则:任一模型判为Female则Female;两者都为Male才Male
try:
how_gender = beauty_gender_result.get("gender")
how_conf = float(beauty_gender_result.get("gender_confidence", 0) or 0)
deep_gender = age_emotion_result.get("gender") if age_emotion_result else None
deep_conf = float(age_emotion_result.get("gender_confidence", 0) or 0) if age_emotion_result else 0.0
final_gender = result.get("gender")
final_conf = float(result.get("gender_confidence", 0) or 0)
# 规则判断
if (str(how_gender) == "Female") or (str(deep_gender) == "Female"):
final_gender = "Female"
final_conf = max(how_conf if how_gender == "Female" else 0,
deep_conf if deep_gender == "Female" else 0)
result["gender_model_used"] = "Combined(H+DF)"
elif (str(how_gender) == "Male") and (str(deep_gender) == "Male"):
final_gender = "Male"
final_conf = max(how_conf if how_gender == "Male" else 0,
deep_conf if deep_gender == "Male" else 0)
result["gender_model_used"] = "Combined(H+DF)"
# 否则保持原判定
result["gender"] = final_gender
result["gender_confidence"] = self._cap_conf(final_conf)
except Exception:
pass
return result
def _predict_with_howcuteami(self, face: np.ndarray) -> Dict[str, Any]:
"""使用HowCuteAmI模型进行完整预测"""
try:
# 性别预测
blob = cv2.dnn.blobFromImage(
face, 1.0, (224, 224), self.MODEL_MEAN_VALUES, swapRB=False
)
self.gender_net.setInput(blob)
gender_preds = self.gender_net.forward()
gender = self.gender_list[gender_preds[0].argmax()]
gender_confidence = float(np.max(gender_preds[0]))
gender_confidence = self._cap_conf(gender_confidence)
# 年龄预测
self.age_net.setInput(blob)
age_preds = self.age_net.forward()
age = self.age_list[age_preds[0].argmax()]
age_confidence = float(np.max(age_preds[0]))
# 颜值预测
blob_beauty = cv2.dnn.blobFromImage(
face, 1.0 / 255, (224, 224), self.MODEL_MEAN_VALUES, swapRB=False
)
self.beauty_net.setInput(blob_beauty)
beauty_preds = self.beauty_net.forward()
raw_beauty_sum = float(np.sum(beauty_preds[0]))
# 十分制计算
beauty_score = round(float(2.0 * raw_beauty_sum), 1)
beauty_score = min(10.0, max(0.0, beauty_score))
beauty_score = self._adjust_beauty_score(beauty_score)
# 百分制计算
beauty_score_100 = float(20.0 * raw_beauty_sum)
beauty_score_100 = min(100.0, max(0.0, beauty_score_100))
beauty_score_100 = self._adjust_beauty_score(beauty_score_100, scale=100.0)
raw_score = raw_beauty_sum
return {
"gender": gender,
"gender_confidence": gender_confidence,
"age": age[1:-1], # 去掉括号
"age_confidence": round(age_confidence, 4),
"beauty_score": beauty_score_100,
"beauty_score_100": beauty_score_100,
"beauty_raw_score": round(raw_score, 4),
"model_used": "HowCuteAmI",
"emotion": "neutral", # HowCuteAmI不支持情绪分析
"emotion_analysis": {"neutral": 100.0},
"age_model_used": "HowCuteAmI",
"gender_model_used": "HowCuteAmI",
"beauty_model_used": "HowCuteAmI",
}
except Exception as e:
logger.error(f"HowCuteAmI prediction failed: {e}")
raise e
def _predict_with_deepface(self, face_image: np.ndarray) -> Dict[str, Any]:
"""使用DeepFace进行预测"""
if not DEEPFACE_AVAILABLE:
raise ValueError("DeepFace未安装")
if face_image is None or face_image.size == 0:
raise ValueError("无效的人脸图像")
try:
deepface_emotion_enabled = bool(getattr(config, "DEEPFACE_EMOTION_ENABLED", True))
actions = ["age", "gender", "emotion"] if deepface_emotion_enabled else ["age", "gender"]
# DeepFace分析 - 禁用进度条和详细输出
result = _deepface_analyze_with_recovery(
img_path=face_image,
actions=actions,
enforce_detection=False,
detector_backend="skip",
silent=True # 禁用进度条输出
)
# 处理结果 (DeepFace返回的结果格式可能是list或dict)
if isinstance(result, list):
result = result[0]
# 提取信息
age = result.get("age", 25)
gender = result.get("dominant_gender", "Woman")
gender_confidence = result.get("gender", {}).get(gender, 0.5) / 100
gender_confidence = self._cap_conf(gender_confidence)
# 统一性别标签
if gender.lower() in ["woman", "female"]:
gender = "Female"
else:
gender = "Male"
# DeepFace没有内置颜值评分,这里使用简单的启发式方法
if deepface_emotion_enabled:
emotion = result.get("dominant_emotion", "neutral")
emotion_scores = result.get("emotion", {}) or {}
else:
emotion = "neutral"
emotion_scores = {"neutral": 100.0}
# 基于情绪和年龄的简单颜值估算
happiness_score = emotion_scores.get("happy", 0) / 100
neutral_score = emotion_scores.get("neutral", 0) / 100
# 简单的颜值算法 (可以改进)
base_beauty = 6.0 # 基础分
emotion_bonus = happiness_score * 2 + neutral_score * 1
age_factor = max(0.5, 1 - abs(age - 25) / 50) # 25岁为最佳年龄
original_score = base_beauty + emotion_bonus + age_factor
beauty_score = round(min(10.0, original_score), 2)
# 对 DeepFace 也应用提分逻辑
beauty_score = self._adjust_beauty_score(beauty_score)
# 百分制
beauty_score_100 = min(100.0, original_score * 10)
beauty_score_100 = self._adjust_beauty_score(beauty_score_100, scale=100.0)
# DeepFace不返回年龄置信度,使用固定参考值 0.85
age_conf = 0.85
return {
"gender": gender,
"gender_confidence": gender_confidence,
"age": str(int(age)),
"age_confidence": age_conf,
"beauty_score": beauty_score_100,
"beauty_score_100": beauty_score_100,
"beauty_raw_score": round(original_score / 10, 4),
"model_used": "DeepFace",
"emotion": emotion,
"emotion_analysis": emotion_scores,
"age_model_used": "DeepFace",
"gender_model_used": "DeepFace",
"beauty_model_used": "Heuristic",
}
except Exception as e:
logger.error(f"DeepFace prediction failed: {e}")
raise e
def analyze_faces(
self,
image: np.ndarray,
original_image_hash: str,
model_type: ModelType = ModelType.HYBRID,
language: str = "en",
) -> Dict[str, Any]:
"""
分析图片中的人脸
:param image: 输入图像
:param original_image_hash: 原始图片的MD5哈希值
:param model_type: 使用的模型类型
:return: 分析结果
"""
if image is None:
raise ValueError("invalid image input")
# 检测人脸
face_boxes = self._detect_faces(image)
if not face_boxes:
return {
"success": False,
"message": "Please upload a clear front-facing photo without any obstructions.",
"face_count": 0,
"faces": [],
"annotated_image": None,
"model_used": model_type.value,
}
results = {
"success": True,
"message": f"success detected {len(face_boxes)} faces",
"face_count": len(face_boxes),
"faces": [],
"model_used": model_type.value,
}
# 复制原图用于绘制
annotated_image = image.copy()
logger.info(
f"Input annotated_image shape: {annotated_image.shape}, dtype: {annotated_image.dtype}, ndim: {annotated_image.ndim}"
)
# 分析每张人脸
for i, face_box in enumerate(face_boxes):
# 裁剪人脸
face_cropped = self._crop_face(image, face_box)
if face_cropped.size == 0:
logger.warning(f"Cropped face {i + 1} is empty, skipping.")
continue
face_resized = cv2.resize(face_cropped, (224, 224))
face_for_deepface = face_cropped.copy()
# 根据模型类型进行预测
try:
if model_type == ModelType.HYBRID:
# 混合模式:颜值性别用HowCuteAmI,年龄情绪用DeepFace
prediction_result = self._predict_with_hybrid_model(
face_resized, face_for_deepface
)
elif model_type == ModelType.HOWCUTEAMI:
prediction_result = self._predict_with_howcuteami(face_resized)
# 非混合模式也进行性别合并:引入DeepFace性别(不需要 emotion,减少耗时)
try:
age_emotion_result = self._predict_age_emotion_with_deepface(
face_for_deepface, include_emotion=False
)
how_gender = prediction_result.get("gender")
how_conf = float(prediction_result.get("gender_confidence", 0) or 0)
deep_gender = age_emotion_result.get("gender")
deep_conf = float(age_emotion_result.get("gender_confidence", 0) or 0)
final_gender = prediction_result.get("gender")
final_conf = float(prediction_result.get("gender_confidence", 0) or 0)
if (str(how_gender) == "Female") or (str(deep_gender) == "Female"):
final_gender = "Female"
final_conf = max(
how_conf if how_gender == "Female" else 0,
deep_conf if deep_gender == "Female" else 0,
)
prediction_result["gender_model_used"] = "Combined(H+DF)"
elif (str(how_gender) == "Male") and (str(deep_gender) == "Male"):
final_gender = "Male"
final_conf = max(
how_conf if how_gender == "Male" else 0,
deep_conf if deep_gender == "Male" else 0,
)
prediction_result["gender_model_used"] = "Combined(H+DF)"
prediction_result["gender"] = final_gender
prediction_result["gender_confidence"] = round(float(final_conf), 4)
except Exception:
pass
elif model_type == ModelType.DEEPFACE and DEEPFACE_AVAILABLE:
prediction_result = self._predict_with_deepface(face_for_deepface)
# 非混合模式也进行性别合并:引入HowCuteAmI性别
try:
beauty_gender_result = self._predict_beauty_gender_with_howcuteami(
face_resized
)
deep_gender = prediction_result.get("gender")
deep_conf = float(prediction_result.get("gender_confidence", 0) or 0)
how_gender = beauty_gender_result.get("gender")
how_conf = float(beauty_gender_result.get("gender_confidence", 0) or 0)
final_gender = prediction_result.get("gender")
final_conf = float(prediction_result.get("gender_confidence", 0) or 0)
if (str(how_gender) == "Female") or (str(deep_gender) == "Female"):
final_gender = "Female"
final_conf = max(
how_conf if how_gender == "Female" else 0,
deep_conf if deep_gender == "Female" else 0,
)
prediction_result["gender_model_used"] = "Combined(H+DF)"
elif (str(how_gender) == "Male") and (str(deep_gender) == "Male"):
final_gender = "Male"
final_conf = max(
how_conf if how_gender == "Male" else 0,
deep_conf if deep_gender == "Male" else 0,
)
prediction_result["gender_model_used"] = "Combined(H+DF)"
prediction_result["gender"] = final_gender
prediction_result["gender_confidence"] = round(float(final_conf), 4)
except Exception:
pass
else:
# 回退到混合模式
prediction_result = self._predict_with_hybrid_model(
face_resized, face_for_deepface
)
logger.warning(
f"Model {model_type.value} is not available, using hybrid mode"
)
except Exception as e:
logger.error(f"Prediction failed, using default values: {e}")
prediction_result = {
"gender": "Unknown",
"gender_confidence": 0.5,
"age": "25-32",
"age_confidence": 0.5,
"beauty_score": 5.0,
"beauty_score_100": 50.0,
"beauty_raw_score": 0.5,
"emotion": "neutral",
"emotion_analysis": {"neutral": 100.0},
"model_used": "fallback",
}
# 五官评分仅为辅助数据,任何异常都不能中断总评分结果。
try:
facial_features = self.facial_analyzer.analyze_facial_features(
face_cropped,
face_box,
language=language,
age_str=str(prediction_result.get("age", "25")),
gender=str(prediction_result.get("gender", "Female")),
beauty_score=float(prediction_result.get("beauty_score", 7.0)),
emotion=str(prediction_result.get("emotion", "neutral")),
)
except Exception as exc:
logger.warning("Optional facial analysis failed; keeping main score: %s", exc)
facial_features = FacialFeatureAnalyzer._unavailable_facial_analysis(
"auxiliary_analysis_failed"
)
# 颜色设置与年龄显示统一(应用女性年龄调整)
gender = prediction_result.get("gender", "Unknown")
color_bgr = self.gender_colors.get(gender, (128, 128, 128))
color_hex = f"#{color_bgr[2]:02x}{color_bgr[1]:02x}{color_bgr[0]:02x}"
# 年龄文本与调整
raw_age_str = prediction_result.get("age", "Unknown")
display_age_str = str(raw_age_str)
age_adjusted_flag = False
age_adjustment_value = int(getattr(config, "FEMALE_AGE_ADJUSTMENT", 0) or 0)
age_adjustment_threshold = int(getattr(config, "FEMALE_AGE_ADJUSTMENT_THRESHOLD", 999) or 999)
# 仅对女性且年龄达到阈值时进行调整
try:
# 支持 "25-32" 或 "25" 格式
if "-" in str(raw_age_str):
age_num = int(str(raw_age_str).split("-")[0].strip("() "))
else:
age_num = int(str(raw_age_str).strip())
if str(gender) == "Female" and age_num >= age_adjustment_threshold and age_adjustment_value > 0:
adjusted_age = max(0, age_num - age_adjustment_value)
display_age_str = str(adjusted_age)
age_adjusted_flag = True
try:
logger.info(f"Adjusted age for female (draw+data): {age_num} -> {adjusted_age}")
except Exception:
pass
except Exception:
# 无法解析年龄时,保持原样
pass
# 保存裁剪的人脸
date_sub = get_date_subfolder(OUTPUT_DIR)
cropped_face_filename = f"{date_sub}/{original_image_hash}_face_{i + 1}.webp"
cropped_face_path = os.path.join(OUTPUT_DIR, cropped_face_filename)
try:
save_image_high_quality(face_cropped, cropped_face_path)
logger.info(f"cropped face: {cropped_face_path}")
except Exception as e:
logger.error(f"Failed to save cropped face {cropped_face_path}: {e}")
cropped_face_filename = None
# 在图片上绘制标注
if config.DRAW_SCORE:
cv2.rectangle(
annotated_image,
(face_box[0], face_box[1]),
(face_box[2], face_box[3]),
color_bgr,
int(round(image.shape[0] / 400)),
8,
)
# 标签文本
beauty_score = prediction_result.get("beauty_score_100", 0)
label = f"{gender}, {display_age_str}, {beauty_score:.2f}"
font_scale = max(
0.3, min(0.7, image.shape[0] / 800)
) # 从500改为800,范围从0.5-1.0改为0.3-0.7
font_thickness = 2
font = cv2.FONT_HERSHEY_SIMPLEX
# 绘制文本
text_x = face_box[0]
text_y = face_box[1] - 10 if face_box[1] - 10 > 20 else face_box[1] + 30
# 计算文字大小(宽高)
(text_width, text_height), baseline = cv2.getTextSize(label, font, font_scale, font_thickness)
# 画黑色矩形背景,稍微比文字框大一点,增加边距
background_tl = (text_x, text_y - text_height - baseline) # 矩形左上角
background_br = (text_x + text_width, text_y + baseline) # 矩形右下角
if config.DRAW_SCORE:
cv2.rectangle(
annotated_image,
background_tl,
background_br,
color_bgr, # 黑色背景
thickness=-1 # 填充
)
cv2.putText(
annotated_image,
label,
(text_x, text_y),
font,
font_scale,
(255, 255, 255),
font_thickness,
cv2.LINE_AA,
)
# 构建人脸结果
face_result = {
"face_id": i + 1,
"gender": gender,
"gender_confidence": prediction_result.get("gender_confidence", 0),
"gender_model_used": prediction_result.get("gender_model_used", prediction_result.get("model_used", model_type.value)),
"age": display_age_str,
"age_confidence": prediction_result.get("age_confidence", 0),
"age_model_used": prediction_result.get("age_model_used", prediction_result.get("model_used", model_type.value)),
"beauty_score": prediction_result.get("beauty_score", 0),
"beauty_score_100": prediction_result.get("beauty_score_100", 0),
"beauty_raw_score": prediction_result.get("beauty_raw_score", 0),
"emotion": prediction_result.get("emotion") or "neutral",
"emotion_analysis": prediction_result.get("emotion_analysis") or {"neutral": 100.0},
"facial_features": facial_features, # 五官分析
"bounding_box": {
"x1": int(face_box[0]),
"y1": int(face_box[1]),
"x2": int(face_box[2]),
"y2": int(face_box[3]),
},
"color": {
"bgr": [int(color_bgr[0]), int(color_bgr[1]), int(color_bgr[2])],
"hex": color_hex,
},
"cropped_face_filename": cropped_face_filename,
"model_used": prediction_result.get("model_used", model_type.value),
}
if age_adjusted_flag:
face_result["age_adjusted"] = True
face_result["age_adjustment_value"] = int(age_adjustment_value)
results["faces"].append(face_result)
results["annotated_image"] = annotated_image
return results
def _warmup_models(self):
"""预热模型,减少首次调用延迟"""
try:
logger.info("Starting to warm up models...")
# 创建一个小的测试图像 (64x64)
test_image = np.ones((64, 64, 3), dtype=np.uint8) * 128
# 预热DeepFace模型(如果可用)
if DEEPFACE_AVAILABLE:
try:
import tempfile
deepface_emotion_enabled = bool(getattr(config, "DEEPFACE_EMOTION_ENABLED", True))
with tempfile.NamedTemporaryFile(suffix='.webp', delete=False) as tmp_file:
cv2.imwrite(tmp_file.name, test_image, [cv2.IMWRITE_WEBP_QUALITY, 95])
# 预热DeepFace - 使用最小的actions集合
_deepface_analyze_with_recovery(
img_path=tmp_file.name,
actions=["age", "gender", "emotion"] if deepface_emotion_enabled else ["age", "gender"],
detector_backend="yolov8",
enforce_detection=False,
silent=True
)
os.unlink(tmp_file.name)
logger.info("DeepFace model warm-up completed")
except Exception as e:
logger.warning(f"DeepFace model warm-up failed: {e}")
# 预热OpenCV DNN模型
try:
# 预热人脸检测模型
blob = cv2.dnn.blobFromImage(test_image, 1.0, (300, 300), (104, 117, 123))
self.face_net.setInput(blob)
self.face_net.forward()
# 预热年龄预测模型
test_face = cv2.resize(test_image, (224, 224))
blob = cv2.dnn.blobFromImage(test_face, 1.0, (224, 224), self.MODEL_MEAN_VALUES, swapRB=False)
self.age_net.setInput(blob)
self.age_net.forward()
# 预热性别预测模型
self.gender_net.setInput(blob)
self.gender_net.forward()
# 预热颜值评分模型
self.beauty_net.setInput(blob)
self.beauty_net.forward()
logger.info("OpenCV DNN model warm-up completed")
except Exception as e:
logger.warning(f"OpenCV DNN model warm-up failed: {e}")
logger.info("Model warm-up completed")
except Exception as e:
logger.warning(f"Error occurred during model warm-up: {e}")