CardVideoSummary/app/services/video_service.py

import math
import os
import re
import uuid
from dataclasses import dataclass
from typing import Any, Optional

import cv2
import numpy as np
import difflib

from app.core.config import settings
from app.core.logger import get_logger
from app.schemas.models import CardInfoInput, CardInfoOutput
import torch
from PIL import Image
from transformers import AutoImageProcessor, AutoModelForSemanticSegmentation

logger = get_logger("VideoService")


@dataclass
class FrameCandidate:
    """
    候选帧数据类：记录了从视频中抽取的某一帧的所有评分维度。
    使用 dataclass 让数据结构非常清晰。
    """
    frame: Any  # 原始图像矩阵 (OpenCV BGR)
    time_ms: int  # 该帧在视频中的时间戳 (毫秒)
    sharpness: float  # 拉普拉斯清晰度绝对值 (越大越清晰)
    time_weight: float  # 时间权重 (距离目标时间戳越近，权重越高)
    segmentation_used: bool = False  # 是否成功启用了分割模型
    has_card: bool = False  # 画面中是否出现了卡
    has_hand: bool = False  # 画面中是否出现了手
    card_area_ratio: float = 0.0  # 卡片占画面比例
    hand_area_ratio: float = 0.0  # 手占画面比例
    card_bbox: Optional[tuple[int, int, int, int]] = None  # (x, y, w, h) 卡片/手的聚焦边界框
    presence_score: float = 0.0  # 实体存在感得分 (卡和手面积越大，得分越高)
    sharpness_score: float = 0.0  # 归一化后的清晰度得分 (0~1)
    dwell_score: float = 0.0  # 停留得分 (连续出现的帧数越多得分越高，用于抗闪烁)
    base_score: float = 0.0  # 基础分 (不包含OCR)
    ocr_text: str = ""  # OCR识别出的文本
    ocr_score: float = 0.0  # OCR文本与预期卡片信息的匹配度得分 (0~1)
    final_score: float = 0.0  # 最终总分

    @property
    def is_present(self) -> bool:
        """只要有卡或者有手，就算该实体在画面中存在"""
        return self.has_card or self.has_hand


