lizhengwei / aigc-embedding-service

Go to a project
Toggle navigation Toggle navigation pinning
lizhengwei
b725d68d 1 parent cfcc7f7a

jira:NYJ-1656 desc: modified cache dir and url of m3u8/mp4

Inline Side-by-side
Showing 3 changed files with 28 additions and 173 deletions
... ... @@ -56,11 +56,21 @@ class FootballReplayMatch:
return None
def det_goal_replay(self, replay, task_id):
return self.videoEventRecognition.video_event(replay, cache_dir=os.path.join(self.cache_dir, "replays"), task_log_dir=os.path.join(self.task_log_dir, task_id))
video_url = replay.get('url')
if video_url.endswith('.m3u8'):
mp4_name = os.path.basename(video_url)
else:
mp4_name = os.path.basename(os.path.dirname(os.path.dirname(os.path.dirname(video_url))))
return self.videoEventRecognition.video_event(replay, cache_dir=os.path.join(self.cache_dir, mp4_name, "replays"), task_log_dir=os.path.join(self.task_log_dir, mp4_name, task_id))
def match_by_llm(self, replay, events, task_id):
return self.videoMatchLive.match_batch(replay, events, max_parallel=2, cache_dir=os.path.join(self.cache_dir, "events"), task_log_dir=os.path.join(self.task_log_dir, task_id))
video_url = replay.get('url')
if video_url.endswith('.m3u8'):
mp4_name = os.path.basename(video_url)
else:
mp4_name = os.path.basename(os.path.dirname(os.path.dirname(os.path.dirname(video_url))))
return self.videoMatchLive.match_batch(replay, events, max_parallel=2, cache_dir=os.path.join(self.cache_dir, mp4_name, "events"), task_log_dir=os.path.join(self.task_log_dir, mp4_name, task_id))
def replay_match_event(self, data):
"""
... ...
"""
api_test.py - 对 api.py 中 FootballReplayMatch.replay_match_event 的单元测试
api_test.py - 对 api.py 中 FootballReplayMatch.replay_match_event 的测试
使用伪数据覆盖主要分支：
1. 时间阈值内直接匹配成功
... ... @@ -20,10 +20,6 @@ try:
except ImportError:
from .api import FootballReplayMatch
from pathlib import Path
def get_video_files_pathlib(folder_path, extensions=None):
if extensions is None:
... ... @@ -36,17 +32,17 @@ def get_video_files_pathlib(folder_path, extensions=None):
if f.is_file() and f.suffix.lower() in extensions]
return sorted(image_files)
def _fake_settings(match_by_time_threshold: int = 30, cache_dir_tag: str = "test"):
def _fake_settings():
"""构造一个 settings 伪对象，兼容 api.py 中的 value_or_default 读取逻辑。"""
return SimpleNamespace(
match_by_time_threshold=match_by_time_threshold,
match_by_time_threshold=30,
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=f"/root/lzw/tmp_0601_replay_cache_{cache_dir_tag}/videos_cache",
task_log_dir=f"/root/lzw/tmp_0601_replay_cache_{cache_dir_tag}/tasks_log"
common=SimpleNamespace(cache_dir=f"/root/lzw/tmp_0602_replay_cache/videos_cache",
task_log_dir=f"/root/lzw/tmp_0602_replay_cache/tasks_log"
),
save_frames_enable=True,
)
... ... @@ -102,36 +98,21 @@ def _base_data(i, replay_path, event_lis):
}
def test_match_by_time_success():
# """场景1：时间差在阈值内，直接通过 match_by_time 命中。"""
replays_and_events_pair = get_replays_and_events_pair()
for replays_mp4_dir, events_mp4_dir in replays_and_events_pair:
# replays_mp4_dir = "/root/lzw/finished/69e5102bfd87d43d09bba90d-auto-10154993/replays/videos"
# events_mp4_dir = r"/root/lzw/finished/69e5102bfd87d43d09bba90d-auto-10154993/live/videos"
print(100*"=")
print(f"replays_mp4_dir: {replays_mp4_dir}, events_mp4_dir: {events_mp4_dir}")
tag = os.path.basename(os.path.dirname(os.path.dirname(replays_mp4_dir)))
settings = _fake_settings(match_by_time_threshold=30, cache_dir_tag=tag) # threshold = 30 * 1000 = 30000 ms
settings = _fake_settings()
frm = FootballReplayMatch(settings)
relays_mp4_names = get_video_files_pathlib(replays_mp4_dir)
event_list = get_event_list(events_mp4_dir)
all_time = 0.0
max_time = 0.0
for i, replay_name in enumerate(relays_mp4_names):
... ... @@ -157,145 +138,5 @@ def test_match_by_time_success():
print(f"all_time: {all_time}")
# exit()
if __name__ == "__main__":
test_match_by_time_success()
# self.assertEqual(result["id"], "task_001")
# self.assertEqual(result["match_id"], "match_001")
# self.assertEqual(result["replay_id"], "replay_001")
# # event_001 与 replay.start_utc 差值 20 ms < 30000 ms，应命中 event_001 的 id
# self.assertEqual(result["event_id"], "event_001")
# # 未调用 LLM 相关方法
# mock_video_cls.return_value.video_event.assert_not_called()
# mock_live_cls.return_value.match_batch.assert_not_called()
# @patch("utils.football_replay_match_live.FootballReplayMatchLive")
# @patch("utils.football_replay_video_event_by_llm.FootballReplayVideoEvent")
# def test_match_by_time_fail_then_not_goal(self, mock_video_cls, mock_live_cls):
# """场景2：时间匹配失败，det_goal_replay 判定不是进球。"""
# mock_video_instance = MagicMock()
# mock_video_instance.video_event.return_value = {"event_name": "无进球", "description": "宣传片"}
# mock_video_cls.return_value = mock_video_instance
# settings = _fake_settings(match_by_time_threshold=30)
# frm = FootballReplayMatch(settings)
# data = _base_data()
# result = frm.replay_match_event(data)
# self.assertEqual(result["event_id"], None)
# mock_video_instance.video_event.assert_called_once()
# mock_live_cls.return_value.match_batch.assert_not_called()
# @patch("utils.football_replay_match_live.FootballReplayMatchLive")
# @patch("utils.football_replay_video_event_by_llm.FootballReplayVideoEvent")
# def test_match_by_time_fail_then_goal_then_llm_match(self, mock_video_cls, mock_live_cls):
# """场景3：时间匹配失败 -> 判定是进球 -> LLM 匹配成功。"""
# mock_video_instance = MagicMock()
# mock_video_instance.video_event.return_value = {"event_name": "进球", "description": "世界波"}
# mock_video_cls.return_value = mock_video_instance
# mock_live_instance = MagicMock()
# mock_live_instance.match_batch.return_value = {"video_id": "event_003", "video_path": "http://example.com/event3.mp4", "asr_text": ""}
# mock_live_cls.return_value = mock_live_instance
# settings = _fake_settings(match_by_time_threshold=30)
# frm = FootballReplayMatch(settings)
# data = _base_data()
# result = frm.replay_match_event(data)
# self.assertEqual(result["event_id"], "event_003")
# mock_video_instance.video_event.assert_called_once()
# mock_live_instance.match_batch.assert_called_once()
# @patch("utils.football_replay_match_live.FootballReplayMatchLive")
# @patch("utils.football_replay_video_event_by_llm.FootballReplayVideoEvent")
# def test_match_by_time_fail_then_goal_then_llm_no_match(self, mock_video_cls, mock_live_cls):
# """场景4：时间匹配失败 -> 判定是进球 -> LLM 未匹配到任何事件。"""
# mock_video_instance = MagicMock()
# mock_video_instance.video_event.return_value = {"event_name": "进球", "description": "头球破门"}
# mock_video_cls.return_value = mock_video_instance
# mock_live_instance = MagicMock()
# mock_live_instance.match_batch.return_value = None
# mock_live_cls.return_value = mock_live_instance
# settings = _fake_settings(match_by_time_threshold=30)
# frm = FootballReplayMatch(settings)
# data = _base_data()
# result = frm.replay_match_event(data)
# self.assertEqual(result["event_id"], None)
# mock_video_instance.video_event.assert_called_once()
# mock_live_instance.match_batch.assert_called_once()
# @patch("utils.football_replay_match_live.FootballReplayMatchLive")
# @patch("utils.football_replay_video_event_by_llm.FootballReplayVideoEvent")
# def test_no_goal_events(self, mock_video_cls, mock_live_cls):
# """场景5：events 中没有任何 type='1' 的进球事件。"""
# mock_video_instance = MagicMock()
# mock_video_instance.video_event.return_value = {"event_name": "无进球", "description": "无候选事件"}
# mock_video_cls.return_value = mock_video_instance
# settings = _fake_settings(match_by_time_threshold=30)
# frm = FootballReplayMatch(settings)
# data = _base_data()
# data["events"] = [
# {"id": "event_005", "type": "2", "url": "http://example.com/event5.mp4", "event_utc": 1020},
# ]
# result = frm.replay_match_event(data)
# self.assertEqual(result["event_id"], None)
# mock_video_instance.video_event.assert_called_once()
# mock_live_cls.return_value.match_batch.assert_not_called()
# @patch("utils.football_replay_match_live.FootballReplayMatchLive")
# @patch("utils.football_replay_video_event_by_llm.FootballReplayVideoEvent")
# def test_empty_events(self, mock_video_cls, mock_live_cls):
# """场景6：events 为空列表。"""
# mock_video_instance = MagicMock()
# mock_video_instance.video_event.return_value = {"event_name": "无进球", "description": "空列表"}
# mock_video_cls.return_value = mock_video_instance
# settings = _fake_settings(match_by_time_threshold=30)
# frm = FootballReplayMatch(settings)
# data = _base_data()
# data["events"] = []
# result = frm.replay_match_event(data)
# self.assertEqual(result["event_id"], None)
# mock_video_instance.video_event.assert_called_once()
# mock_live_cls.return_value.match_batch.assert_not_called()
# @patch("utils.football_replay_match_live.FootballReplayMatchLive")
# @patch("utils.football_replay_video_event_by_llm.FootballReplayVideoEvent")
# def test_missing_replay_key(self, mock_video_cls, mock_live_cls):
# """场景7：data 缺少 replay 键，应抛出 KeyError。"""
# settings = _fake_settings(match_by_time_threshold=30)
# frm = FootballReplayMatch(settings)
# data = _base_data()
# del data["replay"]
# with self.assertRaises(KeyError):
# frm.replay_match_event(data)
... ...
... ... @@ -218,7 +218,7 @@ class FootballReplayMatchLive:
if result.strip() == "null":
result_json = {"video_id": None}
else:
try:
# 1. 深度清洗：去掉 Markdown 标记
cleaned_result = result.replace("```json", "").replace("```", "")
... ... @@ -232,15 +232,19 @@ class FootballReplayMatchLive:
json_str = json_str[:json_str.rfind('}') + 1]
else:
# 如果连花括号都找不到，直接抛出原始内容方便排查
print("无法提取JSON，原始内容:", result)
raise ValueError("响应中未找到有效的 JSON 对象")
json_str = {
"replay_summary": {
"visual_cues": ""
},
"video_id": "null"
}
logger.info("Unable_to_extract_JSON,original_content:", result)
# 3. 使用 strict=False 容忍字符串中的非法控制字符（如直接换行）
result_json = json.loads(json_str, strict=False)
except Exception as e:
print("JSON解析依然失败，清洗后的内容:", json_str if 'json_str' in locals() else result)
raise e
video_id = result_json.get("video_id", None)
result_live = live_map.get(video_id, None)
... ...