from app.utils.scheme import ImageType, CardImageResponse from app.core.logger import get_logger from typing import List, Dict, Any, Optional logger = get_logger(__name__) def calculate_scores_from_images(images: List[CardImageResponse]) -> Dict[str, Any]: """ 根据图片计算分数（兼容 14 图与历史 8 图）。 - 必要条件：front_ring + back_ring 存在 - face 分数来源：同面 stripe(1~4) / coaxial / fusion，按可用图累加 - 不再依赖 coaxial 必须存在 """ scores = { "detection_score": None, "modified_score": None, "detection_score_detail": { "detection_center_score": None, "detection_corner_score": None, "detection_edge_score": None, "detection_face_score": None }, "modified_score_detail": { "modified_center_score": None, "modified_corner_score": None, "modified_edge_score": None, "modified_face_score": None }, "is_edited": False, } def _as_float(value: Any) -> float: try: return float(value or 0) except (TypeError, ValueError): return 0.0 def _get_reducts(src: Dict[str, Any], side: str) -> Dict[str, float]: defect_result = src.get("result", {}).get("defect_result", {}) center_result = src.get("result", {}).get("center_result", {}) return { "total": _as_float(src.get("result", {}).get("_used_compute_deduct_score", 0)), "center": _as_float(center_result.get("deduct_score", 0)), "corner": _as_float(defect_result.get(f"{side}_corner_deduct_score", 0)), "edge": _as_float(defect_result.get(f"{side}_edge_deduct_score", 0)), "face": _as_float(defect_result.get(f"{side}_face_deduct_score", 0)), } front_ring: Optional[CardImageResponse] = None back_ring: Optional[CardImageResponse] = None front_face_imgs: List[CardImageResponse] = [] back_face_imgs: List[CardImageResponse] = [] for img in images: image_type = img.image_type if image_type == ImageType.front_ring: front_ring = img elif image_type == ImageType.back_ring: back_ring = img if image_type in [ ImageType.front_stripe1, ImageType.front_stripe2, ImageType.front_stripe3, ImageType.front_stripe4, ImageType.front_coaxial, ImageType.front_fusion, ]: front_face_imgs.append(img) elif image_type in [ ImageType.back_stripe1, ImageType.back_stripe2, ImageType.back_stripe3, ImageType.back_stripe4, ImageType.back_coaxial, ImageType.back_fusion, ]: back_face_imgs.append(img) # ring 是当前模型计算中心/角/边的必要数据 if not front_ring or not back_ring: return scores try: # ---------- detection_score ---------- detection_score = 10.0 detection_center_score = 10.0 detection_corner_score = 10.0 detection_edge_score = 10.0 detection_face_score = 10.0 for ring_img, side in ((front_ring, "front"), (back_ring, "back")): try: reducts = _get_reducts(ring_img.detection_json or {}, side) detection_score += reducts["total"] detection_center_score += reducts["center"] detection_corner_score += reducts["corner"] detection_edge_score += reducts["edge"] detection_face_score += reducts["face"] except Exception as e: logger.warning(f"解析 detection_json 分数失败 (image_id={ring_img.id}): {e}") # 同面 face 辅助图（stripe/coaxial/fusion）叠加 face 与 total for side, face_imgs in (("front", front_face_imgs), ("back", back_face_imgs)): for img in face_imgs: try: reducts = _get_reducts(img.detection_json or {}, side) detection_score += reducts["total"] detection_face_score += reducts["face"] except Exception as e: logger.warning(f"解析 detection_json 分数失败 (image_id={img.id}): {e}") scores["detection_score"] = detection_score scores["detection_score_detail"] = { "detection_center_score": detection_center_score, "detection_corner_score": detection_corner_score, "detection_edge_score": detection_edge_score, "detection_face_score": detection_face_score, } # ---------- modified_score ---------- modified_score = 10.0 modified_center_score = 10.0 modified_corner_score = 10.0 modified_edge_score = 10.0 modified_face_score = 10.0 all_score_images = [front_ring, back_ring] + front_face_imgs + back_face_imgs is_edited = any(img.modified_json is not None for img in all_score_images if img is not None) scores["is_edited"] = is_edited if is_edited: for ring_img, side in ((front_ring, "front"), (back_ring, "back")): src = ring_img.modified_json if ring_img.modified_json is not None else ring_img.detection_json try: reducts = _get_reducts(src or {}, side) modified_score += reducts["total"] modified_center_score += reducts["center"] modified_corner_score += reducts["corner"] modified_edge_score += reducts["edge"] modified_face_score += reducts["face"] except Exception as e: logger.warning(f"解析 modified_json 分数失败 (image_id={ring_img.id}): {e}") for side, face_imgs in (("front", front_face_imgs), ("back", back_face_imgs)): for img in face_imgs: src = img.modified_json if img.modified_json is not None else img.detection_json try: reducts = _get_reducts(src or {}, side) modified_score += reducts["total"] modified_face_score += reducts["face"] except Exception as e: logger.warning(f"解析 modified_json 分数失败 (image_id={img.id}): {e}") scores["modified_score"] = modified_score scores["modified_score_detail"] = { "modified_center_score": modified_center_score, "modified_corner_score": modified_corner_score, "modified_edge_score": modified_edge_score, "modified_face_score": modified_face_score, } except Exception as e: logger.error(f"计算分数过程异常: {e}") return scores