目录good-badminton 推荐看起来还行racquet-sports-analyzerhuji推理脚本羽球时刻 推荐Badminton-Highlight-Extractiongood-badminton 推荐看起来还行https://github.com/qwpyyx/Good-Badmintonracquet-sports-analyzerBot-Derpy/racquet-sports-analyzerhujihttps://github.com/hhoao/huji https://github.com/hhoao/huji-algorithmconda create -p E:\soft\envs\py312 python3.12 -y conda activate E:\soft\envs\py312 pip install -r requirements.txt推理脚本python main.py --video-path C:\Users\ChanJing-01\Videos\yumao\yumao.mp4 --sport badmintonimport argparse import importlib.util import json import os import sys from pathlib import Path from typing import Any import os import re import sys import shutil from datetime import datetime # 修复 auto_clipper 的排序问题 def patch_auto_clipper(): try: from src.main.core import auto_clipper # 保存原始方法 original_handle_video auto_clipper.AutoClipper.handle_video def patched_handle_video(self, *args, **kwargs): result original_handle_video(self, *args, **kwargs) # 修复 all_ball_video_list 排序 if hasattr(self, all_ball_video_list): def extract_number(filepath): filename os.path.basename(filepath) name_without_ext os.path.splitext(filename)[0] numbers re.findall(r\d, name_without_ext) return int(numbers[0]) if numbers else 0 self.all_ball_video_list.sort(keyextract_number) return result # 替换方法 auto_clipper.AutoClipper.handle_video patched_handle_video print(✅ AutoClipper 排序修复已应用) except Exception as e: print(f⚠️ 无法应用修复: {e}) def _venv_python() - str | None: venv_dir Path(__file__).resolve().parent / .venv if sys.platform win32: candidate venv_dir / Scripts / python.exe else: candidate venv_dir / bin / python return str(candidate) if candidate.is_file() else None def _check_runtime_deps() - None: if importlib.util.find_spec(ruamel.yaml) is None: print(未找到 Python 依赖请先安装并激活虚拟环境, filesys.stderr) if sys.platform win32: print( .\\setup.ps1, filesys.stderr) print( .venv\\Scripts\\activate, filesys.stderr) else: print( ./setup.sh, filesys.stderr) print( source .venv/bin/activate, filesys.stderr) venv_python _venv_python() if venv_python: print(f 或直接: {venv_python} main.py ..., filesys.stderr) sys.exit(1) _check_runtime_deps() from src import CONFIG_PATH from src.main.config.config import Config, load_config from src.main.constant.autoclip_constant import BadmintonAutoClipConfig, MatchType from src.main.constant.common_constant import JobType from src.main.core.badminton_auto_clipper import BadmintonAutoClipper from src.main.core.pingpong_auto_clipper import PingPongAutoClipper from src.main.logger import LOG from src.main.service.large_model_service import LargeModelService from src.main.utils import path_utils def _build_parser() - argparse.ArgumentParser: parser argparse.ArgumentParser(description乒乓球、羽毛球比赛视频自动剪辑, formatter_classargparse.RawDescriptionHelpFormatter, epilog 示例: python main.py --video-path videos/demo.mp4 --sport ping_pong python main.py --video-path videos/demo.mp4 --sport badminton --match-type doubles python main.py --video-path videos/demo.mp4 --sport badminton --export-rounds python main.py --serve python main.py --train python main.py , ) parser.add_argument(--config, defaultCONFIG_PATH, helpf配置目录含 application.yml默认 {CONFIG_PATH}, ) parser.add_argument(--video-path, -v, metavarPATH, defaultrC:\Users\ChanJing-01\Videos\yumao\yumao1.mp4, help本地视频文件路径) parser.add_argument(--sport, defaultbadminton, choices[ping_pong, badminton], help运动类型clip 模式必填, ) parser.add_argument(--match-type, choices[singles, doubles], defaultsingles, help羽毛球比赛类型默认 singles, ) parser.add_argument(--output-dir, -o, metavarDIR, help剪辑输出目录) parser.add_argument(--serve, actionstore_true, help启动 Kafka HTTP 服务) parser.add_argument(--train, actionstore_true, help训练模型) # 新增回合导出相关参数 parser.add_argument(--export-rounds, actionstore_true, defaultTrue, help将每个回合单独导出为独立视频文件) parser.add_argument(--rounds-dir, metavarDIR, help回合视频导出目录默认在输出目录下创建 rounds_时间戳 子目录) parser.add_argument(--export-format, choices[mp4, avi, mov], defaultmp4, help导出视频格式默认 mp4) # 新增低回合阈值参数 parser.add_argument(--low-rounds-threshold, typeint, default5, help低回合阈值默认5个回合少于该值将被分类到低回合文件夹) # 新增视频优化参数 parser.add_argument(--optimize-video, actionstore_true, defaultTrue, help优化导出视频确保播放流畅默认开启) parser.add_argument(--target-fps, typefloat, defaultNone, help目标帧率不指定则保持原始帧率) parser.add_argument(--video-quality, choices[low, medium, high], defaulthigh, help视频质量默认 high) return parser def _resolve_mode(args: argparse.Namespace) - str: modes [bool(args.video_path), args.serve, args.train] if sum(modes) 1: LOG.error(不能同时指定 --video-path、--serve、--train) sys.exit(1) if args.video_path: return clip if args.serve: return serve if args.train: return train return config def _build_auto_clip_config(args: argparse.Namespace) - dict[str, Any] | None: if args.sport ! badminton: return None match_type (MatchType.DOUBLES_MATCH if args.match_type doubles else MatchType.SINGLES_MATCH) return json.loads(BadmintonAutoClipConfig(match_typematch_type).model_dump_json()) def optimize_video_export(input_path: str, output_path: str, target_fps: float None, quality: str high) - bool: try: import cv2 cap cv2.VideoCapture(input_path) if not cap.isOpened(): LOG.error(f无法打开视频: {input_path}) return False # 获取原始参数 original_fps cap.get(cv2.CAP_PROP_FPS) width int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) total_frames int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # 验证参数 if width 0 or height 0: LOG.error(f无效的视频尺寸: {width}x{height}) cap.release() return False # 确定最终帧率 fps target_fps if target_fps and target_fps 0 else original_fps if fps 0 or fps 120: fps 30 # 默认30fps # 根据质量设置编码参数 quality_settings {low: {fourcc: cv2.VideoWriter_fourcc(*mp4v), fps: min(fps, 25), scale: 0.5 # 缩放比例 }, medium: {fourcc: cv2.VideoWriter_fourcc(*mp4v), fps: fps, scale: 0.75}, high: {fourcc: cv2.VideoWriter_fourcc(*mp4v), fps: fps, scale: 1.0}} settings quality_settings.get(quality, quality_settings[high]) # 应用缩放 if settings[scale] ! 1.0: width int(width * settings[scale]) height int(height * settings[scale]) # 确保宽高为偶数编码要求 width width if width % 2 0 else width 1 height height if height % 2 0 else height 1 LOG.info(f导出参数: {width}x{height}, {settings[fps]:.2f}fps, {total_frames}帧, 质量:{quality}) # 创建写入器 out cv2.VideoWriter(str(output_path), settings[fourcc], settings[fps], (width, height)) if not out.isOpened(): LOG.error(无法创建视频写入器) cap.release() return False # 逐帧处理 frame_count 0 success_count 0 while True: ret, frame cap.read() if not ret: break # 如果需要缩放 if settings[scale] ! 1.0: frame cv2.resize(frame, (width, height), interpolationcv2.INTER_LINEAR) # 写入帧 out.write(frame) frame_count 1 success_count 1 # 每100帧显示进度 if frame_count % 100 0: LOG.debug(f处理进度: {frame_count}/{total_frames}) cap.release() out.release() # 验证输出 if success_count 0: LOG.error(没有帧被写入) return False # 验证输出文件是否存在且有效 if os.path.exists(output_path) and os.path.getsize(output_path) 0: LOG.info(f✅ 成功导出 {success_count} 帧到: {output_path}) return True else: LOG.error(f输出文件无效: {output_path}) return False except ImportError: LOG.warning(OpenCV未安装请运行: pip install opencv-python) # 回退到简单复制 try: shutil.copy2(input_path, output_path) LOG.info(f使用简单复制: {output_path}) return True except Exception as e: LOG.error(f简单复制失败: {e}) return False except Exception as e: LOG.error(f视频优化失败: {e}) # 尝试简单复制作为备选 try: shutil.copy2(input_path, output_path) LOG.info(f优化失败使用简单复制: {output_path}) return True except: return False def _export_metadata(export_dir: Path, video_path: str, exported_files: list, result, export_format: str, low_rounds: bool False, total_rounds: int 0, optimize_info: dict None) - Path | None: metadata_file export_dir / rounds_metadata.json try: metadata {source_video: video_path, export_time: datetime.now().isoformat(), total_rounds: len(exported_files), format: export_format, rounds: []} # 如果是低回合视频添加标记 if low_rounds: metadata[is_low_rounds] True metadata[low_rounds_threshold] 5 metadata[actual_rounds] total_rounds metadata[reason] f少于5个回合实际{total_rounds}个 # 添加优化信息 if optimize_info: metadata[optimization] optimize_info # 尝试获取每个回合的时间信息 for i, file_path in enumerate(exported_files, 1): round_info {round_number: i, file_name: Path(file_path).name, file_path: file_path} # 如果有时间信息可以添加 if hasattr(result, round_timestamps) and result.round_timestamps: if i len(result.round_timestamps): round_info[timestamp] result.round_timestamps[i - 1] metadata[rounds].append(round_info) with open(metadata_file, w, encodingutf-8) as f: json.dump(metadata, f, ensure_asciiFalse, indent2) LOG.info(f 元数据已导出: {metadata_file}) return metadata_file except Exception as e: LOG.warning(f导出元数据失败: {e}) return None def export_rounds_videos(result, video_path: str, output_dir: str, export_format: str mp4, low_rounds_threshold: int 5, optimize_video: bool True, target_fps: float None, video_quality: str high) - dict: exported_info {total_rounds: 0, exported_files: [], export_dir: output_dir, success: False, low_rounds_dir: None, low_rounds_files: [], low_rounds_count: 0, is_low_rounds: False, metadata_file: None, low_metadata_file: None, optimization_info: {optimized: optimize_video, target_fps: target_fps, quality: video_quality}} # 尝试获取回合视频列表 - 检查多个可能的属性名 round_videos [] possible_attrs [ball_video_list, round_video_list, all_ball_video_list, segment_video_list] for attr in possible_attrs: if hasattr(result, attr): video_list getattr(result, attr) if video_list and isinstance(video_list, list): round_videos video_list LOG.info(f找到回合视频列表: {attr}, 共 {len(round_videos)} 个) break if not round_videos: LOG.warning(未找到回合视频列表请检查剪辑结果对象) # 尝试从 result 的 dict 中获取 if hasattr(result, __dict__): for key, value in result.__dict__.items(): if isinstance(value, list) and value and any(str(v).endswith((.mp4, .avi, .mov)) for v in value): round_videos value LOG.info(f从 {key} 找到回合视频列表) break if not round_videos: LOG.error(无法找到回合视频请确保剪辑过程生成了独立的回合视频文件) return exported_info # 创建导出目录 export_dir Path(output_dir) export_dir.mkdir(parentsTrue, exist_okTrue) # 创建回合子目录 base_name Path(video_path).stem timestamp datetime.now().strftime(%Y%m%d_%H%M%S) # 先统计总回合数 total_rounds len(round_videos) LOG.info(f检测到总回合数: {total_rounds}) # 判断是否少于阈值 is_low_rounds total_rounds low_rounds_threshold if is_low_rounds: LOG.warning(f⚠️ 视频只有 {total_rounds} 个回合少于{low_rounds_threshold}个将导出到低回合文件夹) rounds_subdir export_dir / f{base_name}_low_rounds_{timestamp} else: LOG.info(f✅ 视频有 {total_rounds} 个回合正常) rounds_subdir export_dir / f{base_name}_rounds_{timestamp} rounds_subdir.mkdir(parentsTrue, exist_okTrue) # 导出每个回合 exported_files [] export_stats {total_frames: 0, total_size: 0, avg_fps: 0} for i, round_video_path in enumerate(round_videos, 1): if not os.path.exists(round_video_path): LOG.warning(f回合 {i} 视频文件不存在: {round_video_path}) continue # 构建目标文件名 round_filename fround_{i:03d}.{export_format} dest_path rounds_subdir / round_filename try: success False # 如果启用优化使用优化导出 if optimize_video: LOG.info(f 优化导出回合 {i}...) success optimize_video_export(input_pathround_video_path, output_pathstr(dest_path), target_fpstarget_fps, qualityvideo_quality) # 如果优化失败或未启用使用简单复制 if not success: LOG.info(f 使用简单复制回合 {i}) shutil.copy2(round_video_path, dest_path) success True if success: exported_files.append(str(dest_path)) # 统计信息 file_size os.path.getsize(dest_path) export_stats[total_size] file_size LOG.info(f✅ 回合 {i} 导出成功: {dest_path} ({file_size / 1024 / 1024:.2f}MB)) else: LOG.error(f❌ 回合 {i} 导出失败) except Exception as e: LOG.error(f❌ 回合 {i} 导出异常: {e}) # 最后尝试直接复制 try: shutil.copy2(round_video_path, dest_path) exported_files.append(str(dest_path)) LOG.warning(f⚠️ 回合 {i} 使用应急复制) except Exception as e2: LOG.error(f❌ 回合 {i} 所有导出方式都失败: {e2}) # 更新统计信息 export_stats[exported_count] len(exported_files) if exported_files: export_stats[avg_size] export_stats[total_size] / len(exported_files) # 导出元数据 metadata_file _export_metadata(export_dirrounds_subdir, video_pathvideo_path, exported_filesexported_files, resultresult, export_formatexport_format, low_roundsis_low_rounds, total_roundstotal_rounds, optimize_infoexported_info[optimization_info] if optimize_video else None) # 更新返回信息 exported_info.update({total_rounds: total_rounds, exported_files: exported_files, export_dir: str(rounds_subdir), metadata_file: str(metadata_file) if metadata_file else None, success: len(exported_files) 0, is_low_rounds: is_low_rounds, low_rounds_count: len(exported_files) if is_low_rounds else 0, low_rounds_dir: str(rounds_subdir) if is_low_rounds else None, export_stats: export_stats}) # 打印总结信息 print(\n * 60) print( 导出完成统计) print( * 60) if is_low_rounds: print(f⚠️ 低回合视频) print(f 总回合数: {total_rounds} (少于{low_rounds_threshold}个)) else: print(f✅ 正常视频) print(f 总回合数: {total_rounds}) print(f 成功导出: {len(exported_files)} 个) print(f 导出目录: {rounds_subdir}) if optimize_video: print(f 视频优化: 已启用) if target_fps: print(f 目标帧率: {target_fps} fps) print(f 视频质量: {video_quality}) else: print(f 视频优化: 未启用) if metadata_file: print(f 元数据: {metadata_file}) print( * 60 \n) return exported_info def run_clip(args: argparse.Namespace, config: Config) - None: video_path os.path.abspath(args.video_path) if not os.path.isfile(video_path): LOG.error(f视频文件不存在: {video_path}) sys.exit(1) if not args.sport: LOG.error(clip 模式需要指定 --sportping_pong 或 badminton) sys.exit(1) auto_clip_config config.auto_clip_config common_options auto_clip_config.common_options if args.output_dir: common_options.output_dir path_utils.get_project_path(args.output_dir) large_model_service LargeModelService(config.large_model_service_config) if args.sport ping_pong: clipper PingPongAutoClipper(auto_clip_config.ping_pong, common_options, large_model_service) else: clipper BadmintonAutoClipper(auto_clip_config.badminton, common_options, large_model_service) LOG.info(f开始剪辑: {video_path}) result clipper.autoclip_video(video_path, auto_clip_config_build_auto_clip_config(args)) # 导出合并视频 merged_video_path result.all_match_merged_video_path LOG.info(f✅ 合并视频完成: {merged_video_path}) print(f\n合并视频: {merged_video_path}) # 如果启用了回合导出 if args.export_rounds: LOG.info( 开始导出回合分段视频...) # 确定输出目录 if args.rounds_dir: output_dir os.path.abspath(args.rounds_dir) else: output_dir args.output_dir or str(common_options.output_dir) # 导出回合视频 exported_info export_rounds_videos(resultresult, video_pathvideo_path, output_diroutput_dir, export_formatargs.export_format, low_rounds_thresholdargs.low_rounds_threshold, optimize_videoargs.optimize_video, target_fpsargs.target_fps, video_qualityargs.video_quality) if exported_info[success]: LOG.info(f✅ 所有回合导出完成共 {exported_info[total_rounds]} 个回合) # 显示详细信息 print(f\n 最终结果:) print(f 视频文件: {Path(video_path).name}) print(f 总回合数: {exported_info[total_rounds]}) print(f 成功导出: {len(exported_info[exported_files])} 个) if exported_info[is_low_rounds]: print(f 状态: ⚠️ 低回合视频 ( {args.low_rounds_threshold} 个)) else: print(f 状态: ✅ 正常视频) print(f 导出目录: {exported_info[export_dir]}) if exported_info.get(metadata_file): print(f 元数据: {exported_info[metadata_file]}) # 显示优化信息 if args.optimize_video: print(f 优化: ✅ 已启用 (质量: {args.video_quality})) else: print(f 优化: ❌ 未启用) else: LOG.warning(⚠️ 没有回合视频被导出请检查剪辑结果) print(f\n✅ 处理完成: {merged_video_path}) def run_serve(config: Config) - None: from src.main.http.internal_http import InternalHttp from src.main.service.video_edit_service import VideoEditService auto_clip_config config.auto_clip_config common_options auto_clip_config.common_options large_model_service LargeModelService(config.large_model_service_config) http_client InternalHttp(config.internal.http).client pingpong_auto_clipper PingPongAutoClipper(auto_clip_config.ping_pong, common_options, large_model_service) badminton_auto_clipper BadmintonAutoClipper(auto_clip_config.badminton, common_options, large_model_service) video_edit_service VideoEditService(service_configconfig.service_config, mysql_configconfig.datasource_config.mysql, pingpong_auto_clipperpingpong_auto_clipper, badminton_auto_clipperbadminton_auto_clipper, kafka_configconfig.kafka_config, http_clienthttp_client, ) try: video_edit_service.start() except Exception as e: video_edit_service.stop() LOG.error(f启动服务失败: {e}) def run_from_config(config: Config) - None: if config.job_type JobType.SERVICE: run_serve(config) else: LOG.error(f未知的 job_type: {config.job_type}) sys.exit(1) def main() - None: parser _build_parser() args parser.parse_args() config load_config(args.config) mode _resolve_mode(args) if mode clip: run_clip(args, config) elif mode serve: run_serve(config) else: run_from_config(config) if __name__ __main__: # patch_auto_clipper() main()羽球时刻 推荐yuqiushike.comBadminton-Highlight-Extractionhttps://github.com/Manoj-A-Anandan/Badminton-Highlight-Extraction
)
