api.py
6.58 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
import logging
from aabd.base.enhance_dict import value_or_default
logger = logging.getLogger(__name__)
class FootballReplayMatch:
def __init__(self, settings):
self.settings = settings
self.match_by_time_threshold = value_or_default(settings.match_by_time_threshold, 30) * 1000
def match_by_time(self, replay, events):
start_utc = replay.get('start_utc')
events = [(start_utc - e.get('event_utc'), e) for e in events]
events.sort(key=lambda x: abs(x[0]))
time_diff, event = events[0]
if time_diff < self.match_by_time_threshold:
return event
else:
return None
def det_goal_replay(self, replay):
pass
def match_by_llm(self, replay, events):
pass
def replay_match_event(self, data):
"""
:param data:
{
id:xx,
match_id: 比赛id，
replay:{
id:,
url:,
start_utc:,
end_utc:,
}
events:[{
id:xx,
type:xx,
url:xx,
event_utc,
}]
}
:return:
"""
# 该时间段附近是否有进球事件
replay_info = data['replay']
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)
if matched_event is None:
replay_det_info = self.det_goal_replay(replay_info)
replay_event_name = replay_det_info.get('event_name', '')
if replay_event_name == '进球':
matched_event = self.match_by_llm(replay_info, goal_live_events)
if matched_event is None:
logger.info(f'LLM认为是进球但是未能找到匹配的事件')
else:
logger.info('LLM找到进球的事件')
else:
logger.info(f'LLM判断不是进球，无需匹配')
else:
logger.info(f'通过时间找到匹配的事件')
from .llm_image import Video2Frame
from football_replay_match.core.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 为完整视频，通过 event_utc 向前延长10秒、向后延长5秒截取片段
live_videos = []
for event in events:
event_id = event.get("id")
event_url = event.get("url")
event_utc = event.get("event_utc")
if not event_url or event_utc is None:
continue
try:
event_start = event_utc - 10
event_end = event_utc + 5
event_frames = v2f.to_frames(
url=event_url,
start_utc=event_start,
end_utc=event_end,
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,
}