企业官网建设流程全解析

目录

good-badminton 推荐，看起来还行

racquet-sports-analyzer

huji

推理脚本：

羽球时刻 推荐：

Badminton-Highlight-Extraction

good-badminton 推荐，看起来还行

https://github.com/qwpyyx/Good-Badminton

racquet-sports-analyzer

Bot-Derpy/racquet-sports-analyzer

huji

https://github.com/hhoao/huji https://github.com/hhoao/huji-algorithm

conda create -p E:\soft\envs\py312 python=3.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 badminton

import 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(key=extract_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 依赖，请先安装并激活虚拟环境：", file=sys.stderr) if sys.platform == "win32": print(" .\\setup.ps1", file=sys.stderr) print(" .venv\\Scripts\\activate", file=sys.stderr) else: print(" ./setup.sh", file=sys.stderr) print(" source .venv/bin/activate", file=sys.stderr) venv_python = _venv_python() if venv_python: print(f" 或直接: {venv_python} main.py ...", file=sys.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_class=argparse.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", default=CONFIG_PATH, help=f"配置目录（含 application.yml），默认 {CONFIG_PATH}", ) parser.add_argument("--video-path", "-v", metavar="PATH", default=r"C:\Users\ChanJing-01\Videos\yumao\yumao1.mp4", help="本地视频文件路径") parser.add_argument("--sport", default="badminton", choices=["ping_pong", "badminton"], help="运动类型（clip 模式必填）", ) parser.add_argument("--match-type", choices=["singles", "doubles"], default="singles", help="羽毛球比赛类型，默认 singles", ) parser.add_argument("--output-dir", "-o", metavar="DIR", help="剪辑输出目录") parser.add_argument("--serve", action="store_true", help="启动 Kafka + HTTP 服务") parser.add_argument("--train", action="store_true", help="训练模型") # 新增回合导出相关参数 parser.add_argument("--export-rounds", action="store_true", default=True, help="将每个回合单独导出为独立视频文件") parser.add_argument("--rounds-dir", metavar="DIR", help="回合视频导出目录（默认在输出目录下创建 rounds_时间戳 子目录）") parser.add_argument("--export-format", choices=["mp4", "avi", "mov"], default="mp4", help="导出视频格式，默认 mp4") # 新增低回合阈值参数 parser.add_argument("--low-rounds-threshold", type=int, default=5, help="低回合阈值，默认5个回合（少于该值将被分类到低回合文件夹）") # 新增视频优化参数 parser.add_argument("--optimize-video", action="store_true", default=True, help="优化导出视频，确保播放流畅（默认开启）") parser.add_argument("--target-fps", type=float, default=None, help="目标帧率（不指定则保持原始帧率）") parser.add_argument("--video-quality", choices=["low", "medium", "high"], default="high", 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_type=match_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), interpolation=cv2.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', encoding='utf-8') as f: json.dump(metadata, f, ensure_ascii=False, indent=2) 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(parents=True, exist_ok=True) # 创建回合子目录 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(parents=True, exist_ok=True) # 导出每个回合 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 = f"round_{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_path=round_video_path, output_path=str(dest_path), target_fps=target_fps, quality=video_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_dir=rounds_subdir, video_path=video_path, exported_files=exported_files, result=result, export_format=export_format, low_rounds=is_low_rounds, total_rounds=total_rounds, optimize_info=exported_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 模式需要指定 --sport（ping_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(result=result, video_path=video_path, output_dir=output_dir, export_format=args.export_format, low_rounds_threshold=args.low_rounds_threshold, optimize_video=args.optimize_video, target_fps=args.target_fps, video_quality=args.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_config=config.service_config, mysql_config=config.datasource_config.mysql, pingpong_auto_clipper=pingpong_auto_clipper, badminton_auto_clipper=badminton_auto_clipper, kafka_config=config.kafka_config, http_client=http_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.com

Badminton-Highlight-Extraction

https://github.com/Manoj-A-Anandan/Badminton-Highlight-Extraction