jira:NYJ-1550 desc:football replay match

+input_kafka:
+  servers: 192.168.0.14:9092
+  group_id: ai_match_service
+  topic: football_replay_match
+  # username: xxx
+  # password: xxx
+
+output_kafka:
+  servers: 192.168.0.14:9092
+  topic: football_replay_match

+app_name: "Football Replay Service"
+app_version: "1.0.0"

+from aabd.base.cfg_loader import load_yaml_by_file_with_env
+from aabd.base.enhance_dict import EnhanceDict, read_prefixed_env_vars
+from pathlib import Path
+from logging import getLogger
+logger = getLogger(__name__)
+cfg_dir = Path(__file__).parent.absolute() / 'cfg'
+settings = EnhanceDict(load_yaml_by_file_with_env((cfg_dir / 'config.yaml').as_posix()))
+settings.update(read_prefixed_env_vars('FRM_'))
+
+logger.info(f'Settings: {settings}')

+def replay_match_event(data):
+    pass

+import json
+import time
+
+from aabd.base.patched_logging import init_logging, get_logger
+
+init_logging()
+logger = get_logger(__name__)
+from threading import Thread
+from .core.api import replay_match_event
+
+from aabd.mq.kafka_client import KafkaMessageIterator, KafkaProducer
+
+
+def get_message_iterator(kafka_config):
+    return KafkaMessageIterator(bootstrap_servers=kafka_config.get('servers', None),
+                                group_id=kafka_config.get('group_id', None),
+                                topic=kafka_config.get('topic', None),
+                                sasl_plain_username=kafka_config.get('username', None),
+                                sasl_plain_password=kafka_config.get('password', None),
+                                value_deserializer="str")
+
+
+def get_message_producer(kafka_config):
+    return KafkaProducer(bootstrap_servers=kafka_config.get('servers', None),
+                         sasl_plain_username=kafka_config.get('username', None),
+                         sasl_plain_password=kafka_config.get('password', None))
+
+
+def get_callback_func(producer, topic):
+    def callback_func(task_id, call_data):
+        try:
+            st = time.time()
+            producer.send_message_async(topic, call_data, key=task_id, timeout=5)
+            logger.info(f"sent_message[{task_id}][{time.time() - st:.2f}s]: {call_data}")
+        except:
+            logger.exception(f"Error sending message: {call_data}")
+
+    return callback_func
+
+
+def kafka_message_iterator_thread(message_iterator, config, callback_func):
+    with message_iterator:
+        for message in message_iterator:
+            try:
+                json_message = json.loads(message)
+                replay_match_event(json_message)
+                logger.info(message)
+            except:
+                logger.exception(f"Error processing message: {message}")
+
+
+if __name__ == '__main__':
+    from config import settings
+
+    message_iter = get_message_iterator(settings.input_kafka)
+    kafka_producer = get_message_producer(settings.output_kafka)
+    try:
+
+        Thread(target=kafka_message_iterator_thread,
+               args=(message_iter, settings,
+                     get_callback_func(kafka_producer, settings.output_kafka.get('topic', None)))).start()
+
+    except KeyboardInterrupt:
+        message_iter.running = False
+        kafka_producer.close()

