CardScoreDataServer/app/utils/card_score_calculate.py

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