lizhengwei / aigc-embedding-service

Go to a project
Toggle navigation Toggle navigation pinning
lizhengwei
84014163 1 parent 51946aee

jira:NYJ-1540 desc: not think

Inline Side-by-side
Showing 4 changed files with 232 additions and 104 deletions
import os
import time
import logging
from aabd.base.enhance_dict import value_or_default
... ... @@ -25,6 +26,8 @@ class FootballReplayMatch:
def match_by_time(self, replay, events):
start_utc = replay.get('start_utc')
if start_utc is None:
return None
events = [(start_utc - e.get('event_utc'), e) for e in events if e.get('event_utc') is not None]
if len(events) == 0:
... ... @@ -79,30 +82,36 @@ class FootballReplayMatch:
# 该时间段附近是否有进球事件
replay_info = data['replay']
replay_id = replay_info.get('id')
replay_start_utc = replay_info.get('start_utc', None)
replay_end_utc = replay_info.get('end_utc', None)
live_events = data['events']
goal_live_events = [e for e in live_events if str(e.get('type', '')) == '1']
matched_event = self.match_by_time(replay_info, goal_live_events)
# print(f"{task_id}_match_by_time matched_event: {matched_event}")
if matched_event is None:
start_time = time.time()
replay_det_info = self.det_goal_replay(replay_info, task_id)
end_time = time.time()
logger.info(f"判断是否是进球耗时： {end_time - start_time} 秒")
print(f"判断是否是进球耗时： {end_time - start_time} 秒")
replay_event_name = replay_det_info.get('event_name', '')
if replay_event_name == '进球':
matched_event = self.match_by_llm(replay_info, goal_live_events, task_id)
if matched_event is None:
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 是进球但是未能找到匹配的事件")
# print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 是进球但是未能找到匹配的事件")
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 是进球但是未能找到匹配的事件")
print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 是进球但是未能找到匹配的事件")
else:
matched_event_id = matched_event.get('video_id')
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
# print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 对应的进球的事件视频是{matched_event_id}")
print(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 对应的进球的事件视频是{matched_event_id}")
else:
logger.info(f"{task_id}_LLM判断{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 不是进球，无需匹配")
# print(f"{task_id}_LLM判断{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 不是进球，无需匹配")
logger.info(f"{task_id}_LLM判断{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 不是进球，无需匹配")
print(f"{task_id}_LLM判断{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 不是进球，无需匹配")
else:
matched_event_id = matched_event.get('id')
logger.info(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
# print(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 对应的进球的事件视频是{matched_event_id}")
logger.info(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 对应的进球的事件视频是{matched_event_id}")
print(f"{task_id}_通过时间找到匹配认为视频{replay_info['url']}_{replay_start_utc}_{replay_end_utc} 对应的进球的事件视频是{matched_event_id}")
return {
"id": task_id,
"match_id": match_id,
... ...
... ... @@ -9,6 +9,7 @@ api_test.py - 对 api.py 中 FootballReplayMatch.replay_match_event 的单元测
5. 无进球事件 / events 为空
6. 边界：缺少 replay 键时抛出异常
"""
import os
import time
import unittest
from types import SimpleNamespace
... ... @@ -19,47 +20,65 @@ try:
except ImportError:
from .api import FootballReplayMatch
from pathlib import Path
def get_video_files_pathlib(folder_path, extensions=None):
if extensions is None:
extensions = {'.mp4'}
else:
extensions = {ext.lower() for ext in extensions}
folder = Path(folder_path)
image_files = [f.name for f in folder.iterdir()
if f.is_file() and f.suffix.lower() in extensions]
return sorted(image_files)
def _fake_settings(match_by_time_threshold: int = 30):
"""构造一个 settings 伪对象，兼容 api.py 中的 value_or_default 读取逻辑。"""
return SimpleNamespace(
match_by_time_threshold=match_by_time_threshold,
llm=SimpleNamespace(base_url="http://192.168.1.59:11434/v1", model_name="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf", temperature=0.7),
common=SimpleNamespace(cache_dir="/root/lzw/tmp_0526_replay_cache0526/videos_cache", task_log_dir="/root/lzw/tmp_0526_replay_cache0526/tasks_log"),
llm=SimpleNamespace(base_url="http://192.168.1.59:11434/v1",
model_name="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
api_key='no_key'
),
common=SimpleNamespace(cache_dir="/root/lzw/tmp_0527_replay_cache_08/videos_cache",
task_log_dir="/root/lzw/tmp_0527_replay_cache_08/tasks_log"
),
save_frames_enable=True,
)
def _base_data():
"""返回一份基础伪数据，replay.start_utc = 1000（毫秒）。"""
def get_event_list():
# url = "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8"
events_mp4_dir = r"/root/lzw/69e4ed7c872d573c9554510b-auto-10154973/goal"
events_mp4_names = get_video_files_pathlib(events_mp4_dir)
event_lis = []
for mp4 in events_mp4_names:
mp4_path = os.path.join(events_mp4_dir, mp4)
event_lis.append({
"id": f"event_{mp4.split('.')[0]}",
"type": "1",
"url": mp4_path,
# "event_utc": None
})
return event_lis
def _base_data(i, replay_path, event_lis):
return {
"id": "task_001",
"match_id": "match_001",
"id": f"task_{str(i).zfill(3)}",
"match_id": f"match_{str(i).zfill(3)}",
"replay": {
"id": "replay_001",
"url": "/root/lzw/finished/69dd5845dd0412067b8d5587-auto-1776074760653/replays/videos/00_14_32_080-00_15_10_560.mp4",
"start_utc": 5_000,
"end_utc": 21_000
},
"events": [
{"id": "event_001",
"type": "1",
"url": "/root/lzw/finished/69dd5845dd0412067b8d5587-auto-1776074760653/live/videos/00-14-09-052.mp4",
"event_utc": None
},
{"id": "event_002",
"type": "1",
"url": "/root/lzw/finished/69dd5845dd0412067b8d5587-auto-1776074760653/live/videos/00-38-40-052.mp4",
"event_utc": None
"id": f"replay_{str(i).zfill(3)}",
"url": replay_path,
# "start_utc": None,
# "end_utc": None
},
{"id": "event_003",
"type": "1",
"url": "/root/lzw/finished/69dd5845dd0412067b8d5587-auto-1776074760653/live/videos/01-36-04-972.mp4",
"event_utc": None
}
],
"events": event_lis,
}
... ... @@ -68,18 +87,39 @@ class TestReplayMatchEvent(unittest.TestCase):
@patch("utils.football_replay_match_live.FootballReplayMatchLive")
@patch("utils.football_replay_video_event_by_llm.FootballReplayVideoEvent")
def test_match_by_time_success(self, mock_video_cls, mock_live_cls):
"""场景1：时间差在阈值内，直接通过 match_by_time 命中。"""
# """场景1：时间差在阈值内，直接通过 match_by_time 命中。"""
settings = _fake_settings(match_by_time_threshold=10) # threshold = 30 * 1000 = 30000 ms
frm = FootballReplayMatch(settings)
data = _base_data()
replays_mp4_dir = "/root/lzw/69e4ed7c872d573c9554510b-auto-10154973/replay"
relays_mp4_names = get_video_files_pathlib(replays_mp4_dir)
event_list = get_event_list()
all_time = 0.0
max_time = 0.0
for i, replay_name in enumerate(relays_mp4_names):
print(100*"*" + f"replay_name: {replay_name}")
replay_path = os.path.join(replays_mp4_dir, replay_name)
data = _base_data(i, replay_path, event_list)
start_time = time.time()
result = frm.replay_match_event(data)
end_time = time.time()
print(f"cost time:{end_time - start_time}")
cost_time = end_time - start_time
all_time += cost_time
print(f"replay_name: {replay_name}, cost time:{cost_time}")
if max_time < cost_time:
max_time = cost_time
print(result)
print()
print()
print()
print(100*"#")
print(f"max_time: {max_time}")
print(f"all_time: {all_time}")
# exit()
# self.assertEqual(result["id"], "task_001")
... ...
import re
import json
import os.path
from pathlib import Path
... ... @@ -11,64 +12,103 @@ try:
except:
from llm_image import Video2Frame
# 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"
# }"""
req_prompt = """
# Role
你是一名拥有20年经验的足球视频技术分析师，擅长结合**视觉画面**与**解说音频（ASR）**进行跨镜头的事件匹配。你能通过解说员的描述锁定时间背景，并通过视觉特征确认物理细节。
你是一名拥有20年经验的足球视频技术分析师，擅长通过**纯视觉画面分析**进行跨镜头的事件匹配。你能精准捕捉球员动作、球衣号码、比赛时钟及比分牌等视觉细节。
# Task
输入包含：
1. 【回放片段】（Replay）：包含视频帧 + **对应的解说文本**。
2. 【直播进球片段列表】（Live Candidates）：包含视频帧 + **对应的解说文本**。
1. 【回放片段】（Replay）：包含视频帧画面。
2. 【直播进球片段列表】（Live Candidates）：包含视频帧画面。
目标：在【直播片段】中找到与【回放片段】属于**同一个进球事件**的片段。如果没有任何片段匹配，返回 null。
# Analysis Workflow (多模态分析流程)
请按照以下步骤进行推理：
### 第一步：解说词元数据提取 (听觉线索)
分析【回放片段】的解说文本，寻找以下关键信息：
- **时间指代**：解说员是否提到了具体时间？（如“上半场”、“第10分钟”、“开场不久”）。
- **事件描述**：解说员如何描述这个进球？（如“世界波”、“点球”、“补射”、“乌龙球”）。
- **球员提及**：解说员念出了谁的名字？
# Visual Analysis Rules (纯视觉匹配规则)
### 第二步：视觉物理指纹提取 (视觉线索)
忽略镜头语言（慢放、特写），提取核心物理特征：
- **进攻与射门**：进攻方式（传中/直塞）、射门部位（头/脚）、射门位置。
- **球路与防守**：球的轨迹（高/低/折射）、门将扑救动作（侧扑/倒地/未动）。
- **庆祝**：进球后的庆祝动作（仅作为辅助验证）。
1. **核心物理特征比对（最高优先级）**：
- **球员身份**：进攻方球衣颜色、进球者号码、助攻者号码。
- **进球方式**：射门部位（左脚/右脚/头球）、射门位置（禁区内/禁区外/点球）、球路轨迹（贴地斩/世界波/挑射）。
- **防守反应**：门将的扑救动作（向左/向右/未动）、后卫的封堵动作。
- **庆祝动作**：进球后的标志性庆祝（如滑跪、指天、双手比心等）。
- *判定标准*：上述物理特征必须完全一致，否则视为不同进球。
### 第三步：跨模态匹配与验证
遍历【直播片段】，结合视觉和听觉进行判断：
- **视觉一致性**：直播片段的动作、轨迹、门将反应是否与回放完全吻合？
- **听觉一致性**：
- 如果回放解说提到“这是第5分钟的进球”，而直播片段的时间戳是90分钟，**不要直接排除**，而是检查直播片段的解说是否也提到了“回顾第5分钟”或者直播片段的视觉内容确实是第5分钟的动作。
- **核心原则**：视觉物理特征 > 解说词描述 > 时间戳元数据。
- **特殊情况**：如果视觉特征高度相似（如同一个球员、同一个角度），但解说词明确说“这是另一个进球（例如：这是他的第二个进球，而回放说是第一个）”，则判定为不匹配。
2. **画面内嵌元数据验证**：
- **比分牌逻辑**：检查回放画面中的比分（如 1-0）与直播片段进球前后的比分变化是否逻辑自洽。
- **比赛时钟**：虽然回放可能是赛后集锦（时间戳不准），但如果回放画面明确显示比赛时间（如 18:36），而直播片段发生在完全不同的时间段（如 85:00），且视觉特征不符，则排除。
# Logic Constraints (逻辑约束 - 必须严格遵守)
3. **无匹配处理**：
- 如果候选片段的进球者号码、射门方式或球衣颜色与回放不一致，必须判定为无匹配。
- **时间单向性原则（铁律）**：
- **回看中的比赛时间一定在比赛片段画面时间之后**。
- 逻辑：回放是对过去发生事件的回顾。如果【回放片段】中解说提到的比赛时间（或画面显示的比赛时钟）**早于**【直播片段】中解说提到的比赛时间（或画面时钟），则**绝对不可能**是同一个事件。
- *示例*：回放解说在描述“第10分钟的进球”，而直播片段明确发生在“第5分钟”，则该直播片段**一定不是**目标。
- **时间陷阱**：回放可能是赛后集锦。如果解说员说“让我们看看**刚才**那个球”或者“**上半场**那个球”，即使当前比赛时间是90分钟，也要去匹配对应时间段的直播片段（或视觉特征）。
- **同名陷阱**：如果解说提到“又是**凯恩**进球了”，不能只看凯恩，必须看**怎么进的**（头球还是点球）。
- **无匹配处理**：如果所有候选片段在视觉动作（如射门方式、进球位置）或关键事件逻辑上与回放明显不符，必须判定为无匹配，将 `video_id` 设为 `null`。
# Output Format (输出要求)
### 输出要求
请**仅**输出一个标准的 JSON 对象，**不要**包含任何推理过程（reasoning）、Markdown 标记（如 ```json）或额外的解释文本。
请仅输出一个JSON格式的结果，不要输出任何分析过程。不要包含 markdown 标记（如 ```json ... ```），不要包含任何解释或额外文本。
格式如下：
JSON 格式如下：
{
"replay_summary": {
"audio_cues": "解说提到的关键信息（如：'第15分钟', '远射', '德布劳内'）",
"visual_cues": "视觉关键特征（如：'禁区外右脚', '球挂死角', '门将飞身扑救'）"
"visual_cues": "提取回放画面的核心特征（如：'蓝队6号禁区外远射', '门将扑救不及', '比分变为1-1'）"
},
"reasoning": "综合分析：回放解说提到是'上半场的远射'，视觉显示'17号球员禁区外起脚'。Candidate_1 视觉是'近距离推射'，排除。Candidate_2 视觉是'禁区外远射'，且门将动作一致，虽然直播时间显示是下半场（可能是集锦回顾），但解说也提到了'回顾上半场'，确认为同一事件。",
"video_id": "Candidate_2"
}"""
"video_id": "匹配到的直播片段ID，如果没有匹配则填 null"
}
"""
class FootballReplayMatchLive:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key', cache_dir:str=None, save_frames_enable:bool=False):
... ... @@ -81,7 +121,12 @@ class FootballReplayMatchLive:
# 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 = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
model_kwargs={
"extra_body": {"enable_thinking": False}
}
)
#extra_body={"chat_template_kwargs": {"enable_thinking": False}})
self.cache_dir = cache_dir
self.video2frame = Video2Frame(cache_dir=cache_dir, save_frames_enable=save_frames_enable)
... ... @@ -91,12 +136,12 @@ class FootballReplayMatchLive:
return None
elif len(live_videos) == 1:
live_video_path = live_videos[0].get("url", 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_videos[0].get("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,
"url": live_video_path,
# "asr_text": asr_text,
}
if record is not None:
record.append({"live": live, "llm_result": None, 'live_list': [live]})
... ... @@ -107,9 +152,9 @@ class FootballReplayMatchLive:
live_records = {}
for live_video in live_videos:
live_video_path = live_video.get("url", None)
live_video_id = live_video.get("video_id", os.path.basename(live_video_path))
live_video_id = live_video.get("id", os.path.basename(live_video_path))
event_utc = live_video.get("event_utc", None)
asr_text = live_video.get("asr_text", '')
# asr_text = live_video.get("asr_text", '')
live_video_contents = live_video.get("llm_contents", None)
if live_video_contents is None:
# event_start = event_utc - timedelta(seconds=10) if event_utc is not None else None
... ... @@ -124,38 +169,67 @@ class FootballReplayMatchLive:
roi=None,
max_px_area=400_000,
prompt_start=f"### 候选片段 video_id: {live_video_id} ###",
prompt_end=f"\n该片段解说内容: {asr_text}\n"
prompt_end=f"\n该片段解说内容: 无\n"
)
live_map[live_video_id] = {
"video_id": live_video_id,
"video_path": live_video_path,
"url": live_video_path,
"event_utc": event_utc,
"asr_text": asr_text,
# "asr_text": asr_text,
"contents": live_video_contents,
}
live_records[live_video_id] = {
"video_id": live_video_id,
"video_path": live_video_path,
"url": live_video_path,
"event_utc": event_utc,
"asr_text": asr_text,
# "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"})
user_contents.append({'type': "text", "text": "\n请根据上述片段进行匹配并按要求输出结果/no_think\n"})
system_message = SystemMessage(content=req_prompt)
user_message = HumanMessage(content=user_contents)
result = self.model.invoke([system_message, user_message]).content
print(100*"3")
print(result)
print(100*"4")
# 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
try:
result_json = json.loads(result)
except json.JSONDecodeError:
try:
result_json = json.loads(result.replace("```json", "").replace("```", ""))
# 1. 深度清洗：去掉 Markdown 标记
cleaned_result = result.replace("```json", "").replace("```", "")
# 2. 核心修复：使用 re.DOTALL 确保正则能跨行匹配，精准提取最外层完整的 JSON 对象
# 这样可以防止因为模型输出换行导致提取不完整
json_match = re.search(r'(\{.*\})', cleaned_result, re.DOTALL)
if json_match:
json_str = json_match.group(1)
# 截断第一个 } 之后的所有内容，彻底干掉 <time_location> 等尾巴
json_str = json_str[:json_str.rfind('}') + 1]
else:
# 如果连花括号都找不到，直接抛出原始内容方便排查
print("无法提取JSON，原始内容:", result)
raise ValueError("响应中未找到有效的 JSON 对象")
# 3. 使用 strict=False 容忍字符串中的非法控制字符（如直接换行）
result_json = json.loads(json_str, strict=False)
except Exception as e:
print("JSON解析失败:", result)
print("JSON解析依然失败，清洗后的内容:", json_str if 'json_str' in locals() else result)
raise e
video_id = result_json.get("video_id", None)
... ... @@ -233,9 +307,9 @@ class FootballReplayMatchLive:
if result is not None:
result_no_content = {
"video_id": result.get("video_id", None),
"video_path": result.get("video_path", None),
"url": result.get("url", None),
"event_utc": result.get("event_utc", None),
"asr_text": result.get("asr_text", None),
# "asr_text": result.get("asr_text", None),
}
else:
result_no_content = None
... ...
... ... @@ -76,15 +76,20 @@ class FootballReplayVideoEvent:
# 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 = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
extra_body={"chat_template_kwargs": {"enable_thinking": False}})
model_kwargs={
"extra_body": {"enable_thinking": False}
}
)
# extra_body={"enable_thinking": False})
# extra_body={"chat_template_kwargs": {"enable_thinking": False}})
self.video2frame = Video2Frame(cache_dir=cache_dir, save_frames_enable=save_frames_enable)
def video_event(self, replay_pack: dict, asr_text: str = '无', cache_dir=None, task_log_dir=None):
replay_video_path = replay_pack.get("url", None)
cache_path = os.path.join(task_log_dir, 'video_event.json')
video_name = os.path.basename(replay_video_path)
cache_path = os.path.join(task_log_dir, f'{video_name}_replay.json')
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())
... ... @@ -104,8 +109,8 @@ class FootballReplayVideoEvent:
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
# asr_message = HumanMessage(content=f"解说内容：{asr_text}")
result = self.model.invoke([system_message, video_message]).content
try:
result_json = json.loads(result)
except json.JSONDecodeError:
... ...