src/humeo/cli.py

"""CLI entry point for the Humeo pipeline."""

import argparse
import logging
import sys
from datetime import datetime
from pathlib import Path

from humeo.config import PipelineConfig
from humeo.pipeline import run_pipeline


def setup_logging(verbose: bool = False):
    """Configure logging with a clean format."""
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s | %(levelname)-7s | %(name)s | %(message)s",
        datefmt="%H:%M:%S",
        handlers=[logging.StreamHandler(sys.stdout)],
    )
    # Suppress noisy third-party loggers
    logging.getLogger("urllib3").setLevel(logging.WARNING)
    logging.getLogger("httpx").setLevel(logging.WARNING)


def build_parser() -> argparse.ArgumentParser:
    """Build the argument parser."""
    parser = argparse.ArgumentParser(
        prog="humeo",
        description="Humeo - Automated podcast-to-shorts pipeline from YouTube or local MP4",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  humeo --long-to-shorts "https://youtube.com/watch?v=abc123"
  humeo --long-to-shorts "C:\\Videos\\episode.mp4"
  humeo --long-to-shorts "https://youtube.com/watch?v=abc123" --work-dir .humeo_work
  humeo --long-to-shorts "https://youtube.com/watch?v=abc123" --gemini-model gemini-2.0-flash
        """,
    )

    parser.add_argument(
        "--long-to-shorts",
        metavar="SOURCE",
        required=True,
        help="YouTube video URL or local MP4 path to process",
    )

    parser.add_argument(
        "--output", "-o",
        type=Path,
        default=Path("output"),
        help="Output directory for final shorts (default: ./output)",
    )

    parser.add_argument(
        "--work-dir",
        type=Path,
        default=None,
        help="Working directory for intermediate files. Default: per-video folder under the "
        "cache root (see docs/ENVIRONMENT.md). Use this to force e.g. ./.humeo_work.",
    )

    parser.add_argument(
        "--no-video-cache",
        action="store_true",
        help="Do not use per-video cache dirs; use ./.humeo_work unless --work-dir is set.",
    )

    parser.add_argument(
        "--cache-root",
        type=Path,
        default=None,
        help="Override cache root for manifests and per-video ingest (env: HUMEO_CACHE_ROOT).",
    )

    parser.add_argument(
        "--gemini-model",
        default=None,
        help="Gemini model id for clip selection (default: GEMINI_MODEL env; see humeo.config).",
    )

    parser.add_argument(
        "--force-clip-selection",
        action="store_true",
        help="Re-run clip-selection LLM even when clips.meta.json matches the transcript.",
    )

    parser.add_argument(
        "--gemini-vision-model",
        default=None,
        help="Gemini model for per-keyframe layout + bbox (default: GEMINI_VISION_MODEL env or --gemini-model).",
    )

    parser.add_argument(
        "--force-layout-vision",
        action="store_true",
        help="Re-run Gemini vision for layouts even when layout_vision.meta.json matches.",
    )

    parser.add_argument(
        "--prune-level",
        choices=["off", "conservative", "balanced", "aggressive"],
        default="balanced",
        help=(
            "Stage 2.5 inner-clip content pruning aggressiveness. "
            "'off' skips pruning entirely; 'conservative' trims <=10%%, "
            "'balanced' <=20%%, 'aggressive' <=35%% of each clip "
            "(always clamped to the MIN_CLIP_DURATION_SEC floor). Default: balanced."
        ),
    )

    parser.add_argument(
        "--force-content-pruning",
        action="store_true",
        help="Re-run content-pruning LLM even when prune.meta.json matches.",
    )

    parser.add_argument(
        "--no-hook-detection",
        action="store_true",
        help=(
            "Skip Stage 2.25 hook detection. The selector's hook window "
            "(possibly the 0.0-3.0s placeholder) will be carried through. "
            "Stage 2.5 content pruning still treats that exact placeholder "
            "as 'no hook' so pruning is not disabled."
        ),
    )

    parser.add_argument(
        "--force-hook-detection",
        action="store_true",
        help="Re-run hook-detection LLM even when hooks.meta.json matches.",
    )

    parser.add_argument(
        "--clean-run",
        action="store_true",
        help=(
            "Run with a fresh work dir and no cache reuse. Implies --no-video-cache, "
            "--force-clip-selection, --force-layout-vision, and overwrite existing outputs."
        ),
    )

    parser.add_argument(
        "--interactive", "-i",
        action="store_true",
        help="Pause after clip selection and after render for human approval.",
    )

    parser.add_argument(
        "--subtitle-font-size",
        type=int,
        default=48,
        help=(
            "Caption font size in output pixels. libass is pinned to "
            "original_size=1080x1920, so this is a true pixel value. "
            "(default: 48)"
        ),
    )

    parser.add_argument(
        "--subtitle-margin-v",
        type=int,
        default=160,
        help="Caption bottom margin in output pixels (default: 160)",
    )

    parser.add_argument(
        "--subtitle-max-words",
        type=int,
        default=4,
        help="Max words per subtitle cue (default: 4)",
    )

    parser.add_argument(
        "--subtitle-max-cue-sec",
        type=float,
        default=2.2,
        help="Max subtitle cue duration in seconds (default: 2.2)",
    )

    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Enable debug logging",
    )

    return parser


def main():
    """CLI entry point."""
    parser = build_parser()
    args = parser.parse_args()
    setup_logging(args.verbose)

    use_video_cache = not args.no_video_cache
    force_clip_selection = args.force_clip_selection
    force_layout_vision = args.force_layout_vision
    force_content_pruning = args.force_content_pruning
    force_hook_detection = args.force_hook_detection
    detect_hooks = not args.no_hook_detection
    overwrite_outputs = False
    work_dir = args.work_dir

    if args.clean_run:
        use_video_cache = False
        force_clip_selection = True
        force_layout_vision = True
        force_content_pruning = True
        force_hook_detection = True
        overwrite_outputs = True
        if work_dir is None:
            stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            work_dir = Path(f".humeo_work_clean_{stamp}")

    config = PipelineConfig(
        youtube_url=args.long_to_shorts,
        output_dir=args.output,
        work_dir=work_dir,
        use_video_cache=use_video_cache,
        cache_root=args.cache_root,
        gemini_model=args.gemini_model,
        gemini_vision_model=args.gemini_vision_model,
        force_clip_selection=force_clip_selection,
        force_layout_vision=force_layout_vision,
        clean_run=args.clean_run,
        overwrite_outputs=overwrite_outputs,
        interactive=args.interactive,
        prune_level=args.prune_level,
        force_content_pruning=force_content_pruning,
        detect_hooks=detect_hooks,
        force_hook_detection=force_hook_detection,
        subtitle_font_size=args.subtitle_font_size,
        subtitle_margin_v=args.subtitle_margin_v,
        subtitle_max_words_per_cue=args.subtitle_max_words,
        subtitle_max_cue_sec=args.subtitle_max_cue_sec,
    )

    try:
        outputs = run_pipeline(config)
        print(f"\nDone. {len(outputs)} shorts generated in: {config.output_dir}")
        for p in outputs:
            print(f"   -> {p}")
    except KeyboardInterrupt:
        print("\nPipeline interrupted.")
        sys.exit(1)
    except Exception as e:
        logging.getLogger(__name__).error("Pipeline failed: %s", e, exc_info=True)
        sys.exit(1)


if __name__ == "__main__":
    main()