class VideoService:
    def __init__(self):
        # 高斯分布的 sigma，用于计算时间权重。更大表示对时间差容忍度更高
        self.weight_sigma = 6.0
        self.search_before_ms = settings.VIDEO_SEARCH_BEFORE_MS
        self.search_after_ms = settings.VIDEO_SEARCH_AFTER_MS
        self.analysis_fps = max(settings.VIDEO_ANALYSIS_FPS, 0.5)

        # 延迟加载组件，节省初始化时的内存占用
        self._ocr_engine = None
        self._ocr_disabled = False
        self._ocr_runtime_warning_sent = False

        self._seg_processor = None
        self._seg_model = None
        self._seg_torch = None
        self._seg_pil_image = None
        self._seg_disabled = False
        self._seg_runtime_warning_sent = False

    def time_str_to_ms(self, time_str: str) -> int:
        """将格式为 'HH:MM:SS' 或 'MM:SS' 的字符串转换为毫秒"""
        try:
            parts = list(map(int, time_str.split(":")))
            if len(parts) == 3:
                h, m, s = parts
                return (h * 3600 + m * 60 + s) * 1000
            if len(parts) == 2:
                m, s = parts
                return (m * 60 + s) * 1000
            return 0
        except ValueError:
            return 0

    def get_laplacian_sharpness(self, frame) -> float:
        """
        计算图像的拉普拉斯方差，这是业界最常用的无参考图像清晰度评估方法。
        方差越大，说明边缘信息越丰富（越不模糊）。
        """
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        # 增加高斯模糊，过滤掉反光产生的噪点和高频毛刺
        blurred = cv2.GaussianBlur(gray, (3, 3), 0)
        return float(cv2.Laplacian(blurred, cv2.CV_64F).var())

    def calculate_weight(self, current_time_ms: int, target_time_ms: int) -> float:
        """利用高斯函数计算时间权重。距离 target_time_ms 越近，返回值越接近 1.0"""
        diff_seconds = abs(current_time_ms - target_time_ms) / 1000.0
        return math.exp(-((diff_seconds ** 2) / (2 * self.weight_sigma ** 2)))

    def _analysis_stride(self, fps: float) -> int:
        """计算视频读取时的跳帧步长，确保处理速度匹配 VIDEO_ANALYSIS_FPS"""
        fps = fps if fps > 0 else 30.0
        return max(1, int(round(fps / self.analysis_fps)))

    def _ensure_ocr_engine(self):
        """单例模式懒加载 OCR 引擎 (RapidOCR)"""
        if self._ocr_disabled:
            return None
        if self._ocr_engine is not None:
            return self._ocr_engine

        try:
            from rapidocr import RapidOCR
            self._ocr_engine = RapidOCR()
        except Exception as exc:
            self._ocr_disabled = True
            logger.warning(f"OCR disabled: init failed: {exc}")
            return None

        return self._ocr_engine

    def _ensure_segmentation_model(self):
        """单例模式懒加载 HuggingFace Segformer 语义分割模型"""
        if self._seg_disabled:
            return None
        if self._seg_processor is not None and self._seg_model is not None:
            return self._seg_processor, self._seg_model

        model_dir = settings.VIDEO_SEG_MODEL_DIR
        if not model_dir or not os.path.exists(model_dir):
            self._seg_disabled = True
            logger.warning(f"Segmentation disabled: model dir not found: {model_dir}")
            return None

        try:
            self._seg_processor = AutoImageProcessor.from_pretrained(model_dir)
            self._seg_model = AutoModelForSemanticSegmentation.from_pretrained(model_dir)
            self._seg_model.eval()  # 开启评估模式

            # 自动分配到 GPU (如果可用) 以加速推理
            if torch.cuda.is_available():
                self._seg_model = self._seg_model.to("cuda")

            self._seg_torch = torch
            self._seg_pil_image = Image
        except Exception as exc:
            self._seg_disabled = True
            logger.warning(f"Segmentation disabled: model loading failed: {exc}")
            return None

        return self._seg_processor, self._seg_model

    def _largest_bbox(self, mask) -> Optional[tuple[int, int, int, int]]:
        """从二进制掩码 (Mask) 中提取面积最大的连通区域的外接矩形"""
        if mask is None or not mask.any():
            return None

        mask_uint8 = (mask.astype("uint8")) * 255
        contours, _ = cv2.findContours(mask_uint8, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        if not contours:
            return None

        largest = max(contours, key=cv2.contourArea)
        x, y, w, h = cv2.boundingRect(largest)

        # 过滤掉噪点 (宽或高小于20像素的通常是识别错误)
        if w < 20 or h < 20:
            return None
        return x, y, w, h

    def _expand_bbox(
            self,
            bbox: Optional[tuple[int, int, int, int]],
            width: int,
            height: int,
            margin_ratio: float = 0.08,
    ) -> Optional[tuple[int, int, int, int]]:
        """适度扩大 Bounding Box (增加 margin_ratio)，防止目标边缘被裁掉，有利于后续 OCR"""
        if bbox is None:
            return None

        x, y, w, h = bbox
        margin_x = int(w * margin_ratio)
        margin_y = int(h * margin_ratio)

        x1 = max(0, x - margin_x)
        y1 = max(0, y - margin_y)
        x2 = min(width, x + w + margin_x)
        y2 = min(height, y + h + margin_y)
        return x1, y1, x2 - x1, y2 - y1

    def _focus_region(self, frame, bbox: Optional[tuple[int, int, int, int]]):
        """裁剪出关注区域。如果没有有效 BBox，则返回原图，作为容错机制。"""
        height, width = frame.shape[:2]
        expanded = self._expand_bbox(bbox, width, height)
        if expanded is None:
            return frame

        x, y, w, h = expanded
        if w < 24 or h < 24:
            return frame
        return frame[y: y + h, x: x + w]

    def _compute_presence_score(
            self,
            segmentation_used: bool,
            has_card: bool,
            has_hand: bool,
            card_area_ratio: float,
            hand_area_ratio: float,
    ) -> float:
        """根据卡片和手的面积占比计算"存在感得分" (0.0 ~ 1.0)"""
        if not segmentation_used:
            return 0.0

        # 对占比进行归一化，最大不超过 1.0
        card_ratio = min(card_area_ratio / max(settings.VIDEO_MIN_CARD_AREA_RATIO, 1e-6), 1.0)
        hand_ratio = min(hand_area_ratio / max(settings.VIDEO_MIN_HAND_AREA_RATIO, 1e-6), 1.0)

        score = 0.0
        if has_card:
            score += 0.70 * max(card_ratio, 0.35)  # 卡片权重占 70%
        if has_hand:
            score += 0.30 * max(hand_ratio, 0.25)  # 手的权重占 30%
        if has_card and has_hand:
            score += 0.10  # 卡和手同框，给予额外 10% 奖励分

        return min(score, 1.0)

    def _batch_analyze_segmentation(self, frames: list[Any]) -> list[dict[str, Any]]:
        """批量对多张图像进行语义分割，极大提高 GPU 利用率"""
        if not frames or self._ensure_segmentation_model() is None:
            return [{"segmentation_used": False, "has_card": False, "has_hand": False,
                     "card_area_ratio": 0.0, "hand_area_ratio": 0.0, "card_bbox": None}] * len(frames)

        try:
            pil_images = [self._seg_pil_image.fromarray(cv2.cvtColor(f, cv2.COLOR_BGR2RGB)) for f in frames]
            device = next(self._seg_model.parameters()).device
            results = []

            # 分块批处理，防止显存 OOM（比如 16 帧一个 Batch）
            batch_size = 16
            for i in range(0, len(pil_images), batch_size):
                batch_imgs = pil_images[i: i + batch_size]

                inputs = self._seg_processor(images=batch_imgs, return_tensors="pt").to(device)

                with self._seg_torch.no_grad():
                    outputs = self._seg_model(**inputs)

                logits = outputs.logits

                # 批量上采样并取 argmax
                preds = self._seg_torch.nn.functional.interpolate(
                    logits,
                    size=batch_imgs[0].size[::-1],  # 假设所有帧分辨率一样
                    mode="bilinear",
                    align_corners=False,
                ).argmax(dim=1).cpu().numpy()

                # 解析每张图的 Mask
                for pred in preds:
                    card_mask = pred == settings.VIDEO_CARD_LABEL_ID
                    hand_mask = pred == settings.VIDEO_HAND_LABEL_ID

                    card_area = float(card_mask.mean()) if card_mask.size else 0.0
                    hand_area = float(hand_mask.mean()) if hand_mask.size else 0.0

                    card_bbox = self._largest_bbox(card_mask)
                    hand_bbox = self._largest_bbox(hand_mask)
                    focus_bbox = card_bbox if card_bbox is not None else hand_bbox

                    results.append({
                        "segmentation_used": True,
                        "has_card": card_area >= settings.VIDEO_MIN_CARD_AREA_RATIO,
                        "has_hand": hand_area >= settings.VIDEO_MIN_HAND_AREA_RATIO,
                        "card_area_ratio": card_area,
                        "hand_area_ratio": hand_area,
                        "card_bbox": focus_bbox,
                    })

                # 及时清理这批显存
                del inputs, outputs, logits, preds
                if self._seg_torch.cuda.is_available():
                    self._seg_torch.cuda.empty_cache()

            return results

        except Exception as exc:
            logger.warning(f"Batch segmentation failed, fallback: {exc}")
            return [{"segmentation_used": False, "has_card": False, "has_hand": False,
                     "card_area_ratio": 0.0, "hand_area_ratio": 0.0, "card_bbox": None}] * len(frames)

    def _analyze_segmentation(self, frame) -> dict[str, Any]:
        """对单帧图像进行语义分割分析，寻找卡片和手的区域"""
        if self._ensure_segmentation_model() is None:
            return {
                "segmentation_used": False,
                "has_card": False,
                "has_hand": False,
                "card_area_ratio": 0.0,
                "hand_area_ratio": 0.0,
                "card_bbox": None,
            }

        try:
            # OpenCV (BGR) 转换为 PIL 所需的 RGB
            rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            image = self._seg_pil_image.fromarray(rgb_frame)

            # 推理所需的数据需放到和模型同一设备上 (CPU or CUDA)
            device = next(self._seg_model.parameters()).device
            inputs = self._seg_processor(images=image, return_tensors="pt").to(device)

            with self._seg_torch.no_grad():
                outputs = self._seg_model(**inputs)

            logits = outputs.logits

            # 上采样回原始分辨率
            pred = self._seg_torch.nn.functional.interpolate(
                logits,
                size=image.size[::-1],
                mode="bilinear",
                align_corners=False,
            ).argmax(dim=1)[0].cpu().numpy()

            card_mask = pred == settings.VIDEO_CARD_LABEL_ID
            hand_mask = pred == settings.VIDEO_HAND_LABEL_ID

            card_area_ratio = float(card_mask.mean()) if card_mask.size else 0.0
            hand_area_ratio = float(hand_mask.mean()) if hand_mask.size else 0.0

            # [核心改进]: 提取两个 bbox
            card_bbox = self._largest_bbox(card_mask)
            hand_bbox = self._largest_bbox(hand_mask)

            has_card = card_area_ratio >= settings.VIDEO_MIN_CARD_AREA_RATIO
            has_hand = hand_area_ratio >= settings.VIDEO_MIN_HAND_AREA_RATIO

            # [核心改进]: 如果卡片太小/没切出来，但有手，把 focus box 降级到手的区域
            # 因为手大概率握着卡片，对“手”周边做 OCR 也能有效提取卡面信息
            focus_bbox = card_bbox if card_bbox is not None else hand_bbox

            # [核心改进]: 主动清理张量内存，防止长视频导致显存/内存溢出
            del inputs, outputs, logits, pred
            if self._seg_torch.cuda.is_available():
                self._seg_torch.cuda.empty_cache()

            return {
                "segmentation_used": True,
                "has_card": has_card,
                "has_hand": has_hand,
                "card_area_ratio": card_area_ratio,
                "hand_area_ratio": hand_area_ratio,
                "card_bbox": focus_bbox,  # 返回 fallback 后的 bbox
            }
        except Exception as exc:
            if not self._seg_runtime_warning_sent:
                logger.warning(f"Segmentation runtime failure, fallback enabled: {exc}")
                self._seg_runtime_warning_sent = True

            return {
                "segmentation_used": False,
                "has_card": False,
                "has_hand": False,
                "card_area_ratio": 0.0,
                "hand_area_ratio": 0.0,
                "card_bbox": None,
            }

    def _build_candidate(self, frame, current_time_ms: int, target_time_ms: int) -> FrameCandidate:
        """整合单帧的所有基础分析数据（分割、清晰度、存在感等），构建候选对象"""
        seg_result = self._analyze_segmentation(frame)

        # 对裁剪后的有效区域求清晰度，比算全图清晰度更准确，抗背景干扰
        focus_region = self._focus_region(frame, seg_result["card_bbox"])
        sharpness = self.get_laplacian_sharpness(focus_region)

        presence_score = self._compute_presence_score(
            segmentation_used=seg_result["segmentation_used"],
            has_card=seg_result["has_card"],
            has_hand=seg_result["has_hand"],
            card_area_ratio=seg_result["card_area_ratio"],
            hand_area_ratio=seg_result["hand_area_ratio"],
        )

        return FrameCandidate(
            frame=frame.copy(),
            time_ms=int(current_time_ms),
            sharpness=sharpness,
            time_weight=self.calculate_weight(current_time_ms, target_time_ms),
            segmentation_used=seg_result["segmentation_used"],
            has_card=seg_result["has_card"],
            has_hand=seg_result["has_hand"],
            card_area_ratio=seg_result["card_area_ratio"],
            hand_area_ratio=seg_result["hand_area_ratio"],
            card_bbox=seg_result["card_bbox"],
            presence_score=presence_score,
        )

    def _collect_candidates(
            self,
            cap: cv2.VideoCapture,
            start_time_ms: int,
            end_time_ms: int,
            target_time_ms: int,
            fps: float,
    ) -> list[FrameCandidate]:
        """在指定时间窗口内滑动，按步长收集视频帧作为候选"""
        candidates: list[FrameCandidate] = []
        raw_frames = []
        time_ms_list = []

        analysis_stride = self._analysis_stride(fps)

        # 预估最大读取次数，防止视频结尾卡死死循环
        max_reads = int((end_time_ms - start_time_ms) / 1000.0 * fps) + analysis_stride + 10

        # 跳转到起始时间 (注意: OpenCV 的 POS_MSEC 某些视频源上可能不精准)
        cap.set(cv2.CAP_PROP_POS_MSEC, start_time_ms)

        read_count = 0
        while read_count < max_reads:
            # 仅仅抓取下一帧的数据流，不进行耗时的图像解码
            ret = cap.grab()
            if not ret:
                break

            current_time_ms = cap.get(cv2.CAP_PROP_POS_MSEC)
            if current_time_ms > end_time_ms:
                break

            # 到达步长，才真正解码成图像矩阵
            if read_count % analysis_stride == 0:
                ret, frame = cap.retrieve()
                if ret:
                    raw_frames.append(frame.copy())
                    time_ms_list.append(current_time_ms)

            read_count += 1

        if not raw_frames:
            return []

        # 1. 批量过分割模型
        seg_results = self._batch_analyze_segmentation(raw_frames)

        # 2. 遍历组装 Candidate 并计算清晰度
        for frame, time_ms, seg_res in zip(raw_frames, time_ms_list, seg_results):
            # 切割关注区域算清晰度
            focus_region = self._focus_region(frame, seg_res["card_bbox"])
            sharpness = self.get_laplacian_sharpness(focus_region)

            presence_score = self._compute_presence_score(
                seg_res["segmentation_used"], seg_res["has_card"],
                seg_res["has_hand"], seg_res["card_area_ratio"], seg_res["hand_area_ratio"]
            )

            candidates.append(FrameCandidate(
                frame=frame,
                time_ms=int(time_ms),
                sharpness=sharpness,
                time_weight=self.calculate_weight(time_ms, target_time_ms),
                presence_score=presence_score,
                **seg_res  # 解包填入 has_card, card_bbox 等属性
            ))

        return candidates

    def _assign_dwell_scores(self, candidates: list[FrameCandidate]) -> None:
        """
        计算"停留得分" (Dwell Score):
        如果一个人抽到卡通常会停留展示一段时间。连续在帧中被检测到的实体，其停留得分会更高。
        这能有效过滤掉发牌时一晃而过的模糊残影。
        """
        if not candidates or not any(candidate.segmentation_used for candidate in candidates):
            return

        target_frames = max(1, int(round(settings.VIDEO_DWELL_TARGET_SECONDS * self.analysis_fps)))
        index = 0

        while index < len(candidates):
            if not candidates[index].is_present:
                index += 1
                continue

            # 寻找连续出现(is_present = True)的片段
            run_end = index
            while run_end + 1 < len(candidates) and candidates[run_end + 1].is_present:
                run_end += 1

            run_length = run_end - index + 1
            dwell_score = min(run_length / target_frames, 1.0)

            # 为这段连续的帧赋相同的停留分
            for pos in range(index, run_end + 1):
                candidates[pos].dwell_score = dwell_score

            index = run_end + 1

    def _normalize_sharpness(self, sharpness: float, max_sharpness: float) -> float:
        """对清晰度进行对数归一化处理。使用 log 防止极值（超锐化噪点）拉爆分数池"""
        if sharpness <= 0 or max_sharpness <= 0:
            return 0.0

        denominator = math.log1p(max_sharpness)
        if denominator <= 0:
            return 0.0
        return min(math.log1p(sharpness) / denominator, 1.0)

    def _normalize_text(self, text: str) -> str:
        """清洗文本：去点，全大写，仅保留英文、数字、中文字符"""
        if not text:
            return ""
        cleaned = text.replace(".", "")
        cleaned = re.sub(r"[^0-9A-Za-z\u4e00-\u9fff]+", " ", cleaned.upper())
        return re.sub(r"\s+", " ", cleaned).strip()

    def _tokenize_text(self, text: str) -> list[str]:
        """将文本拆分为分词列表，去重去单字母（除非是数字）"""
        normalized = self._normalize_text(text)
        if not normalized:
            return []

        tokens: list[str] = []
        seen: set[str] = set()
        for token in normalized.split():
            if len(token) == 1 and not token.isdigit():
                continue
            if token in seen:
                continue
            seen.add(token)
            tokens.append(token)
        return tokens

    def _build_expected_text(self, card: CardInfoOutput) -> dict[str, Any]:
        """从输入的 json 信息中提取期望的卡片名字、系列号、编号等，作为 OCR 的对比基准"""
        name_tokens = self._tokenize_text(card.card_name_en or "")
        if not name_tokens and card.card_name_cn:
            name_tokens = self._tokenize_text(card.card_name_cn)

        all_series_tokens = self._tokenize_text(card.series or "")
        number_tokens = [token for token in all_series_tokens if token.isdigit()][:3]
        series_tokens = [token for token in all_series_tokens if not token.isdigit()]
        series_tokens.sort(key=len, reverse=True)

        return {
            "name_tokens": name_tokens[:4],
            "series_tokens": series_tokens[:6],
            "number_tokens": number_tokens,
            "has_expectation": bool(name_tokens or series_tokens or number_tokens),
        }

    def _extract_ocr_text(self, ocr_result: Any) -> str:
        """递归解析 RapidOCR 返回的复杂嵌套结构，将所有识别出的文本段落拼装成一个大字符串"""
        texts: list[str] = []

        def visit(node: Any) -> None:
            if node is None:
                return
            if isinstance(node, str):
                stripped = node.strip()
                if stripped:
                    texts.append(stripped)
                return
            if hasattr(node, "txts"):
                visit(getattr(node, "txts"))
                return
            if hasattr(node, "ocr_res"):
                visit(getattr(node, "ocr_res"))
                return
            if isinstance(node, dict):
                for value in node.values():
                    visit(value)
                return
            if isinstance(node, (list, tuple)):
                if len(node) >= 2 and isinstance(node[1], str):
                    visit(node[1])
                    return
                for item in node:
                    visit(item)

        visit(ocr_result)

        deduped: list[str] = []
        seen: set[str] = set()
        for text in texts:
            if text in seen:
                continue
            seen.add(text)
            deduped.append(text)
        return " ".join(deduped)

    def _token_overlap_score(self, expected_tokens: list[str], ocr_tokens: list[str]) -> float:
        """计算期望 Token 和 OCR 识别 Token 之间的重叠得分，包含对子串（部分匹配）的兼容分"""
        if not expected_tokens or not ocr_tokens:
            return 0.0

        score = 0.0
        ocr_set = set(ocr_tokens)
        for token in expected_tokens:
            best_ratio = 0.0
            for other in ocr_tokens:
                # 计算字符串相似度 (0 到 1)
                ratio = difflib.SequenceMatcher(None, token, other).ratio()
                if ratio > best_ratio:
                    best_ratio = ratio

            if best_ratio > 0.85:
                score += 1.0  # 相似度极高，视为完全命中
            elif best_ratio > 0.6:
                score += 0.6  # 存在一定错别字，给部分分

        return min(score / len(expected_tokens), 1.0)

    def _score_ocr_match(self, ocr_text: str, expected: dict[str, Any]) -> float:
        """综合评判 OCR 识别文本与目标 JSON 信息的多维度匹配程度"""
        if not ocr_text or not expected["has_expectation"]:
            return 0.0

        normalized_text = self._normalize_text(ocr_text)
        ocr_tokens = self._tokenize_text(ocr_text)
        if not ocr_tokens:
            return 0.0

        name_tokens = expected["name_tokens"]
        series_tokens = expected["series_tokens"]
        number_tokens = expected["number_tokens"]

        name_score = self._token_overlap_score(name_tokens, ocr_tokens)
        if name_tokens:
            joined_name = " ".join(name_tokens)
            if joined_name and joined_name in normalized_text:
                name_score = 1.0  # 名字完全作为整体匹配上，直接满分

        series_score = self._token_overlap_score(series_tokens, ocr_tokens)
        number_score = self._token_overlap_score(number_tokens, ocr_tokens)

        # 加权混合：卡片名字(60%) > 系列名(25%) > 卡片编号(15%)
        if name_tokens:
            return min(0.60 * name_score + 0.25 * series_score + 0.15 * number_score, 1.0)
        return min(0.65 * series_score + 0.35 * number_score, 1.0)

    def _run_ocr(self, frame, bbox: Optional[tuple[int, int, int, int]]) -> str:
        """调用 OCR 引擎对关注区域进行文本识别，加入抗弹幕干扰的分块策略"""
        engine = self._ensure_ocr_engine()
        if engine is None:
            return ""

        focus_region = self._focus_region(frame, bbox)
        if focus_region is None or focus_region.shape[0] == 0 or focus_region.shape[1] == 0:
            return ""

        texts: list[str] = []

        # 1. 常规全图 OCR (可能被中间的弹幕压制，但能提取出分散的编号等)
        try:
            result_full = engine(focus_region)
            texts.append(self._extract_ocr_text(result_full))
        except Exception as exc:
            if not self._ocr_runtime_warning_sent:
                logger.warning(f"OCR full region failure: {exc}")
                self._ocr_runtime_warning_sent = True

        # 2. 分块特写 OCR，避开中心弹幕区，降低识别阈值
        h, w = focus_region.shape[:2]
        if h > 60 and w > 60:
            # A. 专门识别底部 40% (绝大多数球星卡球员名字、宝可梦卡信息在底部)
            try:
                bottom_half = focus_region[int(h * 0.6):h, :]
                result_bottom = engine(bottom_half)
                texts.append(self._extract_ocr_text(result_bottom))
            except Exception:
                pass

            # B. 专门识别顶部 40% (通常有 Bowman 1st 标志、帕尼尼系列名、或宝可梦名字)
            try:
                top_half = focus_region[0:int(h * 0.4), :]
                result_top = engine(top_half)
                texts.append(self._extract_ocr_text(result_top))
            except Exception:
                pass

        # 将全图、顶部、底部的识别结果合并（后续的 token_overlap_score 会自动处理去重）
        combined_text = " ".join(texts)

        # 3. 正则剔除常见的直播间高频干扰词 (防止误匹配)
        # 这里的词汇通常是海外拆卡直播间(Whatnot/TikTok)经常出现的系统提示语
        ignore_words = r"(?i)\b(bought|break|hobby|jumbo|box|close|spot|nice|snack|packs)\b"
        combined_text = re.sub(ignore_words, " ", combined_text)

        return combined_text

    def _score_candidates(
            self,
            candidates: list[FrameCandidate],
            card_output: CardInfoOutput,
    ) -> None:
        """
        核心打分中枢：结合之前计算的各个单项分，得出最终排名分。
        采用二次打分机制：先通过 Base Score 选出 Top K，再让 Top K 过一遍耗时的 OCR，得出 Final Score。
        """
        if not candidates:
            return

        self._assign_dwell_scores(candidates)

        # 只对画面里确认有卡/手的帧进行打分
        scoring_candidates = [candidate for candidate in candidates if candidate.is_present]
        if not scoring_candidates:
            scoring_candidates = candidates

        # 改为使用 90 分位数，防止单帧反光噪点拉爆整个分数池
        if scoring_candidates:
            sharpnesses = [c.sharpness for c in scoring_candidates]
            max_sharpness = float(np.percentile(sharpnesses, 90))
        else:
            max_sharpness = 0.0
        segmentation_used = any(candidate.segmentation_used for candidate in candidates)

        expected = self._build_expected_text(card_output)
        ocr_enabled = self._ensure_ocr_engine() is not None and expected["has_expectation"]

        # 1. 粗排：计算 Base Score
        for candidate in scoring_candidates:
            candidate.sharpness_score = self._normalize_sharpness(candidate.sharpness, max_sharpness)
            if segmentation_used:
                # 若启用了图像分割：存在感(40%) + 清晰度(25%) + 离目标时间点的近度(20%) + 画面稳定性(15%)
                candidate.base_score = (
                        0.40 * candidate.presence_score
                        + 0.25 * candidate.sharpness_score
                        + 0.20 * candidate.time_weight
                        + 0.15 * candidate.dwell_score
                )
            else:
                # fallback: 没有分割模型，只能靠清晰度和时间权重
                candidate.base_score = (
                        0.55 * candidate.sharpness_score
                        + 0.35 * candidate.time_weight
                        + 0.10 * candidate.dwell_score
                )

        # 2. 精排：使用 OCR 计算 Final Score
        if ocr_enabled:
            # 只有 Base Score 排名前 K 的优胜者才会执行 OCR（性能优化）
            top_candidates = sorted(
                scoring_candidates,
                key=lambda item: item.base_score,
                reverse=True,
            )[: max(1, settings.VIDEO_OCR_TOP_K)]

            for candidate in top_candidates:
                candidate.ocr_text = self._run_ocr(candidate.frame, candidate.card_bbox)
                candidate.ocr_score = self._score_ocr_match(candidate.ocr_text, expected)

            # 更新所有入围帧的 Final Score
            for candidate in scoring_candidates:
                if segmentation_used:
                    # OCR占核心大头(40%)，配合其他物理指标
                    candidate.final_score = (
                            0.40 * candidate.ocr_score
                            + 0.25 * candidate.presence_score
                            + 0.20 * candidate.sharpness_score
                            + 0.10 * candidate.time_weight
                            + 0.05 * candidate.dwell_score
                    )
                else:
                    candidate.final_score = (
                            0.45 * candidate.ocr_score
                            + 0.30 * candidate.sharpness_score
                            + 0.20 * candidate.time_weight
                            + 0.05 * candidate.dwell_score
                    )
        else:
            # 如果 OCR 不可用或没配置预期，则直接用 Base Score 作为终局分数
            for candidate in scoring_candidates:
                candidate.final_score = candidate.base_score

    def _select_best_candidate(
            self,
            candidates: list[FrameCandidate],
            target_time_ms: int,
    ) -> Optional[FrameCandidate]:
        """选出最终最能代表"高光时刻"的帧"""
        if not candidates:
            return None

        # 核心逻辑：主比对 final_score；如果最终分一样(比如都为0)，看清晰度；再一样，看谁离打点时间最近。
        return max(
            candidates,
            key=lambda item: (
                item.final_score,
                item.sharpness_score,
                -abs(item.time_ms - target_time_ms),
            ),
        )

    def capture_frames(self, video_path: str, cards: list[CardInfoInput]) -> list[CardInfoOutput]:
        """业务主干：传入视频与目标卡片打点列表，输出高光图片及匹配信息"""
        if not os.path.exists(video_path):
            logger.error(f"Video file not found: {video_path}")
            raise FileNotFoundError(f"Video file not found: {video_path}")

        logger.info(f"Open video: {video_path}")
        logger.info(f"Cards to process: {len(cards)}")

        cap = cv2.VideoCapture(video_path)
        fps = cap.get(cv2.CAP_PROP_FPS)
        if fps <= 0:
            fps = 30.0

        output_list: list[CardInfoOutput] = []
        success_count = 0
        filtered_count = 0

        for idx, card_input in enumerate(cards):
            card_output = CardInfoOutput(**card_input.dict())
            target_time_ms = self.time_str_to_ms(card_output.time)

            # 以打点时间戳为锚，建立一个[过去几s 到 未来 几s] 的搜索窗口
            start_time_ms = max(0, target_time_ms - self.search_before_ms)
            end_time_ms = target_time_ms + self.search_after_ms

            logger.info(
                f"[{idx + 1}/{len(cards)}] analyze {card_output.time} "
                f"for {card_output.card_name_cn or card_output.card_name_en or 'unknown'}"
            )
            logger.info(f"  search window: [{start_time_ms}ms ~ {end_time_ms}ms]")

            # 1. 在窗口内收集所有候选帧
            candidates = self._collect_candidates(
                cap=cap,
                start_time_ms=start_time_ms,
                end_time_ms=end_time_ms,
                target_time_ms=target_time_ms,
                fps=fps,
            )

            if not candidates:
                logger.warning("  no frames sampled in the target window")
                continue

            segmentation_used = any(candidate.segmentation_used for candidate in candidates)
            present_candidates = [candidate for candidate in candidates if candidate.is_present]

            # [需求点 1]: 如果使用了分割模型且这片窗口内完全找不到手/卡，直接判定无效数据
            if segmentation_used and not present_candidates:
                filtered_count += 1
                logger.info("  filtered out: no card/hand found around the timestamp")
                continue

            scoring_candidates = present_candidates if present_candidates else candidates

            # 2. 调用多维度评分枢纽给各个候选帧打分
            self._score_candidates(candidates, card_output)

            # 3. 选出最匹配、最清晰的一张
            best_candidate = self._select_best_candidate(scoring_candidates, target_time_ms)

            if best_candidate is None:
                logger.warning("  no usable candidate after scoring")
                continue

            # 4. 保存为 JPG，构造业务输出数据
            filename = f"{uuid.uuid4()}_{best_candidate.time_ms}.jpg"
            save_path = os.path.join(settings.FRAMES_DIR, filename)

            try:
                cv2.imwrite(save_path, best_candidate.frame)
                image_url = f"{settings.BASE_URL}/static/frames/{filename}"
                card_output.frame_image_path = image_url
                output_list.append(card_output)
                success_count += 1

                time_diff = (best_candidate.time_ms - target_time_ms) / 1000.0
                logger.info(
                    f"  saved {filename} "
                    f"(offset={time_diff:+.2f}s, sharpness={best_candidate.sharpness:.1f}, "
                    f"presence={best_candidate.presence_score:.2f}, "
                    f"ocr={best_candidate.ocr_score:.2f}, score={best_candidate.final_score:.2f})"
                )
            except Exception as exc:
                logger.error(f"  failed to save frame: {exc}")

        # 务必释放 OpenCV 句柄，避免被视频文件死锁
        cap.release()
        logger.info(
            f"Frame capture finished. saved={success_count}, "
            f"filtered={filtered_count}, total={len(cards)}"
        )
        return output_list