highlight_filter_agent.py

"""
highlight_filter_agent.py —— HighlightFilter Agent

项目：《面向无字幕赛事录像的体育视频自动字幕与高光剪辑多智能体系统》
职责：
    1. 读取 MediaParseOutput（音频特征 + 字幕），执行双层融合高光筛选。
    2. 第一层：音频响度突增候选框 + 字幕关键词匹配候选框，窗口并集融合。
    3. 第二层：多模态评分（audio_score / text_score / combined_score）。
    4. 第三层：轻量 LLM 语义二次校验，过滤广告/采访/无效片段。
    5. 输出标准化 HighlightFilterOutput，供 EditPlanner 读取。

设计原则：
    - 零跨模块对象传递：只读写本地 JSON 文件。
    - 关键词兼容模糊匹配：应对 Whisper 错别字（如 "三芬"→"三分"）。
    - LLM 兜底但不强依赖：若 API 不可用，保留全部候选并告警。
    - 允许纯文本触发：客队进球时观众安静，关键词必须能独立触发高光。
"""

from __future__ import annotations

import json
import logging
import os
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple

import requests

from config import settings
from pydantic import BaseModel, Field, ConfigDict

from schema import (
    AudioFeatureSegment,
    HighlightCandidate,
    HighlightFilterOutput,
    HighlightTrigger,
    MediaParseOutput,
    SubtitleSegment,
    save_to_json,
)

# ---------------------------------------------------------------------------
# 日志配置
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=getattr(logging, settings.log_level, logging.INFO),
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("HighlightFilterAgent")


# ---------------------------------------------------------------------------
# 异常类
# ---------------------------------------------------------------------------
class HighlightFilterError(Exception):
    """HighlightFilter Agent 通用异常基类。"""
    pass


class LLMValidationError(HighlightFilterError):
    """LLM 语义校验阶段异常（网络超时、返回格式错误等）。"""
    pass


# ---------------------------------------------------------------------------
# 常量：进球触发词（用于语义化精剪）
# ---------------------------------------------------------------------------
_GOAL_TRIGGER_PATTERNS = [
    # 英文篮球（注意：避免 \b 与标点冲突，使用 (^|\s) 和 (\s|$|!|\?|\.) 作为边界）
    r"(?:^|\s)(it's good|it's going|bang!|downtown!|from downtown|three point|3-pointer|three)(?:\s|$|!|\?|\.)",
    r"(?:^|\s)(goal!|scores?|makes? it|what a finish|equaliser|equalizer|penalty)(?:\s|$|!|\?|\.)",
    # 中文篮球
    r"(?:^|\s)(好球!|进了!|进球!|破门!|点球!|三分!|绝杀!)(?:\s|$|!|\?|\.)",
]

# These terms are direct scoring/play evidence.  A language model may judge a
# short caption too vague, but it must not discard an explicit "three" or
# "dunk" call and leave the montage with a single clip.
_STRONG_EVENT_KEYWORDS = {
    "three", "downtown", "3-pointer", "three point", "dunk", "layup", "jumper", "floater", "hook shot",
    "two point", "up and good", "gets it to go", "goal", "penalty", "red card", "yellow card",
    "offside", "corner", "save", "miss", "missed", "no good", "brick", "airball", "comes up short",
    "scores", "score", "makes", "bang", "buzzer",
    "三分", "扣篮", "进球", "点球", "红牌", "黄牌", "越位", "扑救", "绝杀",
}

# A scored two-pointer and a miss are both meaningful requests in this
# application, even when a general-purpose LLM considers the one-line call
# too mundane for a generic "highlight" reel.  Preserve these calls so the
# later structured editor can accurately honour requests such as "只剪两分"
# and "只剪打铁".
_PROTECTED_BASKETBALL_PLAY = re.compile(
    r"\b(three|3[- ]?pointer|dunk|layup|jumper|floater|hook shot|two[- ]?point|"
    r"miss(?:es|ed)?|no good|comes up short|won'?t go|off the rim|rimmed out|"
    r"brick|air ?ball|up and good|gets? it to go|lays? it in|finishes)\b|"
    r"三分|两分|上篮|跳投|抛投|打铁|不中|没进|命中",
    re.I,
)

# ---------------------------------------------------------------------------
# 语义化精剪动态参数
# ---------------------------------------------------------------------------
# 基础保留时长（秒）
_BASE_PRE_GOAL_SEC = 3.0   # 触发词前基础保留
_BASE_POST_GOAL_SEC = 7.0  # 触发词后基础保留：保留落网、欢呼与运动员反应

# 响度动态调整系数
# 原理：响度突增越大（crowd noise 越大），保留越多内容（欢呼是氛围的一部分）
# peak_dbfs 范围约 [-30, -10]，映射到 [0, 1] 的响度强度
_LOUDNESS_DYNAMIC_MIN_DBFS = -35.0
_LOUDNESS_DYNAMIC_MAX_DBFS = -15.0

# 动态调整范围（在基础时长上增减）
_DYNAMIC_RANGE_PRE = 2.0   # 前最多 +/- 2s
_DYNAMIC_RANGE_POST = 3.0  # 后最多 +/- 3s

# 精剪后时长硬边界
_TRIM_MIN_DURATION = 3.0
_TRIM_MAX_DURATION = 14.0


# ---------------------------------------------------------------------------
# 内部数据结构：原始候选框（融合前）
# ---------------------------------------------------------------------------
@dataclass
class _RawCandidate:
    """融合前的原始候选框，仅用于内部计算。"""
    start_sec: float
    end_sec: float
    source: str  # "audio" | "text"
    # 音频相关
    surge_segments: List[AudioFeatureSegment] = None
    # 文本相关
    matched_subtitles: List[Tuple[SubtitleSegment, List[str]]] = None

    def __post_init__(self):
        if self.surge_segments is None:
            self.surge_segments = []
        if self.matched_subtitles is None:
            self.matched_subtitles = []


