api.py
2.08 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
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'通过时间找到匹配的事件')