src/cli/main.py

"""
Main CLI entry point for Voice Tools.

Provides command-line interface for voice extraction and profiling tasks.
"""

import logging
from pathlib import Path
from typing import List, Optional

import click
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Configure SSL context BEFORE any model-related imports
from src.config.ssl_config import configure_ssl_context

configure_ssl_context()

from ..models.processing_job import ExtractionMode, ProcessingJob
from ..services.batch_processor import BatchProcessor
from .progress import (
    ExtractionProgress,
    display_config,
    display_error,
    display_failures,
    display_header,
    display_info,
    display_statistics,
    display_success,
    display_vad_stats,
    display_warning,
)
from .utils import discover_audio_files, validate_audio_files

# Configure logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


@click.group()
@click.version_option(version="0.1.0", prog_name="voice-tools")
def cli():
    """
    Voice Tools - Extract and profile voices from audio files.

    This tool helps you extract specific voices from audio files using
    speaker diarization and voice matching. It can separate speech from
    nonverbal sounds and apply quality filtering.
    """
    pass


# Import and register commands
from .denoise import denoise
from .extract_speaker import extract_speaker
from .separate import separate

cli.add_command(separate)
cli.add_command(extract_speaker)
cli.add_command(denoise)


@cli.command()
@click.argument("reference_file", type=click.Path(exists=True, path_type=Path))
@click.argument(
    "input_paths", nargs=-1, type=click.Path(exists=True, path_type=Path), required=True
)
@click.option(
    "--output-dir",
    "-o",
    type=click.Path(path_type=Path),
    default="./output",
    help="Output directory for extracted segments (default: ./output)",
)
@click.option(
    "--mode",
    "-m",
    type=click.Choice(["speech", "nonverbal", "both"], case_sensitive=False),
    default="speech",
    help="What to extract: speech, nonverbal, or both (default: speech)",
)
@click.option(
    "--vad-threshold", type=float, default=0.5, help="VAD confidence threshold 0-1 (default: 0.5)"
)
@click.option(
    "--voice-threshold",
    type=float,
    default=0.7,
    help="Voice similarity threshold 0-1 (default: 0.7)",
)
@click.option(
    "--speech-threshold",
    type=float,
    default=0.6,
    help="Speech classification threshold 0-1 (default: 0.6)",
)
@click.option("--no-vad", is_flag=True, help="Disable VAD pre-filtering (slower but more thorough)")
@click.option(
    "--no-quality-filter",
    is_flag=True,
    help="Disable quality filtering (include lower quality segments)",
)
@click.option("--verbose", "-v", is_flag=True, help="Enable verbose logging")
@click.option(
    "--pattern", "-p", default="*.m4a", help="Glob pattern for directory scanning (default: *.m4a)"
)
def extract(
    reference_file: Path,
    input_paths: tuple,
    output_dir: Path,
    mode: str,
    vad_threshold: float,
    voice_threshold: float,
    speech_threshold: float,
    no_vad: bool,
    no_quality_filter: bool,
    verbose: bool,
    pattern: str,
):
    """
    Extract voice segments from audio files.

    REFERENCE_FILE: Audio file containing the reference voice to extract
    INPUT_PATHS: One or more files, directories, or glob patterns

    Examples:

    \b
    # Extract speech from single file
    voice-tools extract reference.m4a input.m4a

    \b
    # Extract from multiple files
    voice-tools extract reference.m4a file1.m4a file2.m4a file3.m4a

    \b
    # Process entire directory
    voice-tools extract reference.m4a ./audio_files/

    \b
    # Process directory with custom pattern
    voice-tools extract reference.m4a ./audio_files/ --pattern "*.wav"

    \b
    # Extract nonverbal sounds only
    voice-tools extract reference.m4a input.m4a --mode nonverbal

    \b
    # Extract both speech and nonverbal
    voice-tools extract reference.m4a input.m4a --mode both

    \b
    # Custom output directory
    voice-tools extract reference.m4a input.m4a -o ./my_output

    \b
    # Adjust voice matching sensitivity
    voice-tools extract reference.m4a input.m4a --voice-threshold 0.8
    """
    # Configure logging
    if verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    display_header("Voice Tools - Extract Voice Segments")

    # Validate reference file
    if not reference_file.exists():
        display_error(f"Reference file not found: {reference_file}")
        raise click.Abort()

    # Discover audio files from input paths (files, directories, or patterns)
    display_info(f"Discovering audio files from {len(input_paths)} input path(s)...")
    input_files_list = discover_audio_files(list(input_paths), pattern=pattern)

    if not input_files_list:
        display_error("No audio files found in the specified paths")
        raise click.Abort()

    display_success(f"Found {len(input_files_list)} audio file(s) to process")

    # Validate discovered files
    valid_files, errors = validate_audio_files(input_files_list)

    if errors:
        display_warning(f"Validation issues found:")
        for error in errors:
            display_warning(f"  {error}")

    if not valid_files:
        display_error("No valid audio files to process")
        raise click.Abort()

    if len(valid_files) < len(input_files_list):
        display_info(
            f"Processing {len(valid_files)} valid files (skipped {len(input_files_list) - len(valid_files)})"
        )

    input_files_list = valid_files

    # Display configuration
    config = {
        "Reference voice": str(reference_file),
        "Input files": len(input_files_list),
        "Output directory": str(output_dir),
        "Extraction mode": mode,
        "VAD enabled": not no_vad,
        "Quality filter": not no_quality_filter,
        "VAD threshold": vad_threshold,
        "Voice threshold": voice_threshold,
        "Speech threshold": speech_threshold,
    }
    display_config(config)

    # Convert mode string to ExtractionMode enum
    mode_map = {
        "speech": ExtractionMode.SPEECH,
        "nonverbal": ExtractionMode.NONVERBAL,
        "both": ExtractionMode.BOTH,
    }
    extraction_mode = mode_map[mode.lower()]

    # Create processing job
    job = ProcessingJob(
        reference_file=str(reference_file),
        input_files=[str(f) for f in input_files_list],
        output_dir=str(output_dir),
        extraction_mode=extraction_mode,
        vad_threshold=vad_threshold,
        voice_similarity_threshold=voice_threshold,
        speech_confidence_threshold=speech_threshold,
        apply_denoising=False,  # Not implemented yet
    )

    # Initialize processor
    processor = BatchProcessor(
        vad_threshold=vad_threshold,
        voice_similarity_threshold=voice_threshold,
        speech_confidence_threshold=speech_threshold,
        enable_vad=not no_vad,
    )

    # Process batch
    try:
        display_info("Starting extraction...")

        with ExtractionProgress() as progress:
            progress.start(len(input_files_list))
            job = processor.process_batch(job)

        # Display results
        display_header("Extraction Complete")

        summary = job.get_summary()
        display_statistics(summary)

        if summary["files_failed"] > 0:
            display_failures(job.failed_files)

        display_success(f"Output saved to: {output_dir}")

        # Generate detailed report
        report_path = output_dir / "extraction_report.txt"
        report_content = job.generate_report()
        report_path.write_text(report_content)
        display_success(f"Detailed report saved to: {report_path}")

    except KeyboardInterrupt:
        click.echo("\nExtraction cancelled by user", err=True)
        raise click.Abort()
    except Exception as e:
        click.echo(f"\nError during extraction: {e}", err=True)
        logger.exception("Extraction failed")
        raise click.Abort()