# ---------------------------------------------------------------------------
# 工具函数：关键词匹配
# ---------------------------------------------------------------------------
def _normalize_text(text: str) -> str:
    """
    文本归一化：转小写、去除多余空白、去除常见标点。

    Args:
        text: 原始字幕文本。

    Returns:
        str: 归一化后的文本。
    """
    text = text.lower().strip()
    # 保留字母、数字、中文、空格，其余替换为空格
    text = re.sub(r"[^\w\s\u4e00-\u9fff]", " ", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text


def _fuzzy_match_score(query: str, text: str) -> float:
    """
    计算 query 与 text 的模糊匹配得分（基于最长公共子串比例）。

    Args:
        query: 关键词（已归一化）。
        text: 待匹配文本（已归一化）。

    Returns:
        float: 相似度 [0.0, 1.0]。
    """
    if not query or not text:
        return 0.0
    if query in text:
        return 1.0

    # 最长公共子串（LCS）长度
    m, n = len(query), len(text)
    # 使用一维 DP 节省内存
    prev = [0] * (n + 1)
    max_len = 0
    for i in range(1, m + 1):
        curr = [0] * (n + 1)
        for j in range(1, n + 1):
            if query[i - 1] == text[j - 1]:
                curr[j] = prev[j - 1] + 1
                max_len = max(max_len, curr[j])
            else:
                curr[j] = 0
        prev = curr

    return max_len / max(m, n)


def _match_keywords_in_text(
    text: str,
    keywords: List[str],
    fuzzy_threshold: float = settings.keyword_fuzzy_threshold,
) -> List[str]:
    """
    在单条字幕文本中匹配关键词列表，支持模糊匹配。

    Args:
        text: 字幕文本。
        keywords: 关键词列表。
        fuzzy_threshold: 模糊匹配阈值。

    Returns:
        List[str]: 命中的关键词列表（去重）。
    """
    normalized = _normalize_text(text)
    if not normalized:
        return []

    matched: List[str] = []
    for kw in keywords:
        kw_norm = _normalize_text(kw)
        if not kw_norm:
            continue
        # 精确匹配或模糊匹配
        if kw_norm in normalized or _fuzzy_match_score(kw_norm, normalized) >= fuzzy_threshold:
            matched.append(kw)

    # 去重并保持顺序
    seen: Set[str] = set()
    result: List[str] = []
    for kw in matched:
        if kw not in seen:
            seen.add(kw)
            result.append(kw)
    return result


# ---------------------------------------------------------------------------
# 第一层：候选框提取
# ---------------------------------------------------------------------------
def _extract_audio_candidates(
    segments: List[AudioFeatureSegment],
    total_duration_sec: Optional[float] = None,
) -> List[_RawCandidate]:
    """
    从音频特征片段中提取响度突增候选框。

    规则：
        - 遍历所有 segment，收集 loudness_surge_flag=True 的片段。
        - 相邻（间隔 <= 2 秒）的突增片段合并为一个候选框。

    Args:
        segments: AudioFeatureSegment 列表。

    Returns:
        List[_RawCandidate]: 音频候选框列表。
    """
    # 先过滤出突增片段
    surge_segs = [seg for seg in segments if seg.loudness_surge_flag]
    if not surge_segs:
        return []

    # 按时间排序
    surge_segs.sort(key=lambda s: s.start_sec)

    candidates: List[_RawCandidate] = []
    current = _RawCandidate(
        start_sec=surge_segs[0].start_sec,
        end_sec=surge_segs[0].end_sec,
        source="audio",
        surge_segments=[surge_segs[0]],
    )

    GAP_THRESHOLD = 2.0  # 相邻合并阈值（秒）

    for seg in surge_segs[1:]:
        if seg.start_sec - current.end_sec <= GAP_THRESHOLD:
            # 合并
            current.end_sec = max(current.end_sec, seg.end_sec)
            current.surge_segments.append(seg)
        else:
            candidates.append(current)
            current = _RawCandidate(
                start_sec=seg.start_sec,
                end_sec=seg.end_sec,
                source="audio",
                surge_segments=[seg],
            )
    candidates.append(current)

    # A crowd peak often happens at the basket, but the replay-worthy reaction
    # happens immediately afterwards.  Expand both sides before planning so a
    # clip cannot end at the exact moment the ball goes in.
    for candidate in candidates:
        candidate.start_sec = max(0.0, candidate.start_sec - 3.0)
        candidate.end_sec += 8.0
        if total_duration_sec is not None:
            candidate.end_sec = min(candidate.end_sec, total_duration_sec)

    logger.info(f"音频突增候选框: {len(candidates)} 个")
    for c in candidates:
        logger.debug(
            f"  音频框 [{c.start_sec:.1f}s-{c.end_sec:.1f}s] "
            f"包含 {len(c.surge_segments)} 个突增段"
        )

    return candidates


def _extract_text_candidates(
    subtitles: List[SubtitleSegment],
    keywords: List[str],
    expand_sec: float = 8.0,
) -> List[_RawCandidate]:
    """
    从字幕中提取关键词命中候选框。

    规则：
        - 遍历字幕，匹配关键词。
        - 命中的字幕时间区间前后各外扩 expand_sec 秒。
        - 相邻（间隔 <= 2 秒）的候选框合并。

    Args:
        subtitles: SubtitleSegment 列表。
        keywords: 关键词列表。
        expand_sec: 前后外扩时长（秒）。

    Returns:
        List[_RawCandidate]: 文本候选框列表。
    """
    hits: List[Tuple[float, float, SubtitleSegment, List[str]]] = []

    for sub in subtitles:
        matched = _match_keywords_in_text(sub.text, keywords)
        if matched:
            # 前后外扩
            start = max(0.0, sub.start_sec - expand_sec)
            end = sub.end_sec + expand_sec
            hits.append((start, end, sub, matched))

    if not hits:
        return []

    # 按开始时间排序
    hits.sort(key=lambda x: x[0])

    candidates: List[_RawCandidate] = []
    current_start, current_end = hits[0][0], hits[0][1]
    current_subs: List[Tuple[SubtitleSegment, List[str]]] = [(hits[0][2], hits[0][3])]

    GAP_THRESHOLD = 2.0

    for start, end, sub, matched in hits[1:]:
        if start - current_end <= GAP_THRESHOLD:
            # 合并
            current_end = max(current_end, end)
            current_subs.append((sub, matched))
        else:
            candidates.append(_RawCandidate(
                start_sec=current_start,
                end_sec=current_end,
                source="text",
                matched_subtitles=current_subs,
            ))
            current_start, current_end = start, end
            current_subs = [(sub, matched)]

    candidates.append(_RawCandidate(
        start_sec=current_start,
        end_sec=current_end,
        source="text",
        matched_subtitles=current_subs,
    ))

    logger.info(f"文本关键词候选框: {len(candidates)} 个")
    for c in candidates:
        kws = set()
        for _, m in c.matched_subtitles:
            kws.update(m)
        logger.debug(
            f"  文本框 [{c.start_sec:.1f}s-{c.end_sec:.1f}s] "
            f"关键词: {', '.join(kws)}"
        )

    return candidates


# ---------------------------------------------------------------------------
# 窗口融合算法（Window Fusion）
# ---------------------------------------------------------------------------
def _fuse_candidates(
    audio_candidates: List[_RawCandidate],
    text_candidates: List[_RawCandidate],
    gap_threshold: float = 2.0,
) -> List[_RawCandidate]:
    """
    将音频与文本候选框进行并集融合。

    规则：
        - 所有候选框按开始时间排序。
        - 重叠或相邻（间隔 <= gap_threshold）的框合并。
        - 合并后的框同时继承音频和文本证据。

    Args:
        audio_candidates: 音频候选框列表。
        text_candidates: 文本候选框列表。
        gap_threshold: 相邻合并阈值（秒）。

    Returns:
        List[_RawCandidate]: 融合后的候选框列表。
    """
    all_boxes = audio_candidates + text_candidates
    if not all_boxes:
        return []

    # 按开始时间排序
    all_boxes.sort(key=lambda b: b.start_sec)

    fused: List[_RawCandidate] = []
    current = _RawCandidate(
        start_sec=all_boxes[0].start_sec,
        end_sec=all_boxes[0].end_sec,
        source=all_boxes[0].source,
        surge_segments=list(all_boxes[0].surge_segments),
        matched_subtitles=list(all_boxes[0].matched_subtitles),
    )

    for box in all_boxes[1:]:
        if box.start_sec - current.end_sec <= gap_threshold:
            # 合并
            current.end_sec = max(current.end_sec, box.end_sec)
            current.source = "fused"  # 混合来源
            current.surge_segments.extend(box.surge_segments)
            current.matched_subtitles.extend(box.matched_subtitles)
        else:
            fused.append(current)
            current = _RawCandidate(
                start_sec=box.start_sec,
                end_sec=box.end_sec,
                source=box.source,
                surge_segments=list(box.surge_segments),
                matched_subtitles=list(box.matched_subtitles),
            )
    fused.append(current)

    logger.info(f"融合后候选框: {len(fused)} 个")
    for f in fused:
        logger.debug(
            f"  融合框 [{f.start_sec:.1f}s-{f.end_sec:.1f}s] "
            f"source={f.source} audio={len(f.surge_segments)} text={len(f.matched_subtitles)}"
        )

    return fused


# ---------------------------------------------------------------------------
# 第二层：多模态评分
# ---------------------------------------------------------------------------
def _dbfs_to_energy(dbfs: float) -> float:
    """
    将 dBFS 对数刻度转换为线性能量域。

    公式：energy = 10^(dbfs / 20)
    0 dBFS → 1.0（满幅），-60 dBFS → 0.001

    Args:
        dbfs: dBFS 值。

    Returns:
        float: 线性能量值（> 0）。
    """
    return 10.0 ** (dbfs / 20.0)


def _compute_audio_score(
    candidate: _RawCandidate,
    global_mean_loudness: float,
    min_dbfs: float = settings.loudness_min_dbfs,
) -> float:
    """
    计算音频维度得分（能量域归一化，避免 dBFS 对数尺度线性运算误差）。

    算法：
        - 若无突增段，返回 0.0。
        - 取候选框内所有突增段的 peak_loudness_dbfs 最大值。
        - 绝对阈值拦截：若 peak < min_dbfs（默认 -35 dBFS），直接返回 0.0，
          避免静音视频里正常语音被误判为高分。
        - 能量域归一化：将对数 dBFS 转换为线性能量后计算相对位置。
          score = (energy_peak - energy_min) / (energy_0dbfs - energy_min)
          其中 energy_0dbfs = 1.0（满幅能量）。

    Args:
        candidate: 融合候选框。
        global_mean_loudness: 全片平均响度（dBFS），保留用于日志/调试，不参与计算。
        min_dbfs: 响度绝对下限（dBFS），低于此值视为无效突增。

    Returns:
        float: [0.0, 1.0]
    """
    if not candidate.surge_segments:
        return 0.0

    peak = max(seg.peak_loudness_dbfs for seg in candidate.surge_segments)

    # 绝对阈值拦截：低于 min_dbfs 的突增视为无效（可能是静音基线上的噪声）
    if peak < min_dbfs:
        logger.debug(
            f"  音频得分=0.0（peak={peak:.1f} dBFS < 阈值={min_dbfs:.1f} dBFS）"
        )
        return 0.0

    # 能量域归一化：避免在 quiet 视频中 -25 dBFS 相对 -85 dBFS 产生 2.0+ 的异常高分
    energy_peak = _dbfs_to_energy(peak)
    energy_min = _dbfs_to_energy(min_dbfs)
    energy_max = 1.0  # 0 dBFS 对应的线性能量

    score = (energy_peak - energy_min) / (energy_max - energy_min)
    return round(max(min(score, 1.0), 0.0), 3)


def _compute_text_score(
    candidate: _RawCandidate,
) -> float:
    """
    计算文本维度得分。

    算法：
        - 若无关键词命中，返回 0.0。
        - 统计命中关键词数量（去重）。
        - 考虑匹配质量（精确匹配 > 模糊匹配）。
        - 归一化：score = min(count / 3.0, 1.0)。
          （假设 3 个不同关键词为满分）

    Args:
        candidate: 融合候选框。

    Returns:
        float: [0.0, 1.0]
    """
    if not candidate.matched_subtitles:
        return 0.0

    unique_keywords: Set[str] = set()
    exact_match_count = 0

    for sub, matched_kws in candidate.matched_subtitles:
        normalized_sub = _normalize_text(sub.text)
        for kw in matched_kws:
            unique_keywords.add(kw)
            kw_norm = _normalize_text(kw)
            if kw_norm in normalized_sub:
                exact_match_count += 1

    # 基础分：关键词数量
    base_score = len(unique_keywords) / 3.0

    # 精确匹配加成
    bonus = exact_match_count * 0.1

    score = min(base_score + bonus, 1.0)
    return round(max(score, 0.0), 3)


def _determine_trigger(candidate: _RawCandidate) -> HighlightTrigger:
    """
    根据候选框的证据来源确定触发类型。

    Args:
        candidate: 融合候选框。

    Returns:
        HighlightTrigger: AUDIO_SURGE / KEYWORD_MATCH / FUSED
    """
    has_audio = len(candidate.surge_segments) > 0
    has_text = len(candidate.matched_subtitles) > 0

    if has_audio and has_text:
        return HighlightTrigger.FUSED
    elif has_audio:
        return HighlightTrigger.AUDIO_SURGE
    else:
        return HighlightTrigger.KEYWORD_MATCH


# ---------------------------------------------------------------------------
# 第三层：LLM 语义二次校验
# ---------------------------------------------------------------------------
def _build_llm_prompt(subtitle_texts: List[str]) -> str:
    """
    构建 LLM 校验 Prompt。

    将高光时间窗口内【所有连续字幕文本】完整拼接为解说词上下文，
    避免仅投喂离散关键词句导致 LLM 语义断层误判。

    Args:
        subtitle_texts: 该高光窗口内按时间顺序排列的全部字幕文本列表。

    Returns:
        str: 完整的 Prompt 字符串。
    """
    combined_text = "\n".join(f"[{i+1}] {t}" for i, t in enumerate(subtitle_texts))
    prompt = (
        "你是一个资深篮球/足球球评。请阅读以下完整的解说词片段，判断其中是否包含"
        "进球、绝杀、精彩过人、强烈身体对抗等核心赛事高光。"
        "注意：客队进球时观众可能很安静，请完全依据解说文本的兴奋度和客观事实判断。"
        "请直接返回 JSON 格式，不要有任何其他内容：\n\n"
        f"{combined_text}\n\n"
        '返回格式：{"is_highlight": true/false, "reason": "简短的中文过滤或保留理由"}'
    )
    return prompt


class _LLMResponse(BaseModel):
    """
    LLM 语义校验返回的 JSON 结构约束。

    使用 Pydantic v2 强类型校验，杜绝 "yes"/"no" 字符串被 bool() 误转为 True 的问题。
    """
    is_highlight: bool = Field(
        description="是否为有效高光，必须是布尔值 true/false，不接受字符串"
    )
    reason: str = Field(
        default="无说明",
        description="校验理由，若 is_highlight=false 则说明过滤原因"
    )

    model_config = ConfigDict(strict=True)


def _parse_llm_json(content: str) -> _LLMResponse:
    """
    将 LLM 返回的 JSON 字符串解析为强类型 _LLMResponse。

    处理常见 LLM 输出陷阱：
        - "yes"/"no" 字符串 → 显式映射为 bool
        - 数字 1/0 → 映射为 bool
        - 字段缺失 → 使用默认值并告警

    Args:
        content: LLM 返回的原始 JSON 字符串。

    Returns:
        _LLMResponse: 校验后的结构化对象。

    Raises:
        LLMValidationError: 当 JSON 格式严重损坏无法挽救时抛出。
    """
    # 先尝试原生 Pydantic 严格解析
    try:
        return _LLMResponse.model_validate_json(content)
    except Exception:
        pass  # 降级到容错逻辑

    # 容错：先解析为 dict，再手动清洗
    try:
        data = json.loads(content)
    except json.JSONDecodeError as exc:
        raise LLMValidationError(f"LLM 返回非合法 JSON: {exc}") from exc

    if not isinstance(data, dict):
        raise LLMValidationError(f"LLM 返回 JSON 根类型为 {type(data).__name__}，期望 dict")

    raw_flag = data.get("is_highlight")
    # 清洗常见 LLM 错误输出
    if isinstance(raw_flag, str):
        raw_flag_lower = raw_flag.strip().lower()
        if raw_flag_lower in ("yes", "true", "1", "是", "正确"):
            raw_flag = True
        elif raw_flag_lower in ("no", "false", "0", "否", "错误"):
            raw_flag = False
        else:
            raw_flag = bool(raw_flag)  # 兜底：非空字符串为 True
    elif isinstance(raw_flag, (int, float)):
        raw_flag = bool(raw_flag)
    elif raw_flag is None:
        raw_flag = True  # 默认保留，避免误杀

    reason = str(data.get("reason", "无说明"))

    return _LLMResponse(is_highlight=raw_flag, reason=reason)


def _call_llm_single(
    prompt: str,
    api_base: Optional[str] = None,
    api_key: Optional[str] = None,
    model: Optional[str] = None,
    timeout: Optional[int] = None,
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None,
) -> Tuple[bool, str]:
    """
    单次调用 LLM API 进行高光校验。

    采用 OpenAI 原生结构化输出（response_format=json_object）+ Pydantic 强类型校验，
    彻底避免模型吐出多余自然语言或弱类型字符串导致解析失败。

    Args:
        prompt: 输入 Prompt。
        api_base, api_key, model, timeout, max_tokens, temperature: API 配置。

    Returns:
        Tuple[bool, str]: (is_highlight, reason)
        若调用失败，返回 (True, "LLM 调用失败，默认保留") 避免误杀。
    """
    api_base = api_base or settings.llm_api_base
    api_key = api_key or settings.llm_api_key or os.getenv("LLM_API_KEY", "")
    model = model or settings.llm_model_name
    timeout = timeout or settings.llm_timeout_sec
    max_tokens = max_tokens or settings.llm_max_tokens
    temperature = temperature if temperature is not None else settings.llm_temperature
    if not api_key:
        # 无 API Key，跳过校验，默认保留
        return True, "LLM_API_KEY 未设置，默认保留"

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": "你是一个体育赛事内容审核助手，只返回 JSON。"},
            {"role": "user", "content": prompt},
        ],
        "max_tokens": max_tokens,
        "temperature": temperature,
        # 强制模型返回合法 JSON，杜绝多余自然语言
        "response_format": {"type": "json_object"},
    }

    raw_content = ""
    try:
        response = requests.post(
            f"{api_base}/chat/completions",
            headers=headers,
            json=payload,
            timeout=timeout,
        )
        response.raise_for_status()
        data = response.json()
        raw_content = data["choices"][0]["message"]["content"].strip()

        parsed = _parse_llm_json(raw_content)
        return parsed.is_highlight, parsed.reason

    except requests.Timeout:
        logger.warning(f"LLM 请求超时 ({timeout}s)，默认保留")
        return True, f"LLM 请求超时 ({timeout}s)，默认保留"
    except requests.RequestException as exc:
        logger.warning(f"LLM 请求失败: {exc}，默认保留")
        return True, f"LLM 请求失败: {exc}，默认保留"
    except LLMValidationError as exc:
        snippet = raw_content[:300] if raw_content else "<空>"
        logger.warning(f"LLM 返回解析失败: {exc}，默认保留。原始返回: {snippet}")
        return True, f"LLM 返回解析失败，默认保留"
    except (json.JSONDecodeError, KeyError, IndexError, TypeError) as exc:
        # 兜底：捕获所有残余解析异常
        snippet = raw_content[:300] if raw_content else "<空>"
        logger.warning(f"LLM 返回解析失败: {exc}，默认保留。原始返回: {snippet}")
        return True, f"LLM 返回解析失败，默认保留"