+import json
+import os.path
+from pathlib import Path
+
+from langchain_core.messages import SystemMessage, HumanMessage
+from langchain_ollama import ChatOllama
+
+try:
+    from .llm_video_content import contents as video_contents
+except:
+    from llm_video_content import contents as video_contents
+
+req_prompt = """
+# Role
+你是一名拥有20年经验的足球视频技术分析师，擅长结合**视觉画面**与**解说音频（ASR）**进行跨镜头的事件匹配。你能通过解说员的描述锁定时间背景，并通过视觉特征确认物理细节。
+
+# Task
+输入包含：
+1. 【回放片段】（Replay）：包含视频帧 + **对应的解说文本**。
+2. 【直播进球片段列表】（Live Candidates）：包含视频帧 + **对应的解说文本**。
+
+目标：在【直播片段】中找到与【回放片段】属于**同一个进球事件**的片段。如果没有任何片段匹配，返回 null。
+
+# Analysis Workflow (多模态分析流程)
+
+请按照以下步骤进行推理：
+
+### 第一步：解说词元数据提取 (听觉线索)
+分析【回放片段】的解说文本，寻找以下关键信息：
+- **时间指代**：解说员是否提到了具体时间？（如“上半场”、“第10分钟”、“开场不久”）。
+- **事件描述**：解说员如何描述这个进球？（如“世界波”、“点球”、“补射”、“乌龙球”）。
+- **球员提及**：解说员念出了谁的名字？
+
+### 第二步：视觉物理指纹提取 (视觉线索)
+忽略镜头语言（慢放、特写），提取核心物理特征：
+- **进攻与射门**：进攻方式（传中/直塞）、射门部位（头/脚）、射门位置。
+- **球路与防守**：球的轨迹（高/低/折射）、门将扑救动作（侧扑/倒地/未动）。
+- **庆祝**：进球后的庆祝动作（仅作为辅助验证）。
+
+### 第三步：跨模态匹配与验证
+遍历【直播片段】，结合视觉和听觉进行判断：
+- **视觉一致性**：直播片段的动作、轨迹、门将反应是否与回放完全吻合？
+- **听觉一致性**：
+    - 如果回放解说提到“这是第5分钟的进球”，而直播片段的时间戳是90分钟，**不要直接排除**，而是检查直播片段的解说是否也提到了“回顾第5分钟”或者直播片段的视觉内容确实是第5分钟的动作。
+    - **核心原则**：视觉物理特征 > 解说词描述 > 时间戳元数据。
+    - **特殊情况**：如果视觉特征高度相似（如同一个球员、同一个角度），但解说词明确说“这是另一个进球（例如：这是他的第二个进球，而回放说是第一个）”，则判定为不匹配。
+
+# Logic Constraints (逻辑约束 - 必须严格遵守)
+
+- **时间单向性原则（铁律）**：
+    - **回看中的比赛时间一定在比赛片段画面时间之后**。
+    - 逻辑：回放是对过去发生事件的回顾。如果【回放片段】中解说提到的比赛时间（或画面显示的比赛时钟）**早于**【直播片段】中解说提到的比赛时间（或画面时钟），则**绝对不可能**是同一个事件。
+    - *示例*：回放解说在描述“第10分钟的进球”，而直播片段明确发生在“第5分钟”，则该直播片段**一定不是**目标。
+- **时间陷阱**：回放可能是赛后集锦。如果解说员说“让我们看看**刚才**那个球”或者“**上半场**那个球”，即使当前比赛时间是90分钟，也要去匹配对应时间段的直播片段（或视觉特征）。
+- **同名陷阱**：如果解说提到“又是**凯恩**进球了”，不能只看凯恩，必须看**怎么进的**（头球还是点球）。
+- **无匹配处理**：如果所有候选片段在视觉动作（如射门方式、进球位置）或关键事件逻辑上与回放明显不符，必须判定为无匹配，将 `video_id` 设为 `null`。
+
+### 输出要求
+
+请仅输出一个JSON格式的结果，不要输出任何分析过程。不要包含 markdown 标记（如 ```json ... ```），不要包含任何解释或额外文本。
+格式如下：
+{
+  "replay_summary": {
+    "audio_cues": "解说提到的关键信息（如：'第15分钟', '远射', '德布劳内'）",
+    "visual_cues": "视觉关键特征（如：'禁区外右脚', '球挂死角', '门将飞身扑救'）"
+  },
+  "reasoning": "综合分析：回放解说提到是'上半场的远射'，视觉显示'17号球员禁区外起脚'。Candidate_1 视觉是'近距离推射'，排除。Candidate_2 视觉是'禁区外远射'，且门将动作一致，虽然直播时间显示是下半场（可能是集锦回顾），但解说也提到了'回顾上半场'，确认为同一事件。",
+  "video_id": "Candidate_2"
+}"""
+
+
+class FootballReplayMatchLive:
+    def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key'):
+        self.base_url = base_url
+        self.model = model
+        self.temperature = temperature
+        self.api_key = api_key
+        # self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
+        #                    keep_alive=-1, reasoning=False)
+        # self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
+        #                        api_key='no_key')
+        self.model = ChatOllama(base_url=base_url, model=model, temperature=temperature, keep_alive=-1, reasoning=False)
+
+    def _match_once(self, replay_video_contents: list, live_videos: list[dict], cache_path=None, record: list = None):
+
+        if len(live_videos) == 0:
+            return None
+        elif len(live_videos) == 1:
+            live_video_path = live_videos[0].get("video_path", None)
+            live_video_id = live_videos[0].get("video_id", os.path.basename(live_video_path))
+            asr_text = live_videos[0].get("asr_text", '')
+            live = {
+                "video_id": live_video_id,
+                "video_path": live_video_path,
+                "asr_text": asr_text,
+            }
+            if record is not None:
+                record.append({"live": live, "llm_result": None, 'live_list': [live]})
+            return live
+        user_contents = []
+        # replay_video_contents = video_contents(replay_video["video_path"], "\n【回放片段信息】\n",
+        #                                        prompt_end=f"\n回放解说内容：{replay_video['asr_text']}\n",
+        #                                        video_name=os.path.basename(replay_video["video_path"]),
+        #                                        fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
+        live_videos_contents = []
+        live_map = {}
+        live_records = {}
+        for live_video in live_videos:
+            live_video_path = live_video.get("video_path", None)
+            live_video_id = live_video.get("video_id", os.path.basename(live_video_path))
+            asr_text = live_video.get("asr_text", '')
+            live_video_contents = live_video.get("llm_contents", None)
+            if live_video_contents is None:
+                live_video_contents = video_contents(live_video_path,
+                                                     prompt_start=f"### 候选片段 video_id: {live_video_id} ###",
+                                                     prompt_end=f"\n该片段解说内容: {asr_text}\n",
+                                                     video_name=os.path.basename(live_video_path),
+                                                     fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
+            live_map[live_video_id] = {
+                "video_id": live_video_id,
+                "video_path": live_video_path,
+                "asr_text": asr_text,
+                "contents": live_video_contents,
+
+            }
+            live_records[live_video_id] = {
+                "video_id": live_video_id,
+                "video_path": live_video_path,
+                "asr_text": asr_text,
+            }
+            live_videos_contents.extend(live_video_contents)
+
+        user_contents.extend(replay_video_contents)
+        user_contents.append({'type': "text", "text": "\n【候选直播片段列表】\n"})
+        user_contents.extend(live_videos_contents)
+        user_contents.append({'type': "text", "text": "\n请根据上述片段进行匹配并按要求输出结果\n"})
+        system_message = SystemMessage(content=req_prompt)
+        user_message = HumanMessage(content=user_contents)
+        result = self.model.invoke([system_message, user_message]).content
+        try:
+            result_json = json.loads(result)
+        except json.JSONDecodeError:
+            try:
+                result_json = json.loads(result.replace("```json", "").replace("```", ""))
+            except Exception as e:
+                print("JSON解析失败:", result)
+                raise e
+
+        video_id = result_json.get("video_id", None)
+        result_live = live_map.get(video_id, None)
+        if record is not None:
+            record.append(
+                {"live": live_records.get(video_id,None), "llm_result": result_json, 'live_list': list(live_records.values())})
+        return result_live
+
+    def _match_batch(self, replay_video_contents: list, live_videos: list[dict], max_parallel: int = 3, cache_path=None,
+                     record: list = None):
+        """
+        Match a replay video with live videos to find the most likely match.
+        :param max_parallel:
+        :param replay_video: Path to the replay video.(video_path,asr_text)
+        :param live_videos: [(video_id,video_path,asr_text)]
+        :param cache_path: Path to cache the result.
+        :return: JSON object containing the match result.
+        """
+        if cache_path is not None and os.path.exists(cache_path):
+            with open(cache_path, 'r', encoding='utf-8') as f:
+                return json.loads(f.read())
+
+        # 按照max_parallel对live_videos进行分组
+        live_videos_groups = [live_videos[i::max_parallel] for i in range(max_parallel)]
+        # 过滤掉空的分组
+        live_videos_groups = [g for g in live_videos_groups if g]
+        # 如果group 大于1 并且最后一个group 只有一个元素，将其唯一元素放入倒数第二个group，并删除最后一个group
+        if len(live_videos_groups) > 1 and len(live_videos_groups[-1]) == 1:
+            live_videos_groups[-2].append(live_videos_groups[-1][0])
+            live_videos_groups.pop()
+
+        if len(live_videos_groups) > 1:
+            match_result = []
+            for live_videos_group in live_videos_groups:
+                g_live = self._match_once(replay_video_contents, live_videos_group, cache_path, record)
+                if g_live is None:
+                    continue
+                match_result.append(g_live)
+            if len(match_result) == 0:
+                return None
+            else:
+                return self._match_batch(replay_video_contents, match_result, max_parallel, cache_path, record)
+        elif len(live_videos_groups) == 1:
+            return self._match_once(replay_video_contents, live_videos_groups[0], cache_path, record)
+        else:
+            return None
+
+    def match_batch(self, replay_video: dict, live_videos: list[dict], max_parallel: int = 3, cache_path=None):
+        if cache_path is not None and os.path.exists(cache_path):
+            try:
+                with open(cache_path, 'r', encoding='utf-8') as f:
+                    return json.loads(f.read()).get("result", None)
+            except:
+                os.remove(cache_path)
+        replay_video_contents = video_contents(replay_video["video_path"], "\n【回放片段信息】\n",
+                                               prompt_end=f"\n回放解说内容：{replay_video['asr_text']}\n",
+                                               video_name=os.path.basename(replay_video["video_path"]),
+                                               fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
+        live_record = []
+        result = self._match_batch(replay_video_contents, live_videos, max_parallel, cache_path, live_record)
+        if result is not None:
+            result_no_content = {
+                "video_id": result.get("video_id", None),
+                "video_path": result.get("video_path", None),
+                "asr_text": result.get("asr_text", None),
+            }
+        else:
+            result_no_content = None
+        record = {
+            "request": {
+                "replay_video": replay_video,
+                "live_videos": live_videos,
+                "max_parallel": max_parallel,
+                "cache_path": cache_path
+            },
+            "result": result_no_content,
+            "live_record": live_record
+        }
+        if cache_path is not None:
+            os.makedirs(Path(cache_path).parent, exist_ok=True)
+            with open(cache_path, 'w', encoding='utf-8') as f:
+                f.write(json.dumps(record, ensure_ascii=False, indent=4))
+        return result_no_content

+import json
+import os.path
+from pathlib import Path
+
+from langchain_core.messages import SystemMessage, HumanMessage
+from langchain_ollama import ChatOllama
+
+try:
+    from .llm_video_content import contents as video_contents
+except:
+    from llm_video_content import contents as video_contents
+
+req_prompt = """
+### 角色设定
+你是一位拥有20年经验的足球赛事视频分析专家。你的任务是结合**视频画面**和**解说音频文本**，精准判断视频片段中是否发生了"有效进球"。
+
+### 输入内容
+
+1. **视频片段**：可能包含赛场、观众、教练、回放、演播室、宣传片、广告等多种画面。
+2. **解说文本**：该片段对应的实时解说内容（ASR）。
+
+### 分析逻辑（思维链）
+
+请综合以下三个维度的信息进行推理：
+
+#### 1. 场景维度（非比赛画面过滤 - 最高优先级）
+
+- **非比赛内容识别**：检查画面是否为**宣传片**、**商业广告**、**纯演播室解说**、**集锦混剪**（无连续比赛画面）或*
+ *静态图文**。
+- **处理规则**：如果视频内容主要是上述非比赛画面，且没有包含明确的实时进球片段，**直接判定为“无进球”**
+ ，无需进行后续的进球逻辑分析。
+
+#### 2. 视觉维度（寻找关键证据）
+
+- **核心画面**：足球入网、球在网内静止、裁判指中圈。
+- **行为线索**：进攻方疯狂庆祝、防守方抱头懊恼、全场观众起立欢呼。
+- **画面容忍度**：即使画面主要是观众或教练特写，只要上下文（如切入的进球回放）或行为暗示了进球，也应视为进球。
+
+#### 3. 听觉维度（解说语义分析）
+
+- **进球关键词**：寻找如“球进啦”、“Goal”、“得分”、“世界波”、“破门”、“1比0”等肯定性词汇。
+- **情绪语调**：解说员音量突然升高、语速加快、情绪激动（通常伴随进球发生）。
+- **否定排除**：如果解说提到“越位在先”、“进球无效”、“击中横梁”、“偏出”，则视为未进球。
+
+### 判定规则
+
+- **判定为“无进球”**：
+   - **画面为宣传片、广告、纯演播室或其他非比赛实时内容；**
+   - 视觉显示未进（偏出/被扑/中柱）；
+   - 解说明确表示未进或进球无效；
+   - 画面与解说均无进球迹象（如普通传球、界外球）。
+- **判定为“进球”**：
+   - 画面为比赛内容，且视觉清晰显示进球；
+   - **或** 画面为比赛内容，视觉模糊但解说员明确喊出“球进了”且情绪激动；
+   - **或** 画面显示庆祝/回放，配合解说确认进球。
+
+### 输出要求
+
+请仅输出一个JSON格式的结果，不要输出任何分析过程。不要包含 markdown 标记（如 ```json ... ```），不要包含任何解释或额外文本。
+格式如下：
+{
+"event_name": "进球" 或 "无进球",
+"description": "判定理由，若是宣传片请直接注明"
+}"""
+
+
+class FootballReplayVideoEvent:
+    def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key'):
+        self.base_url = base_url
+        self.model = model
+        self.temperature = temperature
+        self.api_key = api_key
+        # self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
+        #                    keep_alive=-1, reasoning=False)
+        # self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
+        #                        api_key='no_key')
+        self.model = ChatOllama(base_url=base_url, model=model, temperature=temperature, keep_alive=-1, reasoning=False)
+
+    def video_event(self, video_path: str, asr_text: str = '无', cache_path=None):
+        if cache_path is not None and os.path.exists(cache_path):
+            with open(cache_path, 'r', encoding='utf-8') as f:
+                return json.loads(f.read())
+
+        contents = video_contents(video_path, None, video_name=os.path.basename(video_path),
+                                  fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
+        system_message = SystemMessage(content=req_prompt)
+        video_message = HumanMessage(content=contents)
+        asr_message = HumanMessage(content=f"解说内容：{asr_text}")
+        result = self.model.invoke([system_message, video_message, asr_message]).content
+        try:
+            result_json = json.loads(result)
+        except json.JSONDecodeError:
+            result_json = json.loads(result.replace("```json", "").replace("```", ""))
+        if cache_path is not None:
+            os.makedirs(Path(cache_path).parent, exist_ok=True)
+            with open(cache_path, 'w', encoding='utf-8') as f:
+                f.write(result)
+        return result_json

+import base64
+import tempfile
+from pathlib import Path
+from typing import Union
+import cv2
+import httpx
+
+
+def _resize_frame(frame, max_short_edge: int):
+    """按比例缩放帧，确保最短边不超过指定值"""
+    h, w = frame.shape[:2]
+    short_edge = min(h, w)
+    if short_edge > max_short_edge:
+        scale = max_short_edge / short_edge
+        new_w = int(w * scale)
+        new_h = int(h * scale)
+        frame = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_AREA)
+    return frame
+
+
+def contents(video_source: Union[str, Path], prompt_start: str = None, prompt_end=None, video_name: str = None,
+             fps: float = 1.0,
+             max_frames: int = 10,
+             max_short_edge: int = 768,
+             sampling_mode: str = "uniform") -> list[str]:
+    """
+    从视频中提取帧并构建 LLM 消息内容。
+
+    Args:
+        video_source: 视频源，可以是文件路径或 URL。
+        prompt_start: 提示词的开始部分。
+        prompt_end: 提示词的结束部分。
+        video_name: 视频名称。
+        fps: 提取帧的帧率。
+        max_frames: 最大帧数。
+        max_short_edge: 最大短边长度。
+        sampling_mode: 当帧数超过 max_frames 时的采样策略。
+            - "uniform": 均匀采样（默认）
+            - "head": 保留前面的帧，抛弃后面的帧
+    """
+    source_str = str(video_source)
+    temp_file = None
+
+    try:
+        # 如果是 URL，先下载到临时文件
+        if source_str.startswith(("http://", "https://")):
+            resp = httpx.get(source_str, timeout=60.0)
+            resp.raise_for_status()
+            temp_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
+            temp_file.write(resp.content)
+            temp_file.close()
+            video_path = temp_file.name
+        else:
+            video_path = str(video_source)
+            if not Path(video_path).exists():
+                raise FileNotFoundError(f"视频文件不存在: {video_path}")
+
+        # 打开视频
+        cap = cv2.VideoCapture(video_path)
+        if not cap.isOpened():
+            raise ValueError(f"无法打开视频: {video_source}")
+
+        # 获取视频原始帧率
+        video_fps = cap.get(cv2.CAP_PROP_FPS)
+
+        # 计算帧间隔
+        frame_interval = int(video_fps / fps) if fps > 0 else int(video_fps)
+        if frame_interval < 1:
+            frame_interval = 1
+
+        frames_base64 = []
+        frame_count = 0
+
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+
+            # 按指定间隔提取帧
+            if frame_count % frame_interval == 0:
+                # 缩放帧以控制最短边长度
+                frame = _resize_frame(frame, max_short_edge)
+                # 编码为 JPEG 并转 base64
+                _, buffer = cv2.imencode('.jpg', frame)
+                frame_base64 = base64.b64encode(buffer).decode('utf-8')
+                frames_base64.append(frame_base64)
+
+            frame_count += 1
+
+        cap.release()
+
+        if len(frames_base64) > max_frames:
+            import math
+            step = len(frames_base64) / max_frames
+            sampled = [frames_base64[min(int(i * step), len(frames_base64) - 1)] for i in range(max_frames)]
+            frames_base64 = sampled
+
+        # 构建消息内容：提示词 + 所有帧图片
+        video_prompt = (
+            f"以下是从视频({video_name})中按时间顺序提取的 {len(frames_base64)} 帧画面，视频原始帧率为 {video_fps:.2f} fps，"
+            f"抽帧间隔为 {frame_interval} 帧（约每 {frame_interval / video_fps:.2f} 秒一帧），请将它们视为一个连续的视频进行分析。"
+        )
+        content = []
+        if prompt_start is not None:
+            content.append({"type": "text", "text": prompt_start})
+        content.append({"type": "text", "text": video_prompt})
+
+        for frame_base64 in frames_base64:
+            content.append({
+                "type": "image_url",
+                "image_url": {
+                    "url": f"data:image/jpeg;base64,{frame_base64}"
+                }
+            })
+        if prompt_end is not None:
+            content.append({"type": "text", "text": prompt_end})
+        return content
+
+    finally:
+        # 清理临时文件
+        if temp_file and Path(temp_file.name).exists():
+            Path(temp_file.name).unlink()

+from football_replay_match_live import FootballReplayMatchLive
+from qwen_asr_util import QwenAsr
+from football_replay_video_event_by_llm import FootballReplayVideoEvent
+import os
+import json
+
+
+def batch_match():
+    replay_match_live = FootballReplayMatchLive(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0")
+    qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B")
+    fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7)
+
+    videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
+    for video_name in sorted(os.listdir(videos_dir)):
+
+        live_dir = os.path.join(videos_dir, video_name, 'live')
+        live_video_dir = os.path.join(live_dir, 'videos')
+        live_asr_dir = os.path.join(live_dir, 'asr')
+
+        live_packs = []
+        for live_video_name in sorted(os.listdir(live_video_dir)):
+            live_video_path = os.path.join(live_video_dir, live_video_name)
+            live_asr_path = os.path.join(live_asr_dir, f"{live_video_name}.txt")
+            # live_asr_text = qwen_asr.asr(live_video_path, cache_path=live_asr_path)
+            live_asr_text = ''
+            live_packs.append({
+                "video_id": os.path.basename(live_video_path),
+                "video_path": live_video_path,
+                "asr_text": live_asr_text
+            })
+
+        replays_dir = os.path.join(videos_dir, video_name, 'replays')
+        replay_videos_dir = os.path.join(replays_dir, 'videos')
+        replay_goals_dir = os.path.join(replays_dir, 'goals')
+        replay_asr_dir = os.path.join(replays_dir, 'asr')
+        replay_matches_dir = os.path.join(replays_dir, 'matches')
+
+        for replay_video_name in sorted(os.listdir(replay_videos_dir)):
+            replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
+            replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
+
+            replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.txt")
+            replay_match_path = os.path.join(replay_matches_dir, f"{replay_video_name}.json")
+
+            # replay_asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
+            replay_asr_text = ''
+
+            replay_goals = fbrv.video_event(replay_video_path, cache_path=replay_goals_path)
+            if replay_goals['event_name'] == '无进球':
+                continue
+            replay_pack = {"video_path": replay_video_path, "asr_text": replay_asr_text}
+
+            try:
+                result = replay_match_live.match_batch(replay_pack, live_packs, max_parallel=2,
+                                                       cache_path=replay_match_path)
+                print(replay_video_path)
+                print(result)
+
+            except Exception as e:
+                print(f"Error processing {replay_video_path}: {e}")
+
+            print('-' * 20)
+
+
+if __name__ == '__main__':
+    batch_match()

+import os
+from io import BytesIO
+import re
+import requests
+import base64
+import tempfile
+from pathlib import Path
+from typing import Union
+import httpx
+import subprocess
+
+from langchain_core.messages import HumanMessage
+from langchain_openai import ChatOpenAI
+
+
+class QwenAsr:
+    def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key'):
+        self.base_url = base_url
+        self.model = model
+        self.temperature = temperature
+        self.api_key = api_key
+        # self.asr_model = ChatOpenAI(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B", temperature=0.7,
+        #                        api_key='no_key')
+        self.asr_model = ChatOpenAI(base_url=self.base_url, model=self.model, temperature=self.temperature,
+                                    api_key=self.api_key)
+
+    @staticmethod
+    def _audio_content(source: Union[str, Path]) -> list:
+        """
+        读取音频/视频文件并构建 LLM 消息内容。
+        如果 source 是视频，会先提取音频；如果是 HTTP 音频 URL，直接引用 URL。
+
+        Args:
+            source: 音频/视频文件路径或 URL
+        Returns:
+            list: LLM 消息内容列表
+        """
+        source_str = str(source)
+        audio_exts = {"wav", "mp3", "m4a", "flac", "ogg", "webm", "aac", "wma"}
+        video_exts = {"mp4", "avi", "mkv", "mov", "wmv", "flv", "webm", "mpeg", "mpg", "ts", "3gp", "m4v"}
+        temp_files = []
+
+        try:
+            content = []
+            is_url = source_str.startswith(("http://", "https://"))
+            ext = Path(source_str).suffix.lstrip(".").lower()
+
+            # HTTP 音频 URL：直接引用，无需下载
+            if is_url and ext in audio_exts:
+                # audio_prompt = f"以下是音频文件({audio_name})的内容，请进行分析。"
+                # content = [{"type": "text", "text": audio_prompt}]
+                # if prompt is not None:
+                #     content.append({"type": "text", "text": prompt})
+                content.append({
+                    "type": "audio_url",
+                    "audio_url": {
+                        "url": source_str
+                    }
+                })
+                return content
+
+            # 本地文件或需要下载的 URL（视频或其他媒体）
+            if is_url:
+                resp = httpx.get(source_str, timeout=60.0)
+                resp.raise_for_status()
+                suffix = Path(source_str).suffix or ".mp4"
+                temp_file = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
+                temp_file.write(resp.content)
+                temp_file.close()
+                temp_files.append(temp_file.name)
+                media_path = temp_file.name
+            else:
+                media_path = str(source)
+                if not Path(media_path).exists():
+                    raise FileNotFoundError(f"文件不存在: {media_path}")
+
+            media_ext = Path(media_path).suffix.lstrip(".").lower()
+
+            # 如果是视频，提取音频为 wav
+            if media_ext in video_exts or (is_url and ext not in audio_exts):
+                temp_audio = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
+                temp_audio.close()
+                temp_files.append(temp_audio.name)
+                cmd = [
+                    "ffmpeg",
+                    "-y",
+                    "-i", media_path,
+                    "-vn",
+                    "-acodec", "pcm_s16le",
+                    "-ar", "16000",
+                    "-ac", "1",
+                    "-hide_banner",
+                    "-loglevel", "error",
+                    temp_audio.name
+                ]
+                result = subprocess.run(cmd, capture_output=True, text=True)
+                if result.returncode != 0:
+                    raise RuntimeError(f"ffmpeg 提取音频失败: {result.stderr.strip()}")
+                audio_path = temp_audio.name
+                media_ext = "wav"
+            else:
+                audio_path = media_path
+                if media_ext not in audio_exts:
+                    media_ext = "wav"
+
+            # 读取音频并 base64 编码
+            with open(audio_path, "rb") as f:
+                audio_bytes = f.read()
+            audio_base64 = base64.b64encode(audio_bytes).decode("utf-8")
+
+            # 构建消息内容
+            # audio_prompt = f"以下是音频文件({audio_name})的内容，请进行分析。"
+            # content = [{"type": "text", "text": audio_prompt}]
+            # if prompt is not None:
+            #     content.append({"type": "text", "text": prompt})
+
+            content.append({
+                "type": "audio_url",
+                "audio_url": {
+                    "url": f"data:audio/{media_ext};base64,{audio_base64}"
+                }
+            })
+
+            return content
+        finally:
+            # 清理所有临时文件
+            for tf in temp_files:
+                p = Path(tf)
+                if p.exists():
+                    p.unlink()
+
+    @staticmethod
+    def _extract_asr_text(text: str) -> str:
+        """
+        从ASR响应文本中提取<asr_text>标签内容
+
+        Args:
+            text: 原始响应文本，如 'language Chinese<asr_text>放大一点一倍。'
+
+        Returns:
+            str: 提取的转录文本，如 '放大一点一倍。'
+        """
+        # 尝试匹配 <asr_text>...</asr_text>
+        match = re.search(r'<asr_text>(.*?)(?:</asr_text>|$)', text, re.DOTALL)
+        if match:
+            return match.group(1).strip()
+        return text
+
+    def asr(self, source, cache_path:str=None):
+        if cache_path is not None and os.path.exists(cache_path):
+            with open(cache_path, 'r', encoding='utf-8') as f:
+                return f.read()
+        contents = self._audio_content(source)
+        message = HumanMessage(content=contents)
+        result = self.asr_model.invoke([message])
+        asr_text = self._extract_asr_text(result.content)
+        if cache_path is not None:
+            os.makedirs(Path(cache_path).parent, exist_ok=True)
+            with open(cache_path, 'w', encoding='utf-8') as f:
+                f.write(asr_text)
+        return asr_text

+import json
+import os
+
+from football_replay_video_event_by_llm import FootballReplayVideoEvent
+from qwen_asr_util import QwenAsr
+def batch_with_asr():
+    fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7)
+    qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B",temperature=0.7,
+                           api_key='no_key')
+
+    videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
+    for video_name in sorted(os.listdir(videos_dir)):
+        replays_dir = os.path.join(videos_dir, video_name, 'replays')
+        replay_videos_dir = os.path.join(replays_dir, 'videos')
+        replay_goals_dir = os.path.join(replays_dir, 'goals')
+        replay_asr_dir = os.path.join(replays_dir, 'asr')
+
+        for replay_video_name in sorted(os.listdir(replay_videos_dir)):
+            replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
+            replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
+            replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.json")
+            if os.path.exists(replay_goals_path):
+                print("skip", replay_video_path)
+                continue
+            asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
+            event_json = fbrv.video_event(replay_video_name, asr_text)
+            print(event_json)
+
+def batch_without_asr():
+    fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7)
+    # qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B", temperature=0.7,
+    #                    api_key='no_key')
+
+    videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
+    for video_name in sorted(os.listdir(videos_dir)):
+        replays_dir = os.path.join(videos_dir, video_name, 'replays')
+        replay_videos_dir = os.path.join(replays_dir, 'videos')
+        replay_goals_dir = os.path.join(replays_dir, 'goals')
+        replay_asr_dir = os.path.join(replays_dir, 'asr')
+
+        for replay_video_name in sorted(os.listdir(replay_videos_dir)):
+            replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
+            replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
+            replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.json")
+            if os.path.exists(replay_goals_path):
+                print("skip", replay_video_path)
+                continue
+            # asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
+            event_json = fbrv.video_event(replay_video_path, None, cache_path=replay_goals_path)
+            print(replay_video_name)
+            print(event_json)
+
+def one_video_test(video_path):
+    fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7)
+    event_json = fbrv.video_event(video_path, None)
+    print(event_json)
+
+if __name__ == '__main__':
+    # one_video_test(rf"D:\Code\py260417\src\llm_demo\lc_t\ftb\replay_1.mp4")
+    batch_without_asr()