@cli.command()
@click.argument("audio_file", type=click.Path(exists=True, path_type=Path))
@click.option(
    "--vad-threshold", type=float, default=0.5, help="VAD confidence threshold 0-1 (default: 0.5)"
)
def scan(audio_file: Path, vad_threshold: float):
    """
    Scan an audio file for voice activity.

    Performs a quick VAD scan to estimate processing time and voice activity.
    Useful for determining if a file is worth processing.

    AUDIO_FILE: Audio file to scan

    Example:

    \b
    voice-tools scan input.m4a
    """
    display_header("Voice Tools - Voice Activity Scan")

    processor = BatchProcessor(vad_threshold=vad_threshold)

    try:
        display_info(f"Scanning: {audio_file}")

        estimates = processor.estimate_processing_time(audio_file, enable_vad=True)

        # Create VAD stats for display
        vad_stats = {
            "total_duration": estimates["total_duration"],
            "voice_duration": estimates["voice_duration"],
            "voice_percentage": (estimates["voice_duration"] / estimates["total_duration"]) * 100,
            "worth_processing": estimates["voice_duration"] >= 30,
        }
        display_vad_stats(vad_stats)

        stats = {
            "estimated_processing_time": estimates["estimated_processing_time"],
            "estimated_minutes": estimates["estimated_minutes"],
        }

        from rich.table import Table

        from .progress import console

        table = Table(title="Processing Estimate", show_header=False)
        table.add_column("Metric", style="cyan")
        table.add_column("Value", style="white")
        table.add_row(
            "Estimated processing time",
            f"{stats['estimated_processing_time']:.2f}s ({stats['estimated_minutes']:.2f} min)",
        )
        console.print(table)
        console.print()

        if estimates["voice_duration"] < 30:
            display_warning(
                "Very little voice activity detected. File may not be worth processing."
            )
        elif vad_stats["voice_percentage"] < 10:
            display_info("Low voice activity. VAD optimization will provide significant speedup.")

    except Exception as e:
        display_error(f"Scan failed: {e}")
        logger.exception("Scan failed")
        raise click.Abort()