def _llm_validate_candidates(
    candidates: List[HighlightCandidate],
    subtitles: List[SubtitleSegment],
    max_workers: int = min(4, settings.num_workers),
) -> Tuple[List[HighlightCandidate], int]:
    """
    使用 LLM 对候选列表进行批量语义校验。

    采用线程池并发请求，每个候选独立调用 LLM。
    若 API 未配置，直接返回原列表并记录警告。

    Args:
        candidates: 待校验的 HighlightCandidate 列表。
        subtitles: 全片字幕列表，用于提取关联文本。
        max_workers: 并发线程数。

    Returns:
        Tuple[List[HighlightCandidate], int]: (有效候选列表, 被过滤数量)
    """
    if not (settings.llm_api_key or os.getenv("LLM_API_KEY")):
        logger.warning(
            "LLM_API_KEY 未设置，跳过语义校验。"
            "所有候选默认保留。如需启用，请设置环境变量 LLM_API_KEY。"
        )
        for c in candidates:
            c.llm_validated = True
            c.llm_reason = "LLM 未启用，默认保留"
        return candidates, 0

    # 构建字幕索引，便于快速查找
    subtitle_by_index = {sub.index: sub for sub in subtitles}

    def _validate_one(candidate: HighlightCandidate) -> HighlightCandidate:
        # 根据候选框最终确定的时间轴区间 [start_sec, end_sec]，
        # 提取全片字幕中所有落在该区间内的连续字幕文本，
        # 形成完整的解说词上下文，避免仅投喂离散关键词句导致语义断层。
        window_texts: List[str] = []
        for sub in subtitles:
            if sub.end_sec >= candidate.start_sec and sub.start_sec <= candidate.end_sec:
                window_texts.append(sub.text)

        if not window_texts:
            # 无关联字幕（纯音频触发），默认保留
            candidate.llm_validated = True
            candidate.llm_reason = "纯音频触发，无解说文本，默认保留"
            return candidate

        # Do not make a structured sports-edit request depend on whether a
        # general LLM thinks an ordinary layup or missed shot is "exciting".
        # Event classification and the user's later instruction determine
        # whether to include it in the final reel.
        if _PROTECTED_BASKETBALL_PLAY.search(" ".join(window_texts)):
            candidate.llm_validated = True
            candidate.llm_reason = "规则保留：字幕含明确两分或未命中回合。"
            return candidate

        prompt = _build_llm_prompt(window_texts)
        is_highlight, reason = _call_llm_single(prompt)

        candidate.llm_validated = is_highlight
        candidate.llm_reason = reason
        return candidate

    logger.info(f"开始 LLM 语义校验: {len(candidates)} 个候选，并发数 {max_workers}")
    validated: List[HighlightCandidate] = []
    filtered_count = 0

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_candidate = {
            executor.submit(_validate_one, c): c for c in candidates
        }
        for future in as_completed(future_to_candidate):
            candidate = future_to_candidate[future]
            try:
                result = future.result()
                strong_event = any(
                    keyword.lower() in _STRONG_EVENT_KEYWORDS
                    for keyword in result.matched_keywords
                )
                if not result.llm_validated and strong_event:
                    result.llm_validated = True
                    result.llm_reason = (
                        "规则保留：字幕含明确得分事件词，"
                        "不因 LLM 不确定而误删。"
                    )
                if result.llm_validated:
                    validated.append(result)
                else:
                    filtered_count += 1
                    logger.info(
                        f"LLM 过滤候选 {result.candidate_id}: {result.llm_reason}"
                    )
            except Exception as exc:
                logger.error(f"LLM 校验异常: {exc}，默认保留")
                candidate.llm_validated = True
                candidate.llm_reason = f"校验异常: {exc}，默认保留"
                validated.append(candidate)

    logger.info(
        f"LLM 校验完成: 保留 {len(validated)} 个，过滤 {filtered_count} 个"
    )
    return validated, filtered_count


