sing_detect/batch_detect.py

#!/usr/bin/env python3
"""
Batch Singing Detector
=======================
Process multiple stream recordings at once.
Point it at a folder of recordings and it will process each one.

Usage:
  python batch_detect.py "D:\\streams"
  python batch_detect.py "D:\\streams" --output-dir "D:\\singing_edits" --auto-cut
  python batch_detect.py "D:\\streams" --pattern "*.mp4" --recursive
"""

import argparse
import os
import sys
import time
import glob
from pathlib import Path


def find_videos(input_dir: str, pattern: str = "*", recursive: bool = False, 
                extensions: tuple = (".mp4", ".mkv", ".flv", ".ts", ".avi", ".mov", ".webm")) -> list:
    """Find all video files in directory."""
    videos = []
    
    if recursive:
        for ext in extensions:
            videos.extend(glob.glob(os.path.join(input_dir, "**", f"*{ext}"), recursive=True))
    else:
        for ext in extensions:
            videos.extend(glob.glob(os.path.join(input_dir, f"*{ext}")))
    
    # Filter by pattern if specified
    if pattern != "*":
        videos = [v for v in videos if Path(v).match(pattern)]
    
    # Sort by modification time (newest first)
    videos.sort(key=lambda x: os.path.getmtime(x), reverse=True)
    
    return videos


def main():
    parser = argparse.ArgumentParser(
        description="Batch process multiple stream recordings for singing detection.",
    )
    
    parser.add_argument("input_dir", help="Directory containing stream recordings")
    parser.add_argument("--output-dir", "-o", default=None,
                        help="Output directory (default: same as input)")
    parser.add_argument("--pattern", default="*",
                        help="Filename pattern to match (default: all video files)")
    parser.add_argument("--recursive", "-r", action="store_true",
                        help="Search subdirectories recursively")
    parser.add_argument("--gap", "-g", type=float, default=10.0,
                        help="Max gap to merge singing segments (default: 10s)")
    parser.add_argument("--min-duration", "-m", type=float, default=20.0,
                        help="Minimum segment duration (default: 20s)")
    parser.add_argument("--padding", "-p", type=float, default=2.0,
                        help="Padding before/after segments (default: 2s)")
    parser.add_argument("--fps", type=float, default=29.97,
                        help="Frame rate (default: 29.97)")
    parser.add_argument("--auto-cut", action="store_true",
                        help="Generate auto-cut ffmpeg scripts")
    parser.add_argument("--skip-existing", action="store_true",
                        help="Skip files that already have output .edl files")
    parser.add_argument("--limit", type=int, default=0,
                        help="Process only N files (0 = all)")
    
    args = parser.parse_args()
    
    if not os.path.isdir(args.input_dir):
        print(f"[ERROR] Directory not found: {args.input_dir}")
        sys.exit(1)
    
    # Find videos
    videos = find_videos(args.input_dir, args.pattern, args.recursive)
    
    if not videos:
        print(f"[INFO] No video files found in: {args.input_dir}")
        sys.exit(0)
    
    # Filter already-processed files
    if args.skip_existing:
        output_dir = args.output_dir or args.input_dir
        unprocessed = []
        for v in videos:
            stem = Path(v).stem
            edl = os.path.join(output_dir, f"{stem}_singing.edl")
            if not os.path.exists(edl):
                unprocessed.append(v)
            else:
                print(f"[SKIP] Already processed: {Path(v).name}")
        videos = unprocessed
    
    # Apply limit
    if args.limit > 0:
        videos = videos[:args.limit]
    
    print(f"\n{'='*60}")
    print(f"  Batch Singing Detector")
    print(f"{'='*60}")
    print(f"  Input:    {args.input_dir}")
    print(f"  Videos:   {len(videos)} to process")
    print(f"{'='*60}\n")
    
    # Process each video
    # Import the main detect_singing module
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from detect_singing import (
        extract_audio, run_segmentation, merge_singing_segments,
        export_edl, export_csv, export_davinci_markers_csv,
        export_json, export_ffmpeg_concat, format_timecode,
    )
    
    results = []
    for i, video_path in enumerate(videos, 1):
        video_name = Path(video_path).name
        print(f"\n{'─'*60}")
        print(f"  [{i}/{len(videos)}] Processing: {video_name}")
        print(f"{'─'*60}")
        
        try:
            output_dir = args.output_dir or os.path.dirname(video_path)
            os.makedirs(output_dir, exist_ok=True)
            
            stem = Path(video_path).stem
            wav_path = os.path.join(output_dir, f"{stem}_audio.wav")
            
            # Step 1-3
            extract_audio(video_path, wav_path)
            raw_segments = run_segmentation(wav_path)
            singing = merge_singing_segments(
                raw_segments, 
                max_gap=args.gap,
                min_duration=args.min_duration,
                padding=args.padding,
            )
            
            # Step 4: Export
            if singing:
                edl = os.path.join(output_dir, f"{stem}_singing.edl")
                csv_f = os.path.join(output_dir, f"{stem}_singing.csv")
                markers = os.path.join(output_dir, f"{stem}_markers.csv")
                json_f = os.path.join(output_dir, f"{stem}_singing.json")
                
                export_edl(singing, edl, fps=args.fps,
                          source_filename=Path(video_path).name)
                export_csv(singing, csv_f)
                export_davinci_markers_csv(singing, markers, fps=args.fps)
                export_json(singing, json_f)
                
                if args.auto_cut:
                    export_ffmpeg_concat(singing, video_path, json_f)
                
                total_dur = sum(s["duration"] for s in singing)
                results.append((video_name, len(singing), total_dur, "OK"))
            else:
                results.append((video_name, 0, 0, "No songs found"))
            
            # Cleanup WAV
            if os.path.exists(wav_path):
                os.remove(wav_path)
                
        except Exception as e:
            print(f"  [ERROR] {e}")
            results.append((video_name, 0, 0, f"Error: {e}"))
    
    # Summary
    print(f"\n\n{'='*60}")
    print(f"  BATCH COMPLETE")
    print(f"{'='*60}")
    print(f"  {'Video':<40s} {'Songs':>5s}  {'Singing Time':>12s}  Status")
    print(f"  {'─'*40} {'─'*5}  {'─'*12}  {'─'*15}")
    for name, count, dur, status in results:
        name_short = name[:38] + ".." if len(name) > 40 else name
        print(f"  {name_short:<40s} {count:>5d}  {format_timecode(dur):>12s}  {status}")
    print(f"{'='*60}")


if __name__ == "__main__":
    main()