+import os
+import json
+from pathlib import Path
+
+from aabd.base.time_util import vms2str_auto
+from langchain_ollama import ChatOllama
+
+
+class ClipByEvent:
+    def __init__(self, base_url, model, temperature=0.0, api_key='no_key'):
+        self.base_url = base_url
+        self.model = model
+        self.temperature = temperature
+        self.api_key = api_key
+        self.clip_before = 10 * 1000
+        self.clip_after = 5 * 1000
+
+        # self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
+        #                    keep_alive=-1, reasoning=False)
+        # self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
+        #                        api_key='no_key')
+        self.model = ChatOllama(base_url=base_url, model=model, temperature=temperature, keep_alive=-1, reasoning=False)
+
+    def get_video_match_time(self, video_path, video_time):
+        import cv2
+        import base64
+        from langchain_core.messages import HumanMessage
+
+        # 从视频中精确提取指定时间的帧画面
+        cap = cv2.VideoCapture(video_path)
+        if not cap.isOpened():
+            raise ValueError(f"无法打开视频文件: {video_path}")
+
+        fps = cap.get(cv2.CAP_PROP_FPS)
+        # video_time 单位为毫秒，转换为帧号
+        frame_number = int((video_time / 1000.0) * fps)
+        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_number)
+        ret, frame = cap.read()
+        cap.release()
+
+        if not ret:
+            raise ValueError(f"无法从视频 {video_path} 中提取时间 {video_time}ms 的帧画面")
+
+        # 将帧图像编码为base64
+        _, buffer = cv2.imencode('.jpg', frame)
+        image_base64 = base64.b64encode(buffer).decode('utf-8')
+
+        # 使用大模型分析比赛画面中的时间
+        message = HumanMessage(
+            content=[
+                {
+                    "type": "text",
+                    "text": "请仔细观察这张比赛画面截图，找出画面中显示的比赛计时器或时间信息，并以'MM:SS'格式返回比赛时间。如果无法识别，请返回'unknown'。"
+                },
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": f"data:image/jpeg;base64,{image_base64}"
+                    }
+                }
+            ]
+        )
+        response = self.model.invoke([message])
+        match_time_text = response.content.strip()
+
+        if match_time_text == 'unknown':
+            raise ValueError(f"无法识别视频 {video_path} 中时间 {video_time}ms 的比赛时间")
+        return match_time_text
+
+    def clip_video(self, video_path, clip_time, out_path):
+        # 精确截取视频指定视频段,从新编码
+
+        # 毫秒
+        start_time = clip_time - self.clip_before
+        end_time = clip_time + self.clip_after
+        
+        start_time_s = start_time / 1000.0
+        duration_s = (end_time - start_time) / 1000.0
+        
+        import subprocess
+        import os
+        
+        os.makedirs(os.path.dirname(out_path), exist_ok=True)
+        
+        cmd = [
+            'ffmpeg',
+            '-i', video_path,
+            '-ss', str(start_time_s),
+            '-t', str(duration_s),
+            '-c:v', 'libx264',
+            '-c:a', 'aac',
+            '-y',
+            out_path
+        ]
+        
+        subprocess.run(cmd, check=True)
+
+    def clip_by_event(self, video_dir, json_dir, out_dir):
+
+        for video_name in sorted(os.listdir(video_dir)):
+            video_path = os.path.join(video_dir, video_name)
+            f_name_0 = Path(video_name).stem
+            json_path = os.path.join(json_dir, f"{f_name_0}.json")
+            with open(json_path, 'r', encoding='utf-8') as f:
+                data = json.load(f)
+
+            events = data['dataObject']['data']['events']
+
+            event0 = data['dataObject']['data']['events'][0]
+            event0_sei_utc = event0['seiUtc']
+            event0_text_time = event0['eventTimeText']
+            mm, ss = map(int, event0_text_time.split(':'))
+            event0_time_seconds = mm * 60 + ss
+            event0_time_ms = event0_time_seconds * 1000
+            match_start_sei_utc = event0_sei_utc - event0_time_ms
+
+            take_frame_time = 30 * 60 * 1000
+            match_text_time = self.get_video_match_time(video_path, take_frame_time)
+            mm, ss = map(int, match_text_time.split(':'))
+            match_time_seconds = mm * 60 + ss
+            match_time_ms = match_time_seconds * 1000
+            match_start_video_time = take_frame_time - match_time_ms
+
+            video_start_sei_utc = match_start_sei_utc - match_start_video_time
+
+            start_utc = video_start_sei_utc
+
+            os.makedirs(os.path.join(out_dir, f_name_0), exist_ok=True)
+
+            # 样例
+            # 41	传球	曼联,约罗	00:36	1776106869512	638160	00:10:38.160
+            # 41	传球	曼联,马兹拉维	00:39	1776106872512	641160	00:10:41.160
+            event_list = []
+            for event in events:
+                event_enum = event['eventType']
+                eventTitle = event['eventTitle']
+                event_array = eventTitle.split(' ')
+                event_2 = event_array[-1]
+                event_1 = ','.join(event_array[:-1])
+                eventTimeText = event['eventTimeText']
+                seiUtc = event['seiUtc']
+                event_list.append([event_enum, event_2, event_1, eventTimeText, seiUtc, seiUtc - start_utc,
+                                   vms2str_auto(seiUtc - start_utc)])
+                # out_f.write(
+                #     f'{event_enum}\t{event_2}\t{event_1}\t{eventTimeText}\t{seiUtc}\t{seiUtc - start_utc}\t{vms2str_auto(seiUtc - start_utc)}\n')
+
+            with open(os.path.join(out_dir, f_name_0, f"sport_events.txt"), 'w', encoding='utf-8') as out_f:
+                for event in event_list:
+                    out_f.write(
+                        f'{event[0]}\t{event[1]}\t{event[2]}\t{event[3]}\t{event[4]}\t{event[5]}\t{event[6]}\n')
+
+            for event in event_list:
+                if event[1] == '进球':
+                    out_path = os.path.join(out_dir, f_name_0, 'videos',
+                                            f"{event[6].replace(':', '-').replace('.', '-')}.mp4")
+
+                    if not os.path.exists(out_path):
+                        print(f"Clipping {video_name} {event[6]} to {out_path}")
+                        self.clip_video(video_path, event[5], out_path)
+                    else:
+                        print(f"Clip skip {video_name} {event[6]} as {out_path} already exists")
+
+
+if __name__ == '__main__':
+    cbe = ClipByEvent(base_url="http://192.168.1.215:11434", model="qwen3.5:9b-q8_0", temperature=0.7)
+    cbe.clip_by_event(video_dir=rf"L:\wdx\202604\20260420_download_football_video\downloads",
+                      json_dir=rf"D:\Code\py260417\src\llm_demo\foot_event_clips\format_json",
+                      out_dir=rf"R:\wdx\202604\20260420_download_football_video\finished1")