Spaces:
Paused
Paused
| """Video cropping and clip concatenation using MoviePy.""" | |
| import logging | |
| import traceback | |
| from datetime import datetime | |
| from pathlib import Path | |
| # Support both moviepy 1.x (moviepy.editor) and 2.x (moviepy) | |
| try: | |
| from moviepy import VideoFileClip, concatenate_videoclips # 2.x | |
| except ImportError: | |
| from moviepy.editor import VideoFileClip, concatenate_videoclips # 1.x | |
| from app.utils import format_duration | |
| from config.settings import settings | |
| logger = logging.getLogger("media_intelligence.video_processor") | |
| def _subclip(clip, start: float, end: float): | |
| """Call the correct subclip method for moviepy 1.x and 2.x.""" | |
| if hasattr(clip, "subclipped"): | |
| return clip.subclipped(start, end) # moviepy 2.x | |
| return clip.subclip(start, end) # moviepy 1.x | |
| def crop_video(video_path: str, segments: list) -> tuple: | |
| """ | |
| Extract and concatenate the requested time segments from *video_path*. | |
| Args: | |
| video_path: Path to the source video file. | |
| segments: List of dicts with 'start' and 'end' keys (seconds). | |
| Returns: | |
| (output_path, duration_str) | |
| Raises: | |
| ValueError: No valid segments provided. | |
| RuntimeError: MoviePy processing failure. | |
| """ | |
| if not segments: | |
| raise ValueError( | |
| "No segments provided for cropping. " | |
| "Please select at least one segment and try again." | |
| ) | |
| video_path = Path(video_path) | |
| logger.info(f"Crop start: {video_path.name} — {len(segments)} segment(s)") | |
| output_dir = Path(settings.OUTPUT_DIR) | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| output_path = output_dir / f"short_video_{timestamp}.mp4" | |
| source = None | |
| clips = [] | |
| try: | |
| source = VideoFileClip(str(video_path)) | |
| total_duration = source.duration | |
| logger.info(f"Source duration: {format_duration(total_duration)}") | |
| for seg in segments: | |
| start = float(seg["start"]) | |
| end = float(seg["end"]) | |
| start = max(0.0, min(start, total_duration)) | |
| end = max(0.0, min(end, total_duration)) | |
| if start >= end: | |
| logger.warning(f"Skipping invalid segment: {start:.1f}s – {end:.1f}s") | |
| continue | |
| logger.info(f"Extracting: {start:.1f}s → {end:.1f}s") | |
| clips.append(_subclip(source, start, end)) | |
| if not clips: | |
| raise ValueError( | |
| "All selected segments are outside the video duration. " | |
| "Please verify the video and re-analyse." | |
| ) | |
| logger.info(f"Joining {len(clips)} clip(s)…") | |
| final = concatenate_videoclips(clips) | |
| duration_str = format_duration(final.duration) | |
| logger.info(f"Writing: {output_path}") | |
| final.write_videofile( | |
| str(output_path), | |
| codec="libx264", | |
| audio_codec="aac", | |
| verbose=False, | |
| logger=None, | |
| ) | |
| final.close() | |
| logger.info(f"Crop complete: {output_path.name} ({duration_str})") | |
| return str(output_path), duration_str | |
| except (ValueError, RuntimeError): | |
| raise | |
| except Exception as exc: | |
| msg = ( | |
| f"Video processing failed. " | |
| f"Possible reason: corrupt file or unsupported codec. " | |
| f"Detail: {exc}" | |
| ) | |
| logger.error(f"{msg}\n{traceback.format_exc()}") | |
| raise RuntimeError(msg) from exc | |
| finally: | |
| for clip in clips: | |
| try: | |
| clip.close() | |
| except Exception: | |
| pass | |
| if source is not None: | |
| try: | |
| source.close() | |
| except Exception: | |
| pass | |