lizhengwei / aigc-embedding-service

Go to a project
Toggle navigation Toggle navigation pinning
lizhengwei
7cbb287a 1 parent fe373aed

jira:NYJ-1540 desc: adjust

Inline Side-by-side
Showing 11 changed files with 135 additions and 1532 deletions
... ... @@ -3,12 +3,23 @@ from aabd.base.enhance_dict import value_or_default
logger = logging.getLogger(__name__)
from utils.football_replay_video_event_by_llm import FootballReplayVideoEvent
from utils.football_replay_match_live import FootballReplayMatchLive
class FootballReplayMatch:
def __init__(self, settings):
self.settings = settings
self.match_by_time_threshold = value_or_default(settings.match_by_time_threshold, 30) * 1000
llm_base_url = value_or_default(settings.llm.base_url, None)
model_name = value_or_default(settings.llm.model_name, None)
temperature = value_or_default(settings.llm.temperature, 0.7)
self.cache_dir = value_or_default(settings.common.cache_dir, None)
save_frames_enable = value_or_default(settings.save_frames_enable, False)
self.videoEventRecognition = FootballReplayVideoEvent(llm_base_url, model_name, temperature, 'no_key', self.cache_dir, save_frames_enable)
self.videoMatchLive = FootballReplayMatchLive(llm_base_url, model_name, temperature, 'no_key', self.cache_dir)
def match_by_time(self, replay, events):
start_utc = replay.get('start_utc')
... ... @@ -21,10 +32,11 @@ class FootballReplayMatch:
return None
def det_goal_replay(self, replay):
pass
return self.videoEventRecognition.video_event(replay, cache_path=self.cache_dir)
def match_by_llm(self, replay, events):
pass
return self.videoMatchLive.match_batch(replay, events, max_parallel=2, cache_path=self.cache_dir)
def replay_match_event(self, data):
"""
... ... @@ -48,7 +60,9 @@ class FootballReplayMatch:
}
:return:
"""
task_id = data.get("id")
match_id = data.get("match_id")
# 该时间段附近是否有进球事件
replay_info = data['replay']
live_events = data['events']
... ... @@ -62,146 +76,28 @@ class FootballReplayMatch:
if replay_event_name == '进球':
matched_event = self.match_by_llm(replay_info, goal_live_events)
if matched_event is None:
logger.info(f'LLM认为是进球但是未能找到匹配的事件')
logger.info(f"{task_id}_LLM认为视频{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 是进球但是未能找到匹配的事件")
else:
logger.info('LLM找到进球的事件')
replay_id = replay_info.get('id')
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}")
return {
"id": task_id,
"match_id": match_id,
"replay_id": replay_id,
"event_id": matched_event_id,
}
else:
logger.info(f'LLM判断不是进球，无需匹配')
logger.info(f"{task_id}_LLM判断{replay_info['url']}_{replay_info['start_utc']}_{replay_info['end_utc']} 不是进球，无需匹配")
else:
logger.info(f'通过时间找到匹配的事件')
import json
import os
from pathlib import Path
from .llm_image import Video2Frame
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 为完整视频，通过 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,
}
replay_id = replay_info.get('id')
matched_event_id = matched_event.get('id')
return {
"id": task_id,
"match_id": match_id,
"replay_id": replay_id,
"event_id": matched_event_id,
}
\ No newline at end of file
... ...
import base64
import hashlib
import subprocess
from pathlib import Path
import cv2
try:
from .m3u8_to_mp4_sei import (
download_m3u8_to_mp4,
extract_h264_es_from_mp4,
parse_h264_sei,
extract_utc_from_sei,
)
except ImportError:
from m3u8_to_mp4_sei import (
download_m3u8_to_mp4,
extract_h264_es_from_mp4,
parse_h264_sei,
extract_utc_from_sei,
)
# 用于区分传入的 start_utc/end_utc 是绝对 UTC 时间戳还是秒偏移
_IS_ABSOLUTE_UTC_THRESHOLD = 1_000_000_000
class Video2Frame:
def __init__(self, cache_dir):
self.cache_dir = Path(cache_dir)
self.video_dir = self.cache_dir / "videos"
self.frames_dir = self.cache_dir / "frames"
self.video_dir.mkdir(parents=True, exist_ok=True)
self.frames_dir.mkdir(parents=True, exist_ok=True)
def to_frames(self, url, start_utc, end_utc, fps, roi=None, max_px_area=None) -> list:
"""
下载 m3u8 视频流的指定片段并提取帧，保存到 cache_dir 下。
- video_dir 中保存的是 start_utc ~ end_utc 截取后的视频片段（而非完整视频）。
- 如果视频中包含 SEI UTC 信息且传入的 start_utc/end_utc 为绝对时间戳，
则会基于 SEI UTC 进行帧级定位；否则将 start_utc/end_utc 视为秒偏移量。
Args:
url: m3u8 视频流地址。
start_utc: 截取开始时间（秒偏移或绝对 UTC 时间戳）。
end_utc: 截取结束时间（秒偏移或绝对 UTC 时间戳）。
fps: 目标抽帧帧率。
roi: 感兴趣区域 (x, y, w, h)，可选。
max_px_area: 最大像素面积，超过则等比例缩小，可选。
Returns:
提取的帧图片路径列表。
"""
if end_utc <= start_utc:
raise ValueError("end_utc 必须大于 start_utc")
unique_str = f"{url}_{start_utc}_{end_utc}"
unique_id = hashlib.md5(unique_str.encode("utf-8")).hexdigest()
clip_path = self.video_dir / f"{unique_id}.mp4"
frame_output_dir = self.frames_dir / unique_id
frame_output_dir.mkdir(parents=True, exist_ok=True)
# 判断是否为绝对 UTC 时间戳
is_absolute_utc = (
start_utc > _IS_ABSOLUTE_UTC_THRESHOLD
and end_utc > _IS_ABSOLUTE_UTC_THRESHOLD
)
# 若片段未缓存，先截取目标片段再保存
if not clip_path.exists():
if is_absolute_utc:
# 绝对 UTC 模式：先下载完整视频到临时文件，解析 SEI 后截取片段
temp_path = self.video_dir / f"{unique_id}_full.mp4"
download_m3u8_to_mp4(url, str(temp_path))
# 提取 SEI UTC 信息
utc_records = []
try:
es_data = extract_h264_es_from_mp4(str(temp_path))
sei_list = parse_h264_sei(es_data)
utc_records = extract_utc_from_sei(sei_list)
except Exception:
pass
if utc_records:
utc_list = [r["utc"] for r in utc_records]
start_frame_idx = self._find_frame_idx(
utc_list, start_utc, find_last_not_greater=False
)
end_frame_idx = self._find_frame_idx(
utc_list, end_utc, find_last_not_greater=True
)
cap_temp = cv2.VideoCapture(str(temp_path))
video_fps = cap_temp.get(cv2.CAP_PROP_FPS)
cap_temp.release()
if video_fps <= 0:
video_fps = 25.0
start_sec = start_frame_idx / video_fps
duration = (end_frame_idx - start_frame_idx) / video_fps
self._ffmpeg_extract_clip(
str(temp_path), str(clip_path), start_sec, duration
)
else:
# 无 SEI 无法定位，直接将完整视频重命名为片段
temp_path.rename(clip_path)
# 清理临时完整视频
if temp_path.exists():
temp_path.unlink(missing_ok=True)
else:
# 秒偏移模式：直接用 ffmpeg 从 URL 截取片段
duration = end_utc - start_utc
self._ffmpeg_download_clip(
url, str(clip_path), start_utc, duration
)
# 从截取后的片段中提取帧
cap = cv2.VideoCapture(str(clip_path))
if not cap.isOpened():
raise ValueError(f"无法打开视频片段: {clip_path}")
# total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
video_fps = cap.get(cv2.CAP_PROP_FPS)
if video_fps <= 0:
video_fps = 25.0
# 计算抽帧间隔
frame_interval = int(video_fps / fps) if fps > 0 else int(video_fps)
if frame_interval < 1:
frame_interval = 1
frames = []
saved_count = 0
frame_count = 0
while True:
ret, frame = cap.read()
if not ret:
break
if frame_count % frame_interval == 0:
if roi is not None:
x, y, w, h = roi
frame = frame[y:y + h, x:x + w]
frame = self.reisize_frame(frame, max_px_area)
frames.append(frame)
frame_path = frame_output_dir / f"frame_{saved_count:06d}.jpg"
cv2.imwrite(str(frame_path), frame)
saved_count += 1
frame_count += 1
cap.release()
return frames
def reisize_frame(self, frame, max_px_area):
if max_px_area is None:
return frame
h, w = frame.shape[:2]
area = h * w
if area > max_px_area:
scale = (max_px_area / area) ** 0.5
new_w = int(w * scale)
new_h = int(h * scale)
frame = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_AREA)
return frame
@staticmethod
def _find_frame_idx(utc_list, target_utc, find_last_not_greater=True):
"""
在 utc_list 中查找最接近 target_utc 的帧索引。
Args:
utc_list: 按帧顺序排列的 UTC 列表。
target_utc: 目标 UTC。
find_last_not_greater: True 返回最后一个 <= target_utc 的索引；
False 返回第一个 >= target_utc 的索引。
"""
if not utc_list:
return 0
if find_last_not_greater:
idx = 0
for i, utc in enumerate(utc_list):
if utc > target_utc:
break
idx = i
return idx
else:
for i, utc in enumerate(utc_list):
if utc >= target_utc:
return i
return len(utc_list) - 1
@staticmethod
def _ffmpeg_extract_clip(input_path, output_path, start_sec, duration):
"""使用 ffmpeg 从本地视频截取指定片段。"""
cmd = [
"ffmpeg", "-y",
"-i", input_path,
"-ss", str(start_sec),
"-t", str(duration),
"-c", "copy",
"-bsf:a", "aac_adtstoasc",
"-movflags", "+faststart",
output_path,
]
try:
subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except subprocess.CalledProcessError as e:
raise RuntimeError(
f"ffmpeg 截取片段失败: {e.stderr.decode('utf-8', errors='ignore')}"
)
@staticmethod
def _ffmpeg_download_clip(url, output_path, start_sec, duration):
"""使用 ffmpeg 从 URL 下载并截取指定片段。"""
cmd = [
"ffmpeg", "-y",
"-ss", str(start_sec),
"-fflags", "+discardcorrupt",
"-i", url,
"-t", str(duration),
"-c", "copy",
"-bsf:a", "aac_adtstoasc",
"-movflags", "+faststart",
output_path,
]
try:
subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except subprocess.CalledProcessError as e:
raise RuntimeError(
f"ffmpeg 下载片段失败: {e.stderr.decode('utf-8', errors='ignore')}"
)
def frames2content(frames):
"""
将帧列表（图片路径列表）转为 LLM content 列表。
格式与 llm_video_content.py 中的 contents 返回的 content 一致。
"""
contents = []
video_prompt = (
f"以下是从视频中按时间顺序提取的 {len(frames)} 帧画面，"
f"请将它们视为一个连续的视频进行分析。"
)
contents.append({"type": "text", "text": video_prompt})
for frame in frames:
_, buffer = cv2.imencode('.jpg', frame)
b64_str = base64.b64encode(buffer).decode('utf-8')
# with open(frame_path, "rb") as f:
# img_bytes = f.read()
# b64_str = base64.b64encode(img_bytes).decode("utf-8")
contents.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{b64_str}"
}
})
return contents
if __name__ == "__main__":
# pass
import argparse
import sys
import tempfile
parser = argparse.ArgumentParser(description="Video2Frame 测试脚本")
parser.add_argument(
"--url",
default=r"D:\pythonProject\learn\3b3e99cf4ca84c3782503d8817242de2.mp4",
help="测试用 m3u8 地址（默认使用公开测试流）",
)
parser.add_argument(
"--cache-dir",
default=r"C:\Users\lzw\AppData\Local\Temp\video2frame_test_mn26tcy_my",
help="缓存目录（默认自动创建临时目录）",
)
parser.add_argument(
"--start",
type=float,
default=0,
help="截取开始时间，单位：秒（默认 0）",
)
parser.add_argument(
"--end",
type=float,
default=10,
help="截取结束时间，单位：秒（默认 10）",
)
parser.add_argument(
"--fps",
type=float,
default=2.0,
help="抽帧帧率（默认 1.0）",
)
args = parser.parse_args()
cache_dir = args.cache_dir or tempfile.mkdtemp(prefix="video2frame_test_")
print(f"[INFO] 缓存目录: {cache_dir}")
v2f = Video2Frame(cache_dir)
# TEST 1: 基本抽帧
print(
f"\n[TEST 1] to_frames: url={args.url}, "
f"start={args.start}s, end={args.end}s, fps={args.fps}"
)
try:
frames = v2f.to_frames(
url=args.url,
start_utc=args.start,
end_utc=args.end,
fps=args.fps,
roi=None,
max_px_area=200_000,
)
print(f" 成功提取 {len(frames)} 帧")
if frames:
print(f" 首帧: {frames[0].shape}")
print(f" 末帧: {frames[-1].shape}")
except Exception as e:
print(f" 失败: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
# TEST 2: frames2content
print("\n[TEST 2] frames2content")
content = frames2content(frames)
print(f" content 列表长度: {len(content)}")
for item in content[:3]:
preview = (
item["text"][:60] + "..."
if item["type"] == "text"
else item["image_url"]["url"][:60] + "..."
)
print(f" - type={item['type']}: {preview}")
if len(content) > 3:
print(f" ... 还有 {len(content) - 3} 个元素")
print("\n[TEST 3] FootballReplayVideoEvent 进球识别（可选）")
try:
from ..util.football_replay_video_event_by_llm import FootballReplayVideoEvent
except ImportError:
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from util.football_replay_video_event_by_llm import FootballReplayVideoEvent
fbrv = FootballReplayVideoEvent(
base_url="http://192.168.1.59:11434",
model="qwen3.6:35b-a3b-q8_0",
temperature=0.7,
)
event_json = fbrv.video_event_for_contents(content, None)
print(f" 识别结果: {event_json}")
# # TEST 3: ROI + max_px_area
# print("\n[TEST 3] to_frames with roi + max_px_area")
# try:
# end_crop = min(args.start + 5, args.end)
# frames_cropped = v2f.to_frames(
# url=args.url,
# start_utc=args.start,
# end_utc=end_crop,
# fps=1.0,
# roi=None,
# max_px_area=200_000,
# )
# print(f" 成功提取 {len(frames_cropped)} 帧（带 ROI 裁剪和面积缩放）")
# except Exception as e:
# print(f" 跳过/失败: {e}")
#
# # TEST 4: SEI UTC 模式说明（仅演示，需真实含 SEI 的流才能运行）
# print("\n[TEST 4] SEI UTC 绝对时间戳模式（示例代码，需替换为含 SEI 的流）")
# print(
# " # 示例：假设视频第一帧 UTC 为 1715600000，提取 10 秒片段\n"
# " # frames = v2f.to_frames(url, start_utc=1715600000, end_utc=1715600010, fps=1)"
# )
#
# print("\n[INFO] 所有测试完成")
#!/usr/bin/env python3
"""
m3u8_to_mp4_sei.py
下载 m3u8 为 mp4，并提取视频中的 SEI UTC 信息。
依赖:
ffmpeg (需要添加到系统 PATH)
用法示例:
python m3u8_to_mp4_sei.py "http://example.com/playlist.m3u8"
python m3u8_to_mp4_sei.py "http://example.com/playlist.m3u8" -o myvideo.mp4 --json
python m3u8_to_mp4_sei.py "http://example.com/playlist.m3u8" -t 30 --json
"""
import argparse
import json
import os
import struct
import subprocess
import sys
from pathlib import Path
from typing import Dict, List, Optional
def check_ffmpeg() -> bool:
"""检查系统是否安装了 ffmpeg"""
try:
subprocess.run(["ffmpeg", "-version"], capture_output=True, check=True)
return True
except (FileNotFoundError, subprocess.CalledProcessError):
return False
def download_m3u8_to_mp4(url: str, output_path: str, duration: Optional[int] = None) -> None:
"""
使用 ffmpeg 将 m3u8 下载并封装为 mp4
:param duration: 若指定，则只下载前 N 秒
"""
cmd = [
"ffmpeg",
"-y",
"-fflags", "+discardcorrupt", # 容错：丢弃损坏包
"-i", url,
]
if duration is not None:
cmd.extend(["-t", str(duration)])
cmd.extend([
"-c", "copy",
"-bsf:a", "aac_adtstoasc",
"-movflags", "+faststart",
output_path,
])
print(f"[1/3] 正在下载并封装 mp4: {output_path}")
result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
if result.returncode != 0:
raise RuntimeError(f"ffmpeg 下载失败: {result.stderr}")
def extract_h264_es_from_mp4(mp4_path: str) -> bytes:
"""
使用 ffmpeg 从 mp4 中提取 H.264 Elementary Stream (AnnexB 格式)
返回完整的 ES 字节流
"""
cmd = [
"ffmpeg",
"-y",
"-i", mp4_path,
"-c:v", "copy",
"-an",
"-bsf:v", "h264_mp4toannexb",
"-f", "h264",
"pipe:1",
]
print("[2/3] 正在从 mp4 中提取 H.264 视频流...")
proc = subprocess.run(cmd, capture_output=True)
if proc.returncode != 0:
stderr = proc.stderr.decode("utf-8", errors="replace") if proc.stderr else ""
raise RuntimeError(f"ffmpeg 提取视频流失败: {stderr}")
return proc.stdout
def _sei_type_name(payload_type: int) -> str:
names = {
0: "buffering_period",
1: "pic_timing",
2: "pan_scan_rect",
3: "filler_payload",
4: "user_data_registered_itu_t_t35",
5: "user_data_unregistered",
6: "recovery_point",
7: "dec_ref_pic_marking_repetition",
8: "spare_pic",
9: "scene_info",
10: "sub_seq_info",
11: "sub_seq_layer_characteristics",
12: "sub_seq_characteristics",
13: "full_frame_freeze",
14: "full_frame_freeze_release",
15: "full_frame_snapshot",
16: "progressive_refinement_segment_start",
17: "progressive_refinement_segment_end",
18: "motion_constrained_slice_group_set",
19: "film_grain_characteristics",
20: "deblocking_filter_display_preference",
21: "stereo_video_info",
22: "post_filter_hint",
23: "tone_mapping_info",
24: "scalability_info",
25: "sub_pic_scalable_layer",
26: "non_required_layer_rep",
27: "priority_layer_info",
28: "layers_not_present",
29: "layer_dependency_change",
30: "scalable_nesting",
31: "base_layer_temporal_hrd",
32: "quality_layer_integrity_check",
33: "redundant_pic_property",
34: "tl0_dep_rep_index",
35: "tl_switching_point",
36: "parallel_decoding_info",
37: "mvc_scalable_nesting",
38: "view_scalability_info",
39: "multiview_scene_info",
40: "multiview_acquisition_info",
41: "non_equivalent_view_dependency",
42: "view_dependency_change",
43: "operation_points_not_present",
44: "base_view_temporal_hrd",
45: "frame_packing_arrangement",
46: "multiview_view_position",
47: "display_orientation",
48: "mvcd_scalable_nesting",
49: "mvcd_view_scalability_info",
50: "depth_representation_info",
51: "three_dimensional_reference_displays_info",
52: "depth_timing",
53: "depth_sampling_info",
54: "constrained_depth_parameter_set_identifier",
55: "green_metadata",
56: "mastering_display_colour_volume",
57: "colour_remapping_info",
58: "alternative_transfer_characteristics",
59: "alternative_depth_info",
}
return names.get(payload_type, f"unknown({payload_type})")
def parse_h264_sei(es_data: bytes) -> List[Dict]:
"""
从 H.264 ES 数据中解析 SEI NAL 单元
返回 SEI 列表，包含 payload_type, uuid, text/data
"""
sei_list = []
i = 0
def find_next_start(data: bytes, start: int) -> int:
j = start
while j + 3 < len(data):
if data[j:j + 4] == b"\x00\x00\x00\x01":
return j
if data[j:j + 3] == b"\x00\x00\x01":
return j
j += 1
return len(data)
while i < len(es_data):
if es_data[i:i + 4] == b"\x00\x00\x00\x01":
start_len = 4
elif es_data[i:i + 3] == b"\x00\x00\x01":
start_len = 3
else:
i += 1
continue
if i + start_len >= len(es_data):
break
nal_unit_type = es_data[i + start_len] & 0x1F
if nal_unit_type == 6: # SEI
payload_start = i + start_len + 1
next_start = find_next_start(es_data, payload_start)
sei_payload = es_data[payload_start:next_start]
k = 0
payload_type = 0
while k < len(sei_payload) and sei_payload[k] == 0xFF:
payload_type += 255
k += 1
if k < len(sei_payload):
payload_type += sei_payload[k]
k += 1
payload_size = 0
while k < len(sei_payload) and sei_payload[k] == 0xFF:
payload_size += 255
k += 1
if k < len(sei_payload):
payload_size += sei_payload[k]
k += 1
if k + payload_size <= len(sei_payload):
payload_data = sei_payload[k:k + payload_size]
sei_info = {
"payload_type": payload_type,
"payload_type_name": _sei_type_name(payload_type),
"payload_size": payload_size,
}
if payload_type == 5: # user_data_unregistered
if len(payload_data) >= 16:
uuid_bytes = payload_data[:16]
user_data = payload_data[16:]
while user_data and user_data[-1] == 0x00:
user_data = user_data[:-1]
if user_data and user_data[-1] == 0x80:
user_data = user_data[:-1]
sei_info["uuid_hex"] = uuid_bytes.hex()
sei_info["uuid_ascii"] = uuid_bytes.decode("ascii", errors="replace")
try:
text = user_data.decode("utf-8", errors="replace")
sei_info["text"] = text
try:
sei_info["json"] = json.loads(text)
except json.JSONDecodeError:
pass
except Exception:
sei_info["raw_hex"] = user_data.hex()
sei_list.append(sei_info)
i = next_start
else:
i += 1
return sei_list
def extract_utc_from_sei(sei_list: List[Dict]) -> List[Dict]:
"""从 SEI 列表中提取 UTC 信息"""
utc_records = []
seen = set()
for sei in sei_list:
json_data = sei.get("json")
if isinstance(json_data, dict) and "utc" in json_data:
# 去重：相同的 utc 只保留一次
key = json.dumps(json_data, sort_keys=True, ensure_ascii=False)
if key in seen:
continue
seen.add(key)
utc_records.append({
"utc": json_data["utc"],
"origin_utc": json_data.get("origin_utc"),
"origin_offset": json_data.get("origin_offset"),
"full_json": json_data,
})
return utc_records
def main():
parser = argparse.ArgumentParser(
description="将 m3u8 下载为 mp4 并提取 SEI UTC 信息",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
%(prog)s "http://example.com/playlist.m3u8"
%(prog)s "http://example.com/playlist.m3u8" -o myvideo.mp4 --json
%(prog)s "http://example.com/playlist.m3u8" -t 30 --json -o result.json
""",
)
parser.add_argument("url", help="m3u8 播放列表 URL")
parser.add_argument("-o", "--output", default=None, help="mp4 输出路径（默认根据 URL 自动命名）")
parser.add_argument("-t", "--duration", type=int, default=None, help="只下载前 N 秒（默认下载全部）")
parser.add_argument("--json", action="store_true", help="以 JSON 格式输出结果")
parser.add_argument("--no-download", action="store_true", help="如果本地已有 mp4，跳过下载")
parser.add_argument("--save-es", metavar="FILE", help="保存提取的 H.264 ES 流到指定文件（调试用）")
args = parser.parse_args()
# 强制 stdout UTF-8
import io
try:
if getattr(sys.stdout, "encoding", None) != "utf-8":
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
elif hasattr(sys.stdout, "buffer"):
sys.stdout = io.TextIOWrapper(
sys.stdout.buffer, encoding="utf-8", line_buffering=True
)
except Exception:
pass
if not check_ffmpeg():
print("错误: 未检测到 ffmpeg，请先安装并添加到 PATH。", file=sys.stderr)
print("下载地址: https://ffmpeg.org/download.html", file=sys.stderr)
sys.exit(1)
# 确定 mp4 输出路径
if args.output:
mp4_path = Path(args.output)
else:
base = args.url.split("?")[0].rsplit("/", 1)[-1]
if not base or "." not in base:
base = "output.mp4"
else:
base = base.rsplit(".", 1)[0] + ".mp4"
mp4_path = Path(base)
es_data = b""
try:
# 1. 下载/封装 mp4
if not args.no_download or not mp4_path.exists():
download_m3u8_to_mp4(args.url, str(mp4_path), duration=args.duration)
print(f" mp4 已保存: {mp4_path.absolute()}")
else:
print(f"[1/3] 使用本地 mp4: {mp4_path.absolute()}")
# 2. 提取 H.264 ES
es_data = extract_h264_es_from_mp4(str(mp4_path))
print(f" 提取到 {len(es_data)} 字节的 H.264 ES 数据")
if args.save_es:
es_path = Path(args.save_es)
es_path.write_bytes(es_data)
print(f" H.264 ES 已保存: {es_path.absolute()}")
# 3. 解析 SEI
sei_list = parse_h264_sei(es_data)
print(f"[3/3] 解析完成，发现 {len(sei_list)} 条 SEI 消息")
# 4. 提取 UTC
utc_records = extract_utc_from_sei(sei_list)
result = {
"m3u8_url": args.url,
"mp4_path": str(mp4_path.absolute()),
"sei_count": len(sei_list),
"utc_records": utc_records,
"sei_items": sei_list[:20],
}
# 输出
json_str = json.dumps(result, ensure_ascii=False, indent=2)
if args.json and args.output and args.output.endswith(".json"):
# 如果 --output 以 .json 结尾，且 --json，则保存为 JSON 文件
with open(args.output, "w", encoding="utf-8-sig") as f:
f.write(json_str)
f.write("\n")
print(f"JSON 结果已保存到: {args.output}")
elif args.json:
print(json_str)
else:
print(f"\n{'='*50}")
print("SEI UTC 信息提取结果")
print(f"{'='*50}")
print(f"mp4 文件 : {result['mp4_path']}")
print(f"SEI 消息总数 : {result['sei_count']}")
print(f"UTC 记录数 : {len(utc_records)}")
if utc_records:
print(f"\nUTC 详细信息:")
for idx, rec in enumerate(utc_records, 1):
print(f" [{idx}] utc={rec['utc']}, origin_utc={rec.get('origin_utc')}, origin_offset={rec.get('origin_offset')}")
else:
print("\n未从 SEI 中提取到 UTC 信息。")
print(f"{'='*50}")
except Exception as e:
print(f"错误: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
# python mp4_sei_extractor.py "3b3e99cf4ca84c3782503d8817242de2.mp4" --json -o sei_result.json
\ No newline at end of file
import json
import os.path
from pathlib import Path
from datetime import timedelta
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
try:
from .llm_video_content import contents as video_contents
from .llm_image import Video2Frame
except:
from llm_video_content import contents as video_contents
from llm_image import Video2Frame
req_prompt = """
# Role
... ... @@ -70,7 +71,7 @@ req_prompt = """
class FootballReplayMatchLive:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key'):
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):
self.base_url = base_url
self.model = model
self.temperature = temperature
... ... @@ -81,13 +82,16 @@ class FootballReplayMatchLive:
# 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}})
self.cache_dir = cache_dir
self.video2frame = Video2Frame(cache_dir=cache_dir, save_frames_enable=save_frames_enable)
def _match_once(self, replay_video_contents: list, live_videos: list[dict], cache_path=None, record: list = None):
def _match_once(self, replay_video_contents: list, live_videos: list[dict], record: list = None):
if len(live_videos) == 0:
return None
elif len(live_videos) == 1:
live_video_path = live_videos[0].get("video_path", None)
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 = {
... ... @@ -107,16 +111,24 @@ class FootballReplayMatchLive:
live_map = {}
live_records = {}
for live_video in live_videos:
live_video_path = live_video.get("video_path", None)
live_video_path = live_video.get("url", None)
live_video_id = live_video.get("video_id", os.path.basename(live_video_path))
event_utc = live_video.get("event_utc", None)
asr_text = live_video.get("asr_text", '')
live_video_contents = live_video.get("llm_contents", None)
if live_video_contents is None:
live_video_contents = video_contents(live_video_path,
prompt_start=f"### 候选片段 video_id: {live_video_id} ###",
prompt_end=f"\n该片段解说内容: {asr_text}\n",
video_name=os.path.basename(live_video_path),
fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
event_start = event_utc - timedelta(seconds=10)
event_end = event_utc + timedelta(seconds=5)
live_video_contents = self.video2frame.to_llm_contents( live_video_path,
cache=self.cache_dir,
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start=f"### 候选片段 video_id: {live_video_id} ###",
prompt_end=f"\n该片段解说内容: {asr_text}\n"
)
live_map[live_video_id] = {
"video_id": live_video_id,
"video_path": live_video_path,
... ... @@ -193,17 +205,29 @@ class FootballReplayMatchLive:
else:
return None
def match_batch(self, replay_video: dict, live_videos: list[dict], max_parallel: int = 3, cache_path=None):
def match_batch(self, replay_video: dict, live_videos: list[dict], max_parallel: int = 3, cache_dir=None):
cache_path = os.path.join(cache_dir, 'match_live.json')
if cache_path is not None and os.path.exists(cache_path):
try:
with open(cache_path, 'r', encoding='utf-8') as f:
return json.loads(f.read()).get("result", None)
except:
os.remove(cache_path)
replay_video_contents = video_contents(replay_video["video_path"], "\n【回放片段信息】\n",
prompt_end=f"\n回放解说内容：{replay_video['asr_text']}\n",
video_name=os.path.basename(replay_video["video_path"]),
fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
replay_video_path = replay_video.get("url", None)
event_start = replay_video.get("start_utc", None)
event_end = replay_video.get("end_utc", None)
replay_video_contents = self.video2frame.to_llm_contents(replay_video_path,
cache=self.cache_dir,
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start="\n【回放片段信息】\n",
prompt_end=f"\n回放解说内容：无\n"
)
live_record = []
result = self._match_batch(replay_video_contents, live_videos, max_parallel, cache_path, live_record)
if result is not None:
... ...
... ... @@ -6,9 +6,10 @@ from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
try:
from .llm_video_content import contents as video_contents
from .llm_image import Video2Frame
except:
from llm_video_content import contents as video_contents
from llm_image import Video2Frame
req_prompt = """
### 角色设定
... ... @@ -65,7 +66,7 @@ req_prompt = """
class FootballReplayVideoEvent:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key'):
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):
self.base_url = base_url
self.model = model
self.temperature = temperature
... ... @@ -76,14 +77,35 @@ class FootballReplayVideoEvent:
# 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}})
def video_event(self, video_path: str, asr_text: str = '无', cache_path=None):
self.cache_dir = cache_dir
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):
replay_video_path = replay_pack.get("url", None)
cache_path = os.path.join(cache_dir, 'video_event.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())
contents = video_contents(video_path, None, video_name=os.path.basename(video_path),
fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
# contents = video_contents(video_path, None, video_name=os.path.basename(video_path),
# fps=2, max_frames=999, sampling_mode="head", max_short_edge=480)
event_start = replay_pack.get("start_utc", None)
event_end = replay_pack.get("end_utc", None)
contents = self.video2frame.to_llm_contents(replay_video_path,
cache=self.cache_dir,
fps=2,
start=event_start,
end=event_end,
roi=None,
max_px_area=400_000,
prompt_start="\n【回放片段信息】\n",
prompt_end=f"\n回放解说内容：{asr_text}\n"
)
system_message = SystemMessage(content=req_prompt)
video_message = HumanMessage(content=contents)
asr_message = HumanMessage(content=f"解说内容：{asr_text}")
... ... @@ -98,20 +120,23 @@ class FootballReplayVideoEvent:
f.write(result)
return result_json
def video_event_for_contents(self, contents: list, asr_text: str = '无', cache_path=None):
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())
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
try:
result_json = json.loads(result)
except json.JSONDecodeError:
result_json = json.loads(result.replace("```json", "").replace("```", ""))
if cache_path is not None:
os.makedirs(Path(cache_path).parent, exist_ok=True)
with open(cache_path, 'w', encoding='utf-8') as f:
f.write(result)
return result_json
if __name__ == "__main__":
from aabd.base.patched_logging import init_logging
import os
os.environ["APP_LOG_TYPE"] = "console"
init_logging()
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
cache_dir="/root/lzw/tmp_0518_replay_cache",
save_frames_enable=True
)
replay_pack = {"url": "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8",
"start_utc": 0,
"end_utc": 999999999,
"asr_text": '无'
}
replay_goals = fbrv.video_event(replay_pack, cache_path="/root/lzw/tmp_0518_replay_cache")
print(replay_goals)
\ No newline at end of file
... ...
... ... @@ -153,11 +153,13 @@ if __name__ == '__main__':
import os
os.environ["APP_LOG_TYPE"] = "console"
init_logging()
url = "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8"
# url = rf"R:\wdx\202604\20260420_download_football_video\finished\69dd5845dd0412067b8d5587-auto-1776074760653\live\videos\00-14-09-052.mp4"
vf = Video2Frame(rf'D:\Code\migu\aigc-embedding-service\src\football_replay_match\core', save_frames_enable=True)
a = vf.to_llm_contents(url, fps=2, start=1777444502543, end=1777444531063, max_px_area=1920 * 1080 // 6)
# url = "http://video.mam.miguvideo.com/mnt6/fastclip3/wsc/2026/04/29/df760b8d15394cbc9ed0d0a4a9c4b786_1080PS/29133605/vodtmp/3b3e99cf4ca84c3782503d8817242de2.m3u8"
url = rf"/root/lzw/finished/69dd5845dd0412067b8d5587-auto-1776074760653/live/videos/00-14-09-052.mp4"
caceh_dir = r"/root/lzw/aigc-embedding-service/src/football_replay_match/core"
vf = Video2Frame(caceh_dir, save_frames_enable=True)
# vf = Video2Frame(rf'D:\Code\migu\aigc-embedding-service\src\football_replay_match\core', save_frames_enable=True)
# a = vf.to_llm_contents(url, fps=2, start=1777444502543, end=1777444531063, max_px_area=1920 * 1080 // 6)
a = vf.to_llm_contents(url, fps=2, start=0, end=10, max_px_area=1920 * 1080 // 6)
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
... ...
import base64
import tempfile
from pathlib import Path
from typing import Union
import cv2
import httpx
def _resize_frame(frame, max_short_edge: int):
"""按比例缩放帧，确保最短边不超过指定值"""
h, w = frame.shape[:2]
short_edge = min(h, w)
if short_edge > max_short_edge:
scale = max_short_edge / short_edge
new_w = int(w * scale)
new_h = int(h * scale)
frame = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_AREA)
return frame
def contents(video_source: Union[str, Path], prompt_start: str = None, prompt_end=None, video_name: str = None,
fps: float = 1.0,
max_frames: int = 10,
max_short_edge: int = 768,
sampling_mode: str = "uniform") -> list[str]:
"""
从视频中提取帧并构建 LLM 消息内容。
Args:
video_source: 视频源，可以是文件路径或 URL。
prompt_start: 提示词的开始部分。
prompt_end: 提示词的结束部分。
video_name: 视频名称。
fps: 提取帧的帧率。
max_frames: 最大帧数。
max_short_edge: 最大短边长度。
sampling_mode: 当帧数超过 max_frames 时的采样策略。
- "uniform": 均匀采样（默认）
- "head": 保留前面的帧，抛弃后面的帧
"""
source_str = str(video_source)
temp_file = None
try:
# 如果是 URL，先下载到临时文件
if source_str.startswith(("http://", "https://")):
resp = httpx.get(source_str, timeout=60.0)
resp.raise_for_status()
temp_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
temp_file.write(resp.content)
temp_file.close()
video_path = temp_file.name
else:
video_path = str(video_source)
if not Path(video_path).exists():
raise FileNotFoundError(f"视频文件不存在: {video_path}")
# 打开视频
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
raise ValueError(f"无法打开视频: {video_source}")
# 获取视频原始帧率
video_fps = cap.get(cv2.CAP_PROP_FPS)
# 计算帧间隔
frame_interval = int(video_fps / fps) if fps > 0 else int(video_fps)
if frame_interval < 1:
frame_interval = 1
frames_base64 = []
frame_count = 0
while True:
ret, frame = cap.read()
if not ret:
break
# 按指定间隔提取帧
if frame_count % frame_interval == 0:
# 缩放帧以控制最短边长度
frame = _resize_frame(frame, max_short_edge)
# 编码为 JPEG 并转 base64
_, buffer = cv2.imencode('.jpg', frame)
frame_base64 = base64.b64encode(buffer).decode('utf-8')
frames_base64.append(frame_base64)
frame_count += 1
cap.release()
if len(frames_base64) > max_frames:
import math
step = len(frames_base64) / max_frames
sampled = [frames_base64[min(int(i * step), len(frames_base64) - 1)] for i in range(max_frames)]
frames_base64 = sampled
# 构建消息内容：提示词 + 所有帧图片
video_prompt = (
f"以下是从视频({video_name})中按时间顺序提取的 {len(frames_base64)} 帧画面，视频原始帧率为 {video_fps:.2f} fps，"
f"抽帧间隔为 {frame_interval} 帧（约每 {frame_interval / video_fps:.2f} 秒一帧），请将它们视为一个连续的视频进行分析。"
)
content = []
if prompt_start is not None:
content.append({"type": "text", "text": prompt_start})
content.append({"type": "text", "text": video_prompt})
for frame_base64 in frames_base64:
content.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{frame_base64}"
}
})
if prompt_end is not None:
content.append({"type": "text", "text": prompt_end})
return content
finally:
# 清理临时文件
if temp_file and Path(temp_file.name).exists():
Path(temp_file.name).unlink()
from football_replay_match_live import FootballReplayMatchLive
from qwen_asr_util import QwenAsr
from football_replay_video_event_by_llm import FootballReplayVideoEvent
import os
import json
def batch_match():
replay_match_live = FootballReplayMatchLive(base_url="http://192.168.1.59:11434/v1", model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf")
qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B")
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1", model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf", temperature=0.7)
videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
for video_name in sorted(os.listdir(videos_dir)):
live_dir = os.path.join(videos_dir, video_name, 'live')
live_video_dir = os.path.join(live_dir, 'videos')
live_asr_dir = os.path.join(live_dir, 'asr')
live_packs = []
for live_video_name in sorted(os.listdir(live_video_dir)):
live_video_path = os.path.join(live_video_dir, live_video_name)
live_asr_path = os.path.join(live_asr_dir, f"{live_video_name}.txt")
# live_asr_text = qwen_asr.asr(live_video_path, cache_path=live_asr_path)
live_asr_text = ''
live_packs.append({
"video_id": os.path.basename(live_video_path),
"video_path": live_video_path,
"asr_text": live_asr_text
})
replays_dir = os.path.join(videos_dir, video_name, 'replays')
replay_videos_dir = os.path.join(replays_dir, 'videos')
replay_goals_dir = os.path.join(replays_dir, 'goals')
replay_asr_dir = os.path.join(replays_dir, 'asr')
replay_matches_dir = os.path.join(replays_dir, 'matches')
for replay_video_name in sorted(os.listdir(replay_videos_dir)):
replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.txt")
replay_match_path = os.path.join(replay_matches_dir, f"{replay_video_name}.json")
# replay_asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
replay_asr_text = ''
replay_goals = fbrv.video_event(replay_video_path, cache_path=replay_goals_path)
if replay_goals['event_name'] == '无进球':
continue
replay_pack = {"video_path": replay_video_path, "asr_text": replay_asr_text}
try:
result = replay_match_live.match_batch(replay_pack, live_packs, max_parallel=2,
cache_path=replay_match_path)
print(replay_video_path)
print(result)
except Exception as e:
print(f"Error processing {replay_video_path}: {e}")
print('-' * 20)
if __name__ == '__main__':
batch_match()
import os
from io import BytesIO
import re
import requests
import base64
import tempfile
from pathlib import Path
from typing import Union
import httpx
import subprocess
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
class QwenAsr:
def __init__(self, base_url: str, model: str, temperature: float = 0.0, api_key: str = 'no_key'):
self.base_url = base_url
self.model = model
self.temperature = temperature
self.api_key = api_key
# self.asr_model = ChatOpenAI(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B", temperature=0.7,
# api_key='no_key')
self.asr_model = ChatOpenAI(base_url=self.base_url, model=self.model, temperature=self.temperature,
api_key=self.api_key)
@staticmethod
def _audio_content(source: Union[str, Path]) -> list:
"""
读取音频/视频文件并构建 LLM 消息内容。
如果 source 是视频，会先提取音频；如果是 HTTP 音频 URL，直接引用 URL。
Args:
source: 音频/视频文件路径或 URL
Returns:
list: LLM 消息内容列表
"""
source_str = str(source)
audio_exts = {"wav", "mp3", "m4a", "flac", "ogg", "webm", "aac", "wma"}
video_exts = {"mp4", "avi", "mkv", "mov", "wmv", "flv", "webm", "mpeg", "mpg", "ts", "3gp", "m4v"}
temp_files = []
try:
content = []
is_url = source_str.startswith(("http://", "https://"))
ext = Path(source_str).suffix.lstrip(".").lower()
# HTTP 音频 URL：直接引用，无需下载
if is_url and ext in audio_exts:
# audio_prompt = f"以下是音频文件({audio_name})的内容，请进行分析。"
# content = [{"type": "text", "text": audio_prompt}]
# if prompt is not None:
# content.append({"type": "text", "text": prompt})
content.append({
"type": "audio_url",
"audio_url": {
"url": source_str
}
})
return content
# 本地文件或需要下载的 URL（视频或其他媒体）
if is_url:
resp = httpx.get(source_str, timeout=60.0)
resp.raise_for_status()
suffix = Path(source_str).suffix or ".mp4"
temp_file = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
temp_file.write(resp.content)
temp_file.close()
temp_files.append(temp_file.name)
media_path = temp_file.name
else:
media_path = str(source)
if not Path(media_path).exists():
raise FileNotFoundError(f"文件不存在: {media_path}")
media_ext = Path(media_path).suffix.lstrip(".").lower()
# 如果是视频，提取音频为 wav
if media_ext in video_exts or (is_url and ext not in audio_exts):
temp_audio = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
temp_audio.close()
temp_files.append(temp_audio.name)
cmd = [
"ffmpeg",
"-y",
"-i", media_path,
"-vn",
"-acodec", "pcm_s16le",
"-ar", "16000",
"-ac", "1",
"-hide_banner",
"-loglevel", "error",
temp_audio.name
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"ffmpeg 提取音频失败: {result.stderr.strip()}")
audio_path = temp_audio.name
media_ext = "wav"
else:
audio_path = media_path
if media_ext not in audio_exts:
media_ext = "wav"
# 读取音频并 base64 编码
with open(audio_path, "rb") as f:
audio_bytes = f.read()
audio_base64 = base64.b64encode(audio_bytes).decode("utf-8")
# 构建消息内容
# audio_prompt = f"以下是音频文件({audio_name})的内容，请进行分析。"
# content = [{"type": "text", "text": audio_prompt}]
# if prompt is not None:
# content.append({"type": "text", "text": prompt})
content.append({
"type": "audio_url",
"audio_url": {
"url": f"data:audio/{media_ext};base64,{audio_base64}"
}
})
return content
finally:
# 清理所有临时文件
for tf in temp_files:
p = Path(tf)
if p.exists():
p.unlink()
@staticmethod
def _extract_asr_text(text: str) -> str:
"""
从ASR响应文本中提取<asr_text>标签内容
Args:
text: 原始响应文本，如 'language Chinese<asr_text>放大一点一倍。'
Returns:
str: 提取的转录文本，如 '放大一点一倍。'
"""
# 尝试匹配 <asr_text>...</asr_text>
match = re.search(r'<asr_text>(.*?)(?:</asr_text>|$)', text, re.DOTALL)
if match:
return match.group(1).strip()
return text
def asr(self, source, cache_path:str=None):
if cache_path is not None and os.path.exists(cache_path):
with open(cache_path, 'r', encoding='utf-8') as f:
return f.read()
contents = self._audio_content(source)
message = HumanMessage(content=contents)
result = self.asr_model.invoke([message])
asr_text = self._extract_asr_text(result.content)
if cache_path is not None:
os.makedirs(Path(cache_path).parent, exist_ok=True)
with open(cache_path, 'w', encoding='utf-8') as f:
f.write(asr_text)
return asr_text
\ No newline at end of file
import json
import os
from football_replay_video_event_by_llm import FootballReplayVideoEvent
from qwen_asr_util import QwenAsr
def batch_with_asr():
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1", model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf", temperature=0.7)
qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B",temperature=0.7,
api_key='no_key')
videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
for video_name in sorted(os.listdir(videos_dir)):
replays_dir = os.path.join(videos_dir, video_name, 'replays')
replay_videos_dir = os.path.join(replays_dir, 'videos')
replay_goals_dir = os.path.join(replays_dir, 'goals')
replay_asr_dir = os.path.join(replays_dir, 'asr')
for replay_video_name in sorted(os.listdir(replay_videos_dir)):
replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.json")
if os.path.exists(replay_goals_path):
print("skip", replay_video_path)
continue
asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
event_json = fbrv.video_event(replay_video_name, asr_text)
print(event_json)
def batch_without_asr():
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1", model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf", temperature=0.7)
# qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B", temperature=0.7,
# api_key='no_key')
videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
for video_name in sorted(os.listdir(videos_dir)):
replays_dir = os.path.join(videos_dir, video_name, 'replays')
replay_videos_dir = os.path.join(replays_dir, 'videos')
replay_goals_dir = os.path.join(replays_dir, 'goals')
replay_asr_dir = os.path.join(replays_dir, 'asr')
for replay_video_name in sorted(os.listdir(replay_videos_dir)):
replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.json")
if os.path.exists(replay_goals_path):
print("skip", replay_video_path)
continue
# asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
event_json = fbrv.video_event(replay_video_path, None, cache_path=replay_goals_path)
print(replay_video_name)
print(event_json)
def one_video_test(video_path):
fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434/v1", model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf", temperature=0.7)
event_json = fbrv.video_event(video_path, None)
print(event_json)
if __name__ == '__main__':
# one_video_test(rf"D:\Code\py260417\src\llm_demo\lc_t\ftb\replay_1.mp4")
batch_without_asr()
\ No newline at end of file
import os
import json
from pathlib import Path
from aabd.base.time_util import vms2str_auto
from langchain_openai import ChatOpenAI
class ClipByEvent:
def __init__(self, base_url, model, temperature=0.0, api_key='no_key'):
self.base_url = base_url
self.model = model
self.temperature = temperature
self.api_key = api_key
self.clip_before = 10 * 1000
self.clip_after = 5 * 1000
# self.model = ChatOllama(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# keep_alive=-1, reasoning=False)
# 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)
def get_video_match_time(self, video_path, video_time):
import cv2
import base64
from langchain_core.messages import HumanMessage
# 从视频中精确提取指定时间的帧画面
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
raise ValueError(f"无法打开视频文件: {video_path}")
fps = cap.get(cv2.CAP_PROP_FPS)
# video_time 单位为毫秒，转换为帧号
frame_number = int((video_time / 1000.0) * fps)
cap.set(cv2.CAP_PROP_POS_FRAMES, frame_number)
ret, frame = cap.read()
cap.release()
if not ret:
raise ValueError(f"无法从视频 {video_path} 中提取时间 {video_time}ms 的帧画面")
# 将帧图像编码为base64
_, buffer = cv2.imencode('.jpg', frame)
image_base64 = base64.b64encode(buffer).decode('utf-8')
# 使用大模型分析比赛画面中的时间
message = HumanMessage(
content=[
{
"type": "text",
"text": "请仔细观察这张比赛画面截图，找出画面中显示的比赛计时器或时间信息，并以'MM:SS'格式返回比赛时间。如果无法识别，请返回'unknown'。"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
)
response = self.model.invoke([message])
match_time_text = response.content.strip()
if match_time_text == 'unknown':
raise ValueError(f"无法识别视频 {video_path} 中时间 {video_time}ms 的比赛时间")
return match_time_text
def clip_video(self, video_path, clip_time, out_path):
# 精确截取视频指定视频段,从新编码
# 毫秒
start_time = clip_time - self.clip_before
end_time = clip_time + self.clip_after
start_time_s = start_time / 1000.0
duration_s = (end_time - start_time) / 1000.0
import subprocess
import os
os.makedirs(os.path.dirname(out_path), exist_ok=True)
cmd = [
'ffmpeg',
'-i', video_path,
'-ss', str(start_time_s),
'-t', str(duration_s),
'-c:v', 'libx264',
'-c:a', 'aac',
'-y',
out_path
]
subprocess.run(cmd, check=True)
def clip_by_event(self, video_dir, json_dir, out_dir):
for video_name in sorted(os.listdir(video_dir)):
video_path = os.path.join(video_dir, video_name)
f_name_0 = Path(video_name).stem
json_path = os.path.join(json_dir, f"{f_name_0}.json")
with open(json_path, 'r', encoding='utf-8') as f:
data = json.load(f)
events = data['dataObject']['data']['events']
event0 = data['dataObject']['data']['events'][0]
event0_sei_utc = event0['seiUtc']
event0_text_time = event0['eventTimeText']
mm, ss = map(int, event0_text_time.split(':'))
event0_time_seconds = mm * 60 + ss
event0_time_ms = event0_time_seconds * 1000
match_start_sei_utc = event0_sei_utc - event0_time_ms
take_frame_time = 30 * 60 * 1000
match_text_time = self.get_video_match_time(video_path, take_frame_time)
mm, ss = map(int, match_text_time.split(':'))
match_time_seconds = mm * 60 + ss
match_time_ms = match_time_seconds * 1000
match_start_video_time = take_frame_time - match_time_ms
video_start_sei_utc = match_start_sei_utc - match_start_video_time
start_utc = video_start_sei_utc
os.makedirs(os.path.join(out_dir, f_name_0), exist_ok=True)
# 样例
# 41 传球 曼联,约罗 00:36 1776106869512 638160 00:10:38.160
# 41 传球 曼联,马兹拉维 00:39 1776106872512 641160 00:10:41.160
event_list = []
for event in events:
event_enum = event['eventType']
eventTitle = event['eventTitle']
event_array = eventTitle.split(' ')
event_2 = event_array[-1]
event_1 = ','.join(event_array[:-1])
eventTimeText = event['eventTimeText']
seiUtc = event['seiUtc']
event_list.append([event_enum, event_2, event_1, eventTimeText, seiUtc, seiUtc - start_utc,
vms2str_auto(seiUtc - start_utc)])
# out_f.write(
# f'{event_enum}\t{event_2}\t{event_1}\t{eventTimeText}\t{seiUtc}\t{seiUtc - start_utc}\t{vms2str_auto(seiUtc - start_utc)}\n')
with open(os.path.join(out_dir, f_name_0, f"sport_events.txt"), 'w', encoding='utf-8') as out_f:
for event in event_list:
out_f.write(
f'{event[0]}\t{event[1]}\t{event[2]}\t{event[3]}\t{event[4]}\t{event[5]}\t{event[6]}\n')
for event in event_list:
if event[1] == '进球':
out_path = os.path.join(out_dir, f_name_0, 'videos',
f"{event[6].replace(':', '-').replace('.', '-')}.mp4")
if not os.path.exists(out_path):
print(f"Clipping {video_name} {event[6]} to {out_path}")
self.clip_video(video_path, event[5], out_path)
else:
print(f"Clip skip {video_name} {event[6]} as {out_path} already exists")
if __name__ == '__main__':
cbe = ClipByEvent(base_url="http://192.168.1.215:11434", model="qwen3.5:9b-q8_0", temperature=0.7)
cbe.clip_by_event(video_dir=rf"L:\wdx\202604\20260420_download_football_video\downloads",
json_dir=rf"D:\Code\py260417\src\llm_demo\foot_event_clips\format_json",
out_dir=rf"R:\wdx\202604\20260420_download_football_video\finished1")