fix

api_routes.py

@@ -12,7 +12,7 @@ import uuid
 import subprocess
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import cv2
 import numpy as np
@@ -4186,32 +4186,24 @@ async def face_similarity_verification(
             if not face_boxes2:
                 raise HTTPException(status_code=400, detail="第二张图片中未检测到人脸，请上传包含清晰人脸的图片")
 
-        # 保存原始图片到IMAGES_DIR
+        # 保存原始图片到IMAGES_DIR（先不上传 BOS，供 DeepFace 使用）
         original_path1 = os.path.join(IMAGES_DIR, original_filename1)
-        if save_image_high_quality(image1, original_path1,
-                                   quality=SAVE_QUALITY):
-            await _record_output_file(
-                file_path=original_path1,
-                nickname=nickname,
-                category="original",
-                extra={
-                    "source": "face_verify",
-                    "role": "original1",
-                },
-            )
+        if not save_image_high_quality(
+            image1,
+            original_path1,
+            quality=SAVE_QUALITY,
+            upload_to_bos=False,
+        ):
+            raise HTTPException(status_code=500, detail="保存第一张原始图片失败")
 
         original_path2 = os.path.join(IMAGES_DIR, original_filename2)
-        if save_image_high_quality(image2, original_path2,
-                                   quality=SAVE_QUALITY):
-            await _record_output_file(
-                file_path=original_path2,
-                nickname=nickname,
-                category="original",
-                extra={
-                    "source": "face_verify",
-                    "role": "original2",
-                },
-            )
+        if not save_image_high_quality(
+            image2,
+            original_path2,
+            quality=SAVE_QUALITY,
+            upload_to_bos=False,
+        ):
+            raise HTTPException(status_code=500, detail="保存第二张原始图片失败")
 
         # 调用DeepFace.verify进行人脸比对
         logger.info("Starting DeepFace verification...")
@@ -4294,66 +4286,57 @@ async def face_similarity_verification(
         if analyzer is None:
             _ensure_analyzer()
 
-        # 绘制带特征点的原始图像
-        if analyzer is not None:
+        def _apply_landmarks_on_original(
+            source_image: np.ndarray,
+            region: dict,
+            label: str,
+        ) -> Tuple[np.ndarray, bool]:
+            if analyzer is None or not region:
+                return source_image, False
             try:
-                # 在第一张原始图像上绘制特征点
-                original_with_landmarks_img1 = image1.copy()
-                if img1_region:
-                    # 提取人脸区域坐标
-                    x1, y1, w1, h1 = img1_region.get("x", 0), img1_region.get("y", 0), img1_region.get("w", 0), img1_region.get("h", 0)
-                    # 确保坐标在图像范围内
-                    x1, y1 = max(0, x1), max(0, y1)
-                    x1_end, y1_end = min(image1.shape[1], x1 + w1), min(image1.shape[0], y1 + h1)
-                    # 裁剪人脸区域
-                    face_region1 = image1[y1:y1_end, x1:x1_end]
-                    # 在人脸区域上绘制特征点
-                    face_with_landmarks1 = analyzer.facial_analyzer.draw_facial_landmarks(face_region1)
-                    # 将绘制了特征点的人脸区域放回原图
-                    original_with_landmarks_img1[y1:y1_end, x1:x1_end] = face_with_landmarks1
-
-                # 在第二张原始图像上绘制特征点
-                original_with_landmarks_img2 = image2.copy()
-                if img2_region:
-                    # 提取人脸区域坐标
-                    x2, y2, w2, h2 = img2_region.get("x", 0), img2_region.get("y", 0), img2_region.get("w", 0), img2_region.get("h", 0)
-                    # 确保坐标在图像范围内
-                    x2, y2 = max(0, x2), max(0, y2)
-                    x2_end, y2_end = min(image2.shape[1], x2 + w2), min(image2.shape[0], y2 + h2)
-                    # 裁剪人脸区域
-                    face_region2 = image2[y2:y2_end, x2:x2_end]
-                    # 在人脸区域上绘制特征点
-                    face_with_landmarks2 = analyzer.facial_analyzer.draw_facial_landmarks(face_region2)
-                    # 将绘制了特征点的人脸区域放回原图
-                    original_with_landmarks_img2[y2:y2_end, x2:x2_end] = face_with_landmarks2
-
-                # 保存带特征点的原始图片
-                if save_image_high_quality(original_with_landmarks_img1,
-                                           original_path1,
-                                           quality=SAVE_QUALITY):
-                    await _record_output_file(
-                        file_path=original_path1,
-                        nickname=nickname,
-                        category="original",
-                        extra={
-                            "source": "face_verify",
-                            "role": "original1_landmarks",
-                        },
-                    )
-                if save_image_high_quality(original_with_landmarks_img2,
-                                           original_path2,
-                                           quality=SAVE_QUALITY):
-                    await _record_output_file(
-                        file_path=original_path2,
-                        nickname=nickname,
-                        category="original",
-                        extra={
-                            "source": "face_verify",
-                            "role": "original2_landmarks",
-                        },
-                    )
-            except Exception as e:
-                logger.warning(f"Failed to draw facial landmarks on original images: {e}")
+                x = max(0, region.get("x", 0))
+                y = max(0, region.get("y", 0))
+                w = region.get("w", 0)
+                h = region.get("h", 0)
+                x_end = min(source_image.shape[1], x + w)
+                y_end = min(source_image.shape[0], y + h)
+                if x_end <= x or y_end <= y:
+                    return source_image, False
+                result_img = source_image.copy()
+                face_region = result_img[y:y_end, x:x_end]
+                face_with_landmarks = analyzer.facial_analyzer.draw_facial_landmarks(face_region)
+                result_img[y:y_end, x:x_end] = face_with_landmarks
+                return result_img, True
+            except Exception as exc:
+                logger.warning(f"Failed to draw facial landmarks on original image {label}: {exc}")
+                return source_image, False
+
+        original_output_img1, original1_has_landmarks = _apply_landmarks_on_original(image1, img1_region, "1")
+        original_output_img2, original2_has_landmarks = _apply_landmarks_on_original(image2, img2_region, "2")
+
+        if save_image_high_quality(original_output_img1, original_path1, quality=SAVE_QUALITY):
+            await _record_output_file(
+                file_path=original_path1,
+                nickname=nickname,
+                category="original",
+                extra={
+                    "source": "face_verify",
+                    "role": "original1_landmarks" if original1_has_landmarks else "original1",
+                    "with_landmarks": original1_has_landmarks,
+                },
+            )
+        if save_image_high_quality(original_output_img2, original_path2, quality=SAVE_QUALITY):
+            await _record_output_file(
+                file_path=original_path2,
+                nickname=nickname,
+                category="original",
+                extra={
+                    "source": "face_verify",
+                    "role": "original2_landmarks" if original2_has_landmarks else "original2",
+                    "with_landmarks": original2_has_landmarks,
+                },
+            )
+
         # 如果有区域信息，则裁剪人脸
         if img1_region and img2_region:
             try:
@@ -4371,126 +4354,43 @@ async def face_similarity_verification(
                 face_img1 = image1[y1:y1_end, x1:x1_end]
                 face_img2 = image2[y2:y2_end, x2:x2_end]
 
-                # 保存裁剪后的人脸图片
                 face_path1 = os.path.join(IMAGES_DIR, face_filename1)
                 face_path2 = os.path.join(IMAGES_DIR, face_filename2)
-                if save_image_high_quality(face_img1, face_path1,
-                                           quality=SAVE_QUALITY):
+                # 根据分析器可用性决定是否绘制特征点，仅保存最终版本一次
+                def _prepare_face_image(face_img, face_index):
+                    if analyzer is None:
+                        return face_img, False
+                    try:
+                        return analyzer.facial_analyzer.draw_facial_landmarks(face_img.copy()), True
+                    except Exception as exc:
+                        logger.warning(f"Failed to draw facial landmarks on face{face_index}: {exc}")
+                        return face_img, False
+
+                face_output_img1, face1_has_landmarks = _prepare_face_image(face_img1, 1)
+                face_output_img2, face2_has_landmarks = _prepare_face_image(face_img2, 2)
+
+                if save_image_high_quality(face_output_img1, face_path1, quality=SAVE_QUALITY):
                     await _record_output_file(
                         file_path=face_path1,
                         nickname=nickname,
                         category="face",
                         extra={
                             "source": "face_verify",
-                            "role": "face1",
+                            "role": "face1_landmarks" if face1_has_landmarks else "face1",
+                            "with_landmarks": face1_has_landmarks,
                         },
                     )
-                if save_image_high_quality(face_img2, face_path2,
-                                           quality=SAVE_QUALITY):
+                if save_image_high_quality(face_output_img2, face_path2, quality=SAVE_QUALITY):
                     await _record_output_file(
                         file_path=face_path2,
                         nickname=nickname,
                         category="face",
                         extra={
                             "source": "face_verify",
-                            "role": "face2",
+                            "role": "face2_landmarks" if face2_has_landmarks else "face2",
+                            "with_landmarks": face2_has_landmarks,
                         },
                     )
-
-                # 确保分析器已初始化，用于绘制特征点
-                if analyzer is None:
-                    _ensure_analyzer()
-
-                # 在人脸图像上绘制特征点（如果分析器可用）
-                if analyzer is not None:
-                    try:
-                        face_with_landmarks_img1 = analyzer.facial_analyzer.draw_facial_landmarks(face_img1)
-                        face_with_landmarks_img2 = analyzer.facial_analyzer.draw_facial_landmarks(face_img2)
-
-                        # 保存带特征点的人脸图片（使用原来的文件名）
-                        face_with_landmarks_path1 = os.path.join(IMAGES_DIR, face_filename1)
-                        face_with_landmarks_path2 = os.path.join(IMAGES_DIR, face_filename2)
-                        if save_image_high_quality(face_with_landmarks_img1,
-                                                   face_with_landmarks_path1,
-                                                   quality=SAVE_QUALITY):
-                            await _record_output_file(
-                                file_path=face_with_landmarks_path1,
-                                nickname=nickname,
-                                category="face",
-                                extra={
-                                    "source": "face_verify",
-                                    "role": "face1_landmarks",
-                                },
-                            )
-                        if save_image_high_quality(face_with_landmarks_img2,
-                                                   face_with_landmarks_path2,
-                                                   quality=SAVE_QUALITY):
-                            await _record_output_file(
-                                file_path=face_with_landmarks_path2,
-                                nickname=nickname,
-                                category="face",
-                                extra={
-                                    "source": "face_verify",
-                                    "role": "face2_landmarks",
-                                },
-                            )
-                    except Exception as e:
-                        logger.warning(f"Failed to draw facial landmarks: {e}")
-                        # 如果绘制特征点失败，使用裁剪的人脸图像
-                        face_with_landmarks_path1 = os.path.join(IMAGES_DIR, face_filename1)
-                        face_with_landmarks_path2 = os.path.join(IMAGES_DIR, face_filename2)
-                        if save_image_high_quality(face_img1,
-                                                   face_with_landmarks_path1,
-                                                   quality=SAVE_QUALITY):
-                            await _record_output_file(
-                                file_path=face_with_landmarks_path1,
-                                nickname=nickname,
-                                category="face",
-                                extra={
-                                    "source": "face_verify",
-                                    "role": "face1_landmarks_fallback",
-                                },
-                            )
-                        if save_image_high_quality(face_img2,
-                                                   face_with_landmarks_path2,
-                                                   quality=SAVE_QUALITY):
-                            await _record_output_file(
-                                file_path=face_with_landmarks_path2,
-                                nickname=nickname,
-                                category="face",
-                                extra={
-                                    "source": "face_verify",
-                                    "role": "face2_landmarks_fallback",
-                                },
-                            )
-                else:
-                    # 如果分析器不可用，使用裁剪的人脸图像
-                    face_with_landmarks_path1 = os.path.join(IMAGES_DIR, face_filename1)
-                    face_with_landmarks_path2 = os.path.join(IMAGES_DIR, face_filename2)
-                    if save_image_high_quality(face_img1,
-                                               face_with_landmarks_path1,
-                                               quality=SAVE_QUALITY):
-                        await _record_output_file(
-                            file_path=face_with_landmarks_path1,
-                            nickname=nickname,
-                            category="face",
-                            extra={
-                                "source": "face_verify",
-                                "role": "face1_no_landmarks",
-                            },
-                        )
-                    if save_image_high_quality(face_img2,
-                                               face_with_landmarks_path2,
-                                               quality=SAVE_QUALITY):
-                        await _record_output_file(
-                            file_path=face_with_landmarks_path2,
-                            nickname=nickname,
-                            category="face",
-                            extra={
-                                "source": "face_verify",
-                                "role": "face2_no_landmarks",
-                            },
-                        )
             except Exception as e:
                 logger.warning(f"Failed to crop faces: {e}")
         else:

utils.py

@@ -706,12 +706,19 @@ def move_file_to_archive(file_path: str):
         logger.error(f"Failed to move file {file_path} to archive: {error}")
 
 
-def save_image_high_quality(image: np.ndarray, output_path: str, quality: int = SAVE_QUALITY) -> bool:
+def save_image_high_quality(
+    image: np.ndarray,
+    output_path: str,
+    quality: int = SAVE_QUALITY,
+    *,
+    upload_to_bos: bool = True,
+) -> bool:
     """
     保存图像，保持高质量，不进行压缩
     :param image: 图像数组
     :param output_path: 输出路径
     :param quality: WebP质量 (0-100)，默认95
+    :param upload_to_bos: 是否在写入后同步至 BOS
     :return: 保存是否成功
     """
     try:
@@ -726,7 +733,8 @@ def save_image_high_quality(image: np.ndarray, output_path: str, quality: int =
             f.write(encoded_img)
 
         logger.info(f"High quality image saved successfully: {output_path}, quality: {quality}, size: {len(encoded_img) / 1024:.2f} KB")
-        upload_file_to_bos(output_path)
+        if upload_to_bos:
+            upload_file_to_bos(output_path)
         return True
     except Exception as e:
         logger.error(f"Failed to save image: {output_path}, error: {e}")