def _supplement_with_audio_coverage(
    candidates: List[HighlightCandidate],
    segments: List[AudioFeatureSegment],
    subtitles: List[SubtitleSegment],
    total_duration: float,
) -> List[HighlightCandidate]:
    """Add separated, high-energy windows when captions under-detect fast plays.

    Sports broadcasts often keep crowd volume almost constant, and Whisper can
    miss a rapid sequence of scores.  Rather than returning a one-clip montage,
    keep a small set of non-overlapping audio-peak windows as reviewable
    candidates.  Keyword/LLM candidates remain ranked above these fallbacks.
    """
    desired_count = min(4, max(2, int(total_duration // 12)))
    if len(candidates) >= desired_count:
        return candidates

    existing = [(c.start_sec, c.end_sec) for c in candidates]
    peak_max = max((seg.peak_loudness_dbfs for seg in segments), default=-20.0)
    ranked = sorted(segments, key=lambda seg: (seg.peak_loudness_dbfs, seg.mean_loudness_dbfs), reverse=True)

    for seg in ranked:
        center = (seg.start_sec + seg.end_sec) / 2
        start = round(max(0.0, center - 5.0), 3)
        end = round(min(total_duration, center + 8.0), 3)
        # A montage must never reuse source frames.  Even a one-second
        # overlap reads as an obvious replay when clips are concatenated.
        if any(max(0.0, min(end, old_end) - max(start, old_start)) > 0.0 for old_start, old_end in existing):
            continue
        related = [sub.index for sub in subtitles if sub.end_sec >= start and sub.start_sec <= end]
        relative_peak = max(0.0, min(1.0, (seg.peak_loudness_dbfs + 35.0) / 35.0))
        candidates.append(HighlightCandidate(
            candidate_id=f"HL_{len(candidates) + 1:03d}",
            start_sec=start,
            end_sec=end,
            duration_sec=round(end - start, 3),
            trigger=HighlightTrigger.AUDIO_SURGE,
            audio_score=round(max(0.25, relative_peak), 3),
            text_score=0.0,
            # Crowd volume is only a recall fallback.  It deliberately has a
            # small score so it cannot outrank a subtitle or verified-score
            # candidate later in the edit planner.
            combined_score=round(max(0.25, relative_peak) * settings.candidate_audio_weight, 3),
            matched_keywords=[],
            related_subtitle_indices=related,
            llm_validated=True,
            llm_reason="字幕召回不足时的音频峰值补充候选",
        ))
        existing.append((start, end))
        if len(candidates) >= desired_count:
            break
    return candidates


# ---------------------------------------------------------------------------
# 主流程：从 _RawCandidate 构建 HighlightCandidate
# ---------------------------------------------------------------------------
def _enforce_min_duration(
    raw: _RawCandidate,
    min_duration: float = settings.min_highlight_duration_sec,
    video_end_sec: Optional[float] = None,
) -> Optional[_RawCandidate]:
    """
    强制保证候选框时长不低于 min_duration。

    策略：
        - 若时长已 >= min_duration，直接返回原框。
        - 否则尝试向两侧对称扩展，每次扩展 0.5 秒，直到满足 min_duration
          或触及原始视频边界。
        - 若扩展后仍不足 min_duration，返回 None（丢弃该候选）。

    Args:
        raw: 原始候选框。
        min_duration: 最小允许时长（秒）。
        video_end_sec: 视频总时长（秒），作为右边界；None 则不限制右边界。

    Returns:
        Optional[_RawCandidate]: 扩展后的候选框，或 None（若无法达标则丢弃）。
    """
    duration = raw.end_sec - raw.start_sec
    if duration >= min_duration:
        return raw

    # 需要扩展的总量
    deficit = min_duration - duration
    expand_each_side = deficit / 2.0

    new_start = max(0.0, raw.start_sec - expand_each_side)
    new_end = raw.end_sec + expand_each_side
    if video_end_sec is not None:
        new_end = min(new_end, video_end_sec)

    # 如果单侧顶到边界，尝试把 deficit 全加到另一侧
    if new_start == 0.0 and new_end < raw.end_sec + deficit:
        new_end = min(raw.end_sec + deficit, video_end_sec if video_end_sec is not None else float("inf"))
    if video_end_sec is not None and new_end == video_end_sec and new_start > max(0.0, raw.start_sec - deficit):
        new_start = max(0.0, raw.start_sec - deficit)

    expanded_duration = new_end - new_start
    if expanded_duration < min_duration:
        logger.debug(
            f"  丢弃过短候选框 [{raw.start_sec:.1f}s-{raw.end_sec:.1f}s] "
            f"扩展后仍仅 {expanded_duration:.1f}s < {min_duration:.1f}s"
        )
        return None

    # 重新收集扩展区间内的音频和文本证据
    expanded_surges = [
        seg for seg in raw.surge_segments
        if seg.end_sec >= new_start and seg.start_sec <= new_end
    ]
    expanded_subs = [
        (sub, kws) for sub, kws in raw.matched_subtitles
        if sub.end_sec >= new_start and sub.start_sec <= new_end
    ]

    return _RawCandidate(
        start_sec=new_start,
        end_sec=new_end,
        source=raw.source,
        surge_segments=expanded_surges,
        matched_subtitles=expanded_subs,
    )


def _split_oversized_candidate(
    raw: _RawCandidate,
    max_duration: float = settings.max_highlight_duration_sec,
    expand_sec: float = 8.0,
) -> List[_RawCandidate]:
    """
    将过大的候选框按关键词命中点拆分为多个子候选框。

    策略：
        - 若候选框时长 <= max_duration，直接返回原框。
        - 否则，以每个关键词命中的字幕为中心，前后扩展 expand_sec 秒，
          形成独立的子高光框。
        - 子框之间允许少量重叠，由后续去重处理。
        - 合并后强制过滤时长 < min_highlight_duration_sec 的子框，
          尝试扩展达标，扩展失败则丢弃。

    Args:
        raw: 原始融合候选框。
        max_duration: 最大允许时长（秒）。
        expand_sec: 子框前后扩展时长（秒）。

    Returns:
        List[_RawCandidate]: 拆分并过滤后的候选框列表。
    """
    duration = raw.end_sec - raw.start_sec
    if duration <= max_duration:
        return [raw]

    # 按关键词命中字幕拆分
    sub_boxes: List[_RawCandidate] = []
    for sub, matched_kws in raw.matched_subtitles:
        center = (sub.start_sec + sub.end_sec) / 2
        box_start = max(raw.start_sec, center - expand_sec)
        box_end = min(raw.end_sec, center + expand_sec)

        # 收集该子框内的音频证据
        sub_surges = [
            seg for seg in raw.surge_segments
            if seg.end_sec >= box_start and seg.start_sec <= box_end
        ]

        sub_boxes.append(_RawCandidate(
            start_sec=box_start,
            end_sec=box_end,
            source=raw.source,
            surge_segments=sub_surges,
            matched_subtitles=[(sub, matched_kws)],
        ))

    # 合并相邻的子框（间隔 <= 2秒），但严格拦截合并后总时长超过 max_duration
    if not sub_boxes:
        return [raw]

    sub_boxes.sort(key=lambda b: b.start_sec)
    merged: List[_RawCandidate] = []
    current = sub_boxes[0]

    for box in sub_boxes[1:]:
        # 条件1：相邻间隔 <= 2秒
        gap_ok = box.start_sec - current.end_sec <= 2.0
        # 条件2：合并后总时长不超过硬上限
        merged_end = max(current.end_sec, box.end_sec)
        duration_ok = (merged_end - current.start_sec) <= max_duration

        if gap_ok and duration_ok:
            # 安全合并：两个条件同时满足
            current.end_sec = merged_end
            current.surge_segments.extend(box.surge_segments)
            current.matched_subtitles.extend(box.matched_subtitles)
        else:
            # 任一条件不满足，果断切断，当前框入栈，开启新框
            merged.append(current)
            current = box
    merged.append(current)

    # === 关键修复：强制保证 min_duration 下限 ===
    # 对合并后的每个子框执行时长下限检查，不足则扩展或丢弃
    final_boxes: List[_RawCandidate] = []
    for box in merged:
        enforced = _enforce_min_duration(box)
        if enforced is not None:
            final_boxes.append(enforced)

    logger.info(
        f" oversized框 [{raw.start_sec:.1f}s-{raw.end_sec:.1f}s] "
        f"拆分为 {len(merged)} 个子框，过滤后保留 {len(final_boxes)} 个"
    )
    return final_boxes


def _find_goal_trigger_timestamp(
    subtitles: List[SubtitleSegment],
    candidate_start: float,
    candidate_end: float,
) -> Optional[float]:
    """
    在候选框时间范围内，查找第一个进球触发词出现的时间戳。
    """
    for sub in subtitles:
        if sub.end_sec < candidate_start or sub.start_sec > candidate_end:
            continue
        text_lower = sub.text.lower()
        for pattern in _GOAL_TRIGGER_PATTERNS:
            if re.search(pattern, text_lower, re.IGNORECASE):
                return sub.start_sec
    return None


def _compute_dynamic_keep_sec(
    audio_score: float,
    audio_peak_dbfs: Optional[float] = None,
) -> Tuple[float, float]:
    """
    根据音频特征计算动态保留时长。

    算法：
        1. 基础时长：_BASE_PRE_GOAL_SEC / _BASE_POST_GOAL_SEC
        2. 响度强度归一化：将 peak_dbfs 映射到 [0, 1]
           intensity = (peak_dbfs - _LOUDNESS_DYNAMIC_MIN_DBFS) / (_LOUDNESS_DYNAMIC_MAX_DBFS - _LOUDNESS_DYNAMIC_MIN_DBFS)
           intensity = clamp(intensity, 0, 1)
        3. 动态调整：
           pre = base_pre + intensity * _DYNAMIC_RANGE_PRE
           post = base_post + intensity * _DYNAMIC_RANGE_POST

    Args:
        audio_score: 音频得分（作为响度强度的替代指标，如果 peak_dbfs 不可用）。
        audio_peak_dbfs: 候选框内峰值响度（dBFS），可选。

    Returns:
        Tuple[float, float]: (pre_goal_sec, post_goal_sec)
    """
    if audio_peak_dbfs is not None:
        # 使用实际响度计算强度
        intensity = (audio_peak_dbfs - _LOUDNESS_DYNAMIC_MIN_DBFS) / (
            _LOUDNESS_DYNAMIC_MAX_DBFS - _LOUDNESS_DYNAMIC_MIN_DBFS
        )
        intensity = max(0.0, min(1.0, intensity))
    else:
        # 回退：使用 audio_score 作为响度强度的代理
        intensity = audio_score

    pre = _BASE_PRE_GOAL_SEC + intensity * _DYNAMIC_RANGE_PRE
    post = _BASE_POST_GOAL_SEC + intensity * _DYNAMIC_RANGE_POST

    return round(pre, 2), round(post, 2)


def _semantic_trim_candidate(
    candidate: HighlightCandidate,
    subtitles: List[SubtitleSegment],
    audio_segments: Optional[List[AudioFeatureSegment]] = None,
) -> HighlightCandidate:
    """
    对高光候选片段进行语义化精剪（支持响度动态调整）。

    策略：
        1. 在候选框内查找进球触发词（如 "BANG!" / "it's going!" / "downtown!"）。
        2. 以触发词为中心，根据响度动态计算前后保留时长。
        3. 确保精剪后的时长在 [_TRIM_MIN_DURATION, _TRIM_MAX_DURATION] 范围内。
        4. 如果找不到触发词，返回原始候选框。
    """
    trigger_ts = _find_goal_trigger_timestamp(
        subtitles, candidate.start_sec, candidate.end_sec
    )

    if trigger_ts is None:
        return candidate

    # 计算动态保留时长
    peak_dbfs = None
    if audio_segments:
        # 找到候选框时间范围内的音频段，取峰值
        relevant = [
            seg for seg in audio_segments
            if seg.end_sec >= candidate.start_sec and seg.start_sec <= candidate.end_sec
        ]
        if relevant:
            peak_dbfs = max(seg.peak_loudness_dbfs for seg in relevant)

    pre_sec, post_sec = _compute_dynamic_keep_sec(
        candidate.audio_score,
        audio_peak_dbfs=peak_dbfs,
    )

    new_start = max(candidate.start_sec, trigger_ts - pre_sec)
    new_end = min(candidate.end_sec, trigger_ts + post_sec)

    # 确保最小时长
    if new_end - new_start < _TRIM_MIN_DURATION:
        new_end = min(candidate.end_sec, new_start + _TRIM_MIN_DURATION)

    # 确保最大时长
    if new_end - new_start > _TRIM_MAX_DURATION:
        new_end = new_start + _TRIM_MAX_DURATION

    new_start = max(candidate.start_sec, new_start)
    new_end = min(candidate.end_sec, new_end)

    if new_start >= new_end:
        return candidate

    new_duration = round(new_end - new_start, 3)

    logger.info(
        f"语义化精剪: {candidate.candidate_id} "
        f"[{candidate.start_sec:.1f}s-{candidate.end_sec:.1f}s] "
        f"触发词@{trigger_ts:.1f}s 响度={peak_dbfs:.1f}dBFS "
        f"动态保留(pre={pre_sec:.1f}s post={post_sec:.1f}s) → "
        f"[{new_start:.1f}s-{new_end:.1f}s] duration={new_duration:.1f}s"
    )

    return HighlightCandidate(
        candidate_id=candidate.candidate_id,
        start_sec=round(new_start, 3),
        end_sec=round(new_end, 3),
        duration_sec=new_duration,
        trigger=candidate.trigger,
        audio_score=candidate.audio_score,
        text_score=candidate.text_score,
        combined_score=candidate.combined_score,
        matched_keywords=candidate.matched_keywords,
        related_subtitle_indices=candidate.related_subtitle_indices,
        llm_validated=candidate.llm_validated,
        llm_reason=candidate.llm_reason,
    )


def _build_highlight_candidates(
    fused: List[_RawCandidate],
    global_mean_loudness: float,
    subtitles: List[SubtitleSegment],
    audio_segments: Optional[List[AudioFeatureSegment]] = None,
) -> List[HighlightCandidate]:
    """
    将融合后的原始候选框转换为标准化的 HighlightCandidate 列表。

    步骤：
        1. 对过大的融合框进行拆分。
        2. 为每个框计算音频/文本/综合得分。
        3. 语义化精剪：以进球触发词为中心裁剪（支持响度动态调整）。
        4. 收集关联证据（关键词、字幕序号）。

    Args:
        fused: 融合后的 _RawCandidate 列表。
        global_mean_loudness: 全片平均响度。
        subtitles: 全片字幕列表，用于计算 related_subtitle_indices。
        audio_segments: 音频特征段列表，用于响度动态调整。

    Returns:
        List[HighlightCandidate]: 结构化候选列表。
    """
    # 先拆分过大的框
    split_boxes: List[_RawCandidate] = []
    for raw in fused:
        split_boxes.extend(_split_oversized_candidate(raw))

    candidates: List[HighlightCandidate] = []

    for idx, raw in enumerate(split_boxes, start=1):
        audio_score = _compute_audio_score(raw, global_mean_loudness)
        text_score = _compute_text_score(raw)
        # Audio finds exciting moments, but commentary/text is more reliable
        # for identifying the actual sporting action.  Scoreboard OCR adds a
        # still stronger evidence boost later in the application pipeline.
        combined_score = round(
            settings.candidate_audio_weight * audio_score
            + settings.candidate_text_weight * text_score,
            3,
        )
        trigger = _determine_trigger(raw)

        # 收集匹配到的关键词（去重）
        matched_keywords: List[str] = []
        seen_kw: Set[str] = set()
        for _, kws in raw.matched_subtitles:
            for kw in kws:
                if kw not in seen_kw:
                    seen_kw.add(kw)
                    matched_keywords.append(kw)

        # 计算关联字幕序号（与候选框时间重叠的字幕）
        related_indices: List[int] = []
        for sub in subtitles:
            if sub.end_sec >= raw.start_sec and sub.start_sec <= raw.end_sec:
                related_indices.append(sub.index)

        candidate = HighlightCandidate(
            candidate_id=f"HL_{idx:03d}",
            start_sec=round(raw.start_sec, 3),
            end_sec=round(raw.end_sec, 3),
            duration_sec=round(raw.end_sec - raw.start_sec, 3),
            trigger=trigger,
            audio_score=audio_score,
            text_score=text_score,
            combined_score=combined_score,
            matched_keywords=matched_keywords,
            related_subtitle_indices=related_indices,
            llm_validated=False,
            llm_reason=None,
        )
        candidates.append(candidate)

    # 语义化精剪：以进球触发词为中心裁剪（支持响度动态调整）
    trimmed_candidates = [
        _semantic_trim_candidate(c, subtitles, audio_segments=audio_segments)
        for c in candidates
    ]

    # 按 combined_score 降序排列
    trimmed_candidates.sort(key=lambda c: c.combined_score, reverse=True)
    return trimmed_candidates


# ---------------------------------------------------------------------------
# Agent 主入口
# ---------------------------------------------------------------------------
class HighlightFilterAgent:
    """
    HighlightFilter Agent 封装类。

    对外提供统一的 `process(media_parse_json_path: str) -> str` 接口，
    返回产出的 JSON 结果文件路径。
    """

    def process(self, media_parse_json_path: str) -> str:
        """
        执行完整的 HighlightFilter 流水线。

        Args:
            media_parse_json_path: MediaParseOutput JSON 文件路径。

        Returns:
            str: HighlightFilterOutput JSON 文件的绝对路径。

        Raises:
            HighlightFilterError: 各阶段异常。
        """
        media_parse_json_path = str(Path(media_parse_json_path).resolve())
        if not Path(media_parse_json_path).exists():
            raise HighlightFilterError(f"输入文件不存在: {media_parse_json_path}")

        logger.info(
            f"========== HighlightFilter Agent 开始处理: {media_parse_json_path} =========="
        )

        # 1. 读取 MediaParseOutput
        from schema import load_media_parse_output
        media_data = load_media_parse_output(media_parse_json_path)
        logger.info(
            f"读取 MediaParseOutput: {len(media_data.segments)} 段音频, "
            f"{len(media_data.subtitles)} 条字幕"
        )

        # 2. 提取音频候选框
        audio_candidates = _extract_audio_candidates(
            media_data.segments, total_duration_sec=media_data.total_duration_sec
        )

        # 3. 提取文本候选框
        text_candidates = _extract_text_candidates(
            media_data.subtitles,
            settings.sports_keywords,
        )

        # 4. 窗口融合
        fused = _fuse_candidates(audio_candidates, text_candidates)
        if not fused:
            logger.warning("未检测到任何高光候选，输出空结果")
            output = HighlightFilterOutput(
                source_video=media_data.source_video,
                candidates=[],
                filter_config_snapshot={},
                raw_candidate_count=0,
                filtered_count=0,
            )
        else:
            # 5. 构建结构化候选（含评分）
            candidates = _build_highlight_candidates(
                fused,
                media_data.global_mean_loudness_dbfs,
                media_data.subtitles,
                audio_segments=media_data.segments,
            )
            raw_count = len(candidates)

            # 6. LLM 语义校验
            validated_candidates, filtered_count = _llm_validate_candidates(
                candidates,
                media_data.subtitles,
            )

            # Ensure a short, high-event game does not collapse to a single
            # ten-second clip solely because speech recognition missed calls.
            validated_candidates = _supplement_with_audio_coverage(
                validated_candidates,
                media_data.segments,
                media_data.subtitles,
                media_data.total_duration_sec,
            )

            # 7. 再次按 combined_score 排序（LLM 不改变分数，仅过滤）
            validated_candidates.sort(key=lambda c: c.combined_score, reverse=True)

            # 8. 配置快照
            config_snapshot = {
                "loudness_surge_multiplier": settings.loudness_surge_multiplier,
                "min_highlight_duration_sec": settings.min_highlight_duration_sec,
                "max_highlight_duration_sec": settings.max_highlight_duration_sec,
                "sports_keywords": settings.sports_keywords,
                "keyword_fuzzy_threshold": settings.keyword_fuzzy_threshold,
                "llm_model_name": settings.llm_model_name,
                "llm_api_base": settings.llm_api_base,
            }

            output = HighlightFilterOutput(
                source_video=media_data.source_video,
                candidates=validated_candidates,
                filter_config_snapshot=config_snapshot,
                raw_candidate_count=raw_count,
                filtered_count=filtered_count,
            )

        # 9. 序列化输出
        video_stem = Path(media_data.source_video).stem
        output_dir = settings.get_temp_subdir("highlight_results")
        output_path = str(output_dir / f"{video_stem}_highlight_candidates.json")
        save_to_json(output, output_path)

        logger.info(
            f"HighlightFilter Agent 处理完成，结果已保存: {output_path}"
        )
        logger.info(
            f"统计: 原始候选 {output.raw_candidate_count} 个, "
            f"LLM 过滤 {output.filtered_count} 个, "
            f"最终保留 {len(output.candidates)} 个"
        )

        return output_path


# ---------------------------------------------------------------------------
# 命令行直接运行入口
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="HighlightFilter Agent —— 高光片段筛选")
    parser.add_argument("input_json", help="MediaParseOutput JSON 文件路径")
    parser.add_argument(
        "--output-json",
        default=None,
        help="指定输出 JSON 路径（默认按 settings.temp_dir 规则生成）",
    )
    args = parser.parse_args()

    agent = HighlightFilterAgent()
    try:
        result_path = agent.process(args.input_json)
        print(f"\n✅ HighlightFilter Agent 执行成功")
        print(f"📄 结果文件: {result_path}")
    except HighlightFilterError as exc:
        logger.error(f"HighlightFilter Agent 执行失败: {exc}")
        raise SystemExit(1) from exc