@cli.command()
@click.option("--host", default="0.0.0.0", help="Server hostname (default: 0.0.0.0)")
@click.option("--port", default=7860, type=int, help="Server port (default: 7860)")
@click.option("--share", is_flag=True, help="Create public share link")
def web(host: str, port: int, share: bool):
    """
    Launch the web interface.

    Opens a browser-based UI for voice extraction with file upload,
    configuration, and result download.

    Example:

    \b
    voice-tools web
    voice-tools web --port 8080 --share
    """
    from ..web.app import launch

    display_header("Voice Tools - Web Interface")
    display_info(f"Starting web server on http://{host}:{port}")

    if share:
        display_info("Creating public share link...")

    display_success("Server starting... Open the URL in your browser")

    try:
        launch(server_name=host, server_port=port, share=share, debug=False)
    except KeyboardInterrupt:
        display_info("Server stopped")
    except Exception as e:
        display_error(f"Failed to start server: {e}")
        raise click.Abort()


@cli.command()
def info():
    """
    Display information about Voice Tools.

    Shows configuration, model information, and system details.
    """
    import torch
    from rich.table import Table

    from ..services.model_manager import ModelManager
    from .progress import console

    display_header("Voice Tools - System Information")

    # Version info
    info_table = Table(title="Version", show_header=False)
    info_table.add_column("Key", style="cyan")
    info_table.add_column("Value", style="white")
    info_table.add_row("Version", "0.1.0")
    console.print(info_table)
    console.print()

    # Models info
    models_table = Table(title="Models", show_header=False)
    models_table.add_column("Component", style="cyan")
    models_table.add_column("Model", style="white")
    models_table.add_row("Speaker Diarization", "pyannote/speaker-diarization-3.1")
    models_table.add_row("Voice Embedding", "pyannote/embedding")
    models_table.add_row("Speech Classifier", "MIT/ast-finetuned-audioset-10-10-0.4593")
    models_table.add_row("VAD", "Silero VAD v4.0")
    console.print(models_table)
    console.print()

    # Environment info
    env_table = Table(title="Environment", show_header=False)
    env_table.add_column("Key", style="cyan")
    env_table.add_column("Value", style="white")
    env_table.add_row("PyTorch version", torch.__version__)
    env_table.add_row("CUDA available", "Yes" if torch.cuda.is_available() else "No")
    if torch.cuda.is_available():
        env_table.add_row("CUDA device", torch.cuda.get_device_name(0))
    console.print(env_table)
    console.print()

    # Check for HuggingFace token
    model_manager = ModelManager()
    token = model_manager.get_hf_token()
    if token:
        display_success("HuggingFace token: Configured")
    else:
        display_warning("HuggingFace token: Not configured")
        display_info("Some models require authentication. Set HF_TOKEN environment variable.")


def main():
    """Entry point for the CLI."""
    cli()


if __name__ == "__main__":
    main()