api.py
4.67 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import json
import os
from pathlib import Path
try:
from .llm_image import Video2Frame, frames2content
from ..util.football_replay_match_live import FootballReplayMatchLive
from config import settings
except ImportError:
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from core.llm_image import Video2Frame, frames2content
from util.football_replay_match_live import FootballReplayMatchLive
from config import settings
# 从配置中读取参数（配置项需在 config.yaml 或对应环境配置中定义）
_cache_dir = settings.common.get('cache_dir', None)
_llm_base_url = settings.llm.get('base_url', None)
_llm_model = settings.llm.get('model_name', None)
_llm_temperature = settings.llm.get('temperature', None)
# 初始化视频处理与匹配器
v2f = Video2Frame(_cache_dir)
matcher = FootballReplayMatchLive(
base_url=_llm_base_url,
model=_llm_model,
temperature=_llm_temperature,
)
def replay_match_event(data: dict) -> dict:
"""
足球回看与直播事件匹配接口。
请求参数:
id : str 任务唯一键
match_id : str 比赛id
replay : dict 回看信息
|- id : str 回看id
|- url : str 回看url(m3u8)
|- start_utc : int 回看开始utc时间
|- end_utc : int 回看结束utc时间
events : list 赛事事件片段（直播候选）
|- id : str 事件id
|- type : str 事件类型（1：进球）
|- url : str 事件片段url(m3u8)
|- event_utc : int 事件发生的utc时间点
响应参数:
id : str 任务唯一键
match_id : str 比赛id
replay_id : str 回看id
event_id : str|null 匹配到的比赛片段id，null表示无匹配
"""
task_id = data.get("id")
match_id = data.get("match_id")
replay = data.get("replay", {})
events = data.get("events", [])
replay_id = replay.get("id")
replay_url = replay.get("url")
replay_start = replay.get("start_utc")
replay_end = replay.get("end_utc")
if not replay_url or replay_start is None or replay_end is None:
raise ValueError("replay 参数不完整，需要 url、start_utc、end_utc")
# 1. 为回看视频提取帧并构建 LLM content
try:
replay_frames = v2f.to_frames(
url=replay_url,
start_utc=replay_start,
end_utc=replay_end,
fps=2,
)
replay_contents = frames2content(replay_frames)
# 保持与 football_replay_match_live.py 中一致的提示结构
replay_contents.insert(0, {"type": "text", "text": "\n【回放片段信息】\n"})
replay_contents.append({"type": "text", "text": "\n回放解说内容：无\n"})
except Exception as e:
raise RuntimeError(f"回看视频处理失败: {e}")
# 2. 为每个 event（直播候选）提取帧并构建 LLM content
# event 的 url 本身已是视频片段，直接处理完整片段
live_videos = []
for event in events:
event_id = event.get("id")
event_url = event.get("url")
if not event_url:
continue
try:
event_frames = v2f.to_frames(
url=event_url,
start_utc=0,
end_utc=999999, # 传极大值，实际以视频真实长度为准
fps=2,
)
event_contents = frames2content(event_frames)
event_contents.insert(
0, {"type": "text", "text": f"### 候选片段 video_id: {event_id} ###"}
)
event_contents.append({"type": "text", "text": "\n该片段解说内容: 无\n"})
except Exception:
# 单个 event 处理失败不影响整体流程
continue
live_videos.append({
"video_id": event_id,
"video_path": "", # 已提供 llm_contents，无需本地路径
"asr_text": "",
"llm_contents": event_contents,
})
# 3. 执行匹配
if len(live_videos) == 0:
matched_event_id = None
elif len(live_videos) == 1:
# 只有一个候选片段，直接视为匹配结果
matched_event_id = live_videos[0]["video_id"]
else:
try:
result = matcher._match_batch(replay_contents, live_videos, max_parallel=3)
matched_event_id = result.get("video_id") if result else None
except Exception:
matched_event_id = None
return {
"id": task_id,
"match_id": match_id,
"replay_id": replay_id,
"event_id": matched_event_id,
}