app.py

# ==============================================================================
# PitchPerfect AI: Enterprise-Grade Sales Coach (Single File Application)
#
# This single file contains the complete application code, including a
# self-contained dependency installer, YouTube support, JAX-based quantitative
# analysis, and a robust agentic architecture.
# ==============================================================================

# ==============================================================================
# DYNAMIC DEPENDENCY INSTALLATION
# This block checks for and installs missing packages.
# ==============================================================================
import sys
import subprocess
import importlib.util
import logging

# Configure basic logging for the installation process
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def install_dependencies():
    """
    Checks for required packages and installs them if they are not found.
    This makes the script self-contained and removes the need for a
    separate requirements.txt and `pip install` step.
    """
    required_packages = [
        # Core application
        ("gradio", "gradio"),
        ("google.cloud.aiplatform", "google-cloud-aiplatform"),
        ("google.cloud.storage", "google-cloud-storage"),
        ("moviepy", "moviepy"),
        # For JAX and Quantitative Analysis
        ("jax", "jax"),
        ("jaxlib", "jaxlib"),
        ("librosa", "librosa"),
        ("speechrecognition", "SpeechRecognition"),
        ("whisper", "openai-whisper"),
        # For YouTube support
        ("yt_dlp", "yt-dlp"),
    ]

    print("="*80)
    print("PitchPerfect AI: Checking for required dependencies...")
    print("="*80)

    for import_name, install_name in required_packages:
        spec = importlib.util.find_spec(import_name)
        if spec is None:
            print(f"INFO: Package '{install_name}' not found. Attempting to install...")
            try:
                subprocess.check_call([sys.executable, "-m", "pip", "install", "--quiet", install_name])
                print(f"INFO: Successfully installed '{install_name}'.")
            except subprocess.CalledProcessError:
                print(f"ERROR: Failed to install '{install_name}'. Please install it manually using:\n"
                      f"pip install {install_name}")
                sys.exit(1)
        else:
            print(f"INFO: Dependency '{install_name}' is already installed.")
    
    print("="*80)
    print("All dependencies are satisfied. Starting the application.")
    print("="*80)

# Run the dependency check and installation before anything else.
install_dependencies()


# ==============================================================================
# File: README.md (Instructions are now part of the script's execution)
# ==============================================================================
"""
# PitchPerfect AI: Enterprise-Grade Sales Coach

This application provides AI-powered feedback on sales pitches using Google's most advanced multimodal AI, all managed through the Vertex AI platform. It analyzes your content, vocal delivery, and visual presence to give you actionable insights for improvement.

This advanced version includes:
- Support for local video uploads and YouTube URLs.
- Quantitative vocal analysis powered by JAX for high performance.
- An agentic architecture where specialized tools (YouTube Downloader, JAX Analyzer) work in concert with the Gemini 1.5 Pro model.
- A self-contained dependency installer for simplified setup.

## 🔑 Prerequisites

1.  A Google Cloud Platform (GCP) project with billing enabled.
2.  The Vertex AI API and Cloud Storage API enabled in your GCP project.
3.  The `gcloud` CLI installed and authenticated on your local machine.

## 셋업

1.  **Create a Google Cloud Storage (GCS) Bucket:**
    * In your GCP project, create a new GCS bucket. It must have a globally unique name.
    * **Example name:** `your-project-id-pitch-videos`

2.  **Authenticate with Google Cloud:**
    Run the following command in your terminal and follow the prompts. This sets up Application Default Credentials (ADC).
    ```bash
    gcloud auth application-default login
    ```
    *Note: The user/principal needs `Storage Object Admin` and `Vertex AI User` roles.*

3.  **Configure Project Details:**
    * In this file, scroll down to the "CONFIGURATION" section.
    * Set your `GCP_PROJECT_ID`, `GCP_LOCATION`, and `GCS_BUCKET_NAME`.

4.  **Run the Application:**
    Simply run the script. It will automatically install any missing dependencies.
    ```bash
    python app.py
    ```
    This will launch a Gradio web server. **Look for a public URL ending in `.gradio.live` in the output and open it in your browser.**
"""

# ==============================================================================
# IMPORTS
# ==============================================================================
import json
import uuid
import os
import re
from typing import Dict, Any
import gradio as gr
import vertexai
from google.cloud import storage
from vertexai.generative_models import (
    GenerativeModel, Part, GenerationConfig,
    HarmCategory, HarmBlockThreshold
)

# Third-party imports for advanced features
import yt_dlp
import librosa
import numpy as np
import whisper
import jax
import jax.numpy as jnp
from moviepy.editor import VideoFileClip


# ==============================================================================
# CONFIGURATION
# ==============================================================================
# --- GCP and Vertex AI Configuration ---
GCP_PROJECT_ID = "aniket-personal"
GCP_LOCATION = "us-central1"

# --- GCS Configuration ---
GCS_BUCKET_NAME = "ghiblify"

# --- Model Configuration ---
MODEL_GEMINI_PRO = "gemini-1.5-pro-preview-0514"

# --- Example Videos ---
# These are publicly accessible videos for demonstration purposes.
EXAMPLE_VIDEOS = [
    ["Confident Business Presentation", "https://storage.googleapis.com/pitchperfect-ai-examples/business_pitch_example.mp4"],
    ["Casual Tech Talk", "https://storage.googleapis.com/pitchperfect-ai-examples/tech_talk_example.mp4"],
]

# --- Schemas for Controlled Generation (as Dictionaries) ---
FEEDBACK_ITEM_SCHEMA = {
    "type": "object",
    "properties": {
        "score": {"type": "integer", "minimum": 1, "maximum": 10},
        "feedback": {"type": "string"}
    },
    "required": ["score", "feedback"]
}
HOLISTIC_ANALYSIS_SCHEMA = {
    "type": "object",
    "properties": {
        "content_analysis": {"type": "object", "properties": {"clarity": FEEDBACK_ITEM_SCHEMA, "structure": FEEDBACK_ITEM_SCHEMA, "value_proposition": FEEDBACK_ITEM_SCHEMA, "cta": FEEDBACK_ITEM_SCHEMA}},
        "vocal_analysis": {"type": "object", "properties": {"pacing": FEEDBACK_ITEM_SCHEMA, "vocal_variety": FEEDBACK_ITEM_SCHEMA, "confidence_energy": FEEDBACK_ITEM_SCHEMA, "clarity_enunciation": FEEDBACK_ITEM_SCHEMA}},
        "visual_analysis": {"type": "object", "properties": {"eye_contact": FEEDBACK_ITEM_SCHEMA, "body_language": FEEDBACK_ITEM_SCHEMA, "facial_expressions": FEEDBACK_ITEM_SCHEMA}}
    },
    "required": ["content_analysis", "vocal_analysis", "visual_analysis"]
}
FINAL_SYNTHESIS_SCHEMA = {
    "type": "object",
    "properties": {
        "key_strengths": {"type": "string"},
        "growth_opportunities": {"type": "string"},
        "executive_summary": {"type": "string"}
    },
    "required": ["key_strengths", "growth_opportunities", "executive_summary"]
}

# --- Enhanced Prompts ---
PROMPT_HOLISTIC_VIDEO_ANALYSIS = """
You are an expert sales coach. Analyze the provided video and the supplementary quantitative metrics to generate a structured, holistic feedback report. Your output MUST strictly conform to the provided JSON schema, including the 1-10 score range.

**Quantitative Metrics (for additional context):**
{quantitative_metrics_json}

**Evaluation Framework (Analyze the video directly):**
1.  **Content & Structure:** Analyze clarity, flow, value proposition, and the call to action.
2.  **Vocal Delivery:** Analyze pacing, vocal variety, confidence, energy, and enunciation. Use the quantitative metrics to inform your qualitative assessment.
3.  **Visual Delivery:** Analyze eye contact, body language, and facial expressions.

Provide specific examples from the video to support your points.
"""

PROMPT_FINAL_SYNTHESIS = """
You are a senior executive coach. Synthesize the provided detailed analysis data into a high-level summary. Your output MUST strictly conform to the provided JSON schema.

- "key_strengths" should be a single string with bullet points (e.g., "- Point one\\n- Point two").
- "growth_opportunities" should be a single string, formatted similarly.
- "executive_summary" should be a single string paragraph.

**Detailed Analysis Data:**
---
{full_analysis_json}
---
"""

# ==============================================================================
# AGENT TOOLKIT
# ==============================================================================
class YouTubeDownloaderTool:
    """A tool to download a YouTube video to a local path."""
    def run(self, url: str, output_dir: str = "temp_downloads") -> str:
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
        
        filepath = os.path.join(output_dir, f"{uuid.uuid4()}.mp4")
        ydl_opts = {
            'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
            'outtmpl': filepath,
            'quiet': True,
        }
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            ydl.download([url])
        return filepath

class QuantitativeAudioTool:
    """A tool for performing objective, numerical analysis on an audio track."""
    class JAXAudioProcessor:
        """A nested class demonstrating JAX for high-performance audio processing."""
        def __init__(self):
            self.jit_rms_energy = jax.jit(self._calculate_rms_energy)
        @staticmethod
        @jax.jit
        def _calculate_rms_energy(waveform: jnp.ndarray) -> jnp.ndarray:
            return jnp.sqrt(jnp.mean(jnp.square(waveform)))
        def analyze_energy_variation(self, waveform_np):
            if waveform_np is None or waveform_np.size == 0: return 0.0
            waveform_jnp = jnp.asarray(waveform_np)
            frame_length, hop_length = 2048, 512
            num_frames = (waveform_jnp.shape[0] - frame_length) // hop_length
            start_positions = jnp.arange(num_frames) * hop_length
            offsets = jnp.arange(frame_length)
            frame_indices = start_positions[:, None] + offsets[None, :]
            frames = waveform_jnp[frame_indices]
            frame_energies = jax.vmap(self.jit_rms_energy)(frames)
            return float(jnp.std(frame_energies))

    def __init__(self):
        self.jax_processor = self.JAXAudioProcessor()
        # Lazily load the whisper model to avoid loading it if not needed
        self._whisper_model = None

    @property
    def whisper_model(self):
        if self._whisper_model is None:
            logging.info("Loading whisper model 'base.en' for the first time...")
            self._whisper_model = whisper.load_model("base.en")
            logging.info("Whisper model loaded.")
        return self._whisper_model

    def run(self, video_path: str, output_dir: str = "temp_output"):
        if not os.path.exists(output_dir): os.makedirs(output_dir)
        video = None
        try:
            video = VideoFileClip(video_path)
            
            if video.audio is None:
                raise ValueError("The provided video file does not contain an audio track, or it could not be decoded. Analysis cannot proceed.")

            audio_path = os.path.join(output_dir, f"audio_{uuid.uuid4()}.wav")
            video.audio.write_audiofile(audio_path, codec='pcm_s16le', fps=16000, logger=None)
            
            transcript_result = self.whisper_model.transcribe(audio_path, fp16=False)
            word_count = len(transcript_result['text'].split())
            duration = video.duration
            pace = (word_count / duration) * 60 if duration > 0 else 0
            
            y, sr = librosa.load(audio_path, sr=16000)
            energy_variation = self.jax_processor.analyze_energy_variation(y)
            
            os.remove(audio_path)
            
            return {
                "speaking_pace_wpm": round(pace, 2),
                "vocal_energy_variation": round(energy_variation, 4),
            }
        finally:
            if video:
                video.close()

# ==============================================================================
# VERTEX AI MANAGER CLASS
# ==============================================================================
class VertexAIManager:
    def __init__(self):
        vertexai.init(project=GCP_PROJECT_ID, location=GCP_LOCATION)
        self.model = GenerativeModel(MODEL_GEMINI_PRO)

    def run_multimodal_analysis(self, video_gcs_uri: str, prompt: str) -> dict:
        video_part = Part.from_uri(uri=video_gcs_uri, mime_type="video/mp4")
        contents = [video_part, prompt]
        config = GenerationConfig(response_schema=HOLISTIC_ANALYSIS_SCHEMA, temperature=0.2, response_mime_type="application/json")
        response = self.model.generate_content(contents, generation_config=config)
        return json.loads(response.text)

    def run_synthesis(self, prompt: str) -> dict:
        config = GenerationConfig(response_schema=FINAL_SYNTHESIS_SCHEMA, temperature=0.3, response_mime_type="application/json")
        response = self.model.generate_content(prompt, generation_config=config)
        return json.loads(response.text)

# ==============================================================================
# AGENT CLASS
# ==============================================================================
class PitchAnalyzerAgent:
    def __init__(self):
        self.vertex_manager = VertexAIManager()
        self.storage_client = storage.Client(project=GCP_PROJECT_ID)
        self.youtube_tool = YouTubeDownloaderTool()
        self.quant_tool = QuantitativeAudioTool()
        self._check_bucket()

    def _check_bucket(self):
        logging.info(f"Checking access to GCS Bucket: {GCS_BUCKET_NAME}")
        self.storage_client.get_bucket(GCS_BUCKET_NAME)
        logging.info("GCS Bucket access confirmed.")


    def _upload_to_gcs(self, path: str) -> str:
        bucket = self.storage_client.bucket(GCS_BUCKET_NAME)
        blob_name = f"pitch-videos/{uuid.uuid4()}.mp4"
        blob = bucket.blob(blob_name)
        blob.upload_from_filename(path)
        logging.info(f"Successfully uploaded video to gs://{GCS_BUCKET_NAME}/{blob_name}")
        return f"gs://{GCS_BUCKET_NAME}/{blob_name}"

    def _delete_from_gcs(self, gcs_uri: str):
        try:
            bucket_name, blob_name = gcs_uri.replace("gs://", "").split("/", 1)
            self.storage_client.bucket(bucket_name).blob(blob_name).delete()
            logging.info(f"Successfully deleted GCS object: {gcs_uri}")
        except Exception as e:
            logging.warning(f"Failed to delete GCS object {gcs_uri}: {e}")

    def run_analysis_pipeline(self, video_path_or_url: str, progress_callback):
        local_video_path = None
        video_gcs_uri = None
        is_youtube_download = False
        try:
            if re.match(r"^(https?://)?(www\.)?(youtube\.com|youtu\.?be)/.+$", video_path_or_url):
                progress_callback(0.1, "Downloading video from YouTube...")
                local_video_path = self.youtube_tool.run(video_path_or_url)
                is_youtube_download = True
            else:
                local_video_path = video_path_or_url

            progress_callback(0.3, "Performing JAX-based quantitative analysis...")
            quant_metrics = self.quant_tool.run(local_video_path)

            progress_callback(0.5, "Uploading video to secure Cloud Storage...")
            video_gcs_uri = self._upload_to_gcs(local_video_path)

            progress_callback(0.7, "Gemini 1.5 Pro is analyzing the video...")
            analysis_prompt = PROMPT_HOLISTIC_VIDEO_ANALYSIS.format(quantitative_metrics_json=json.dumps(quant_metrics, indent=2))
            multimodal_analysis = self.vertex_manager.run_multimodal_analysis(video_gcs_uri, analysis_prompt)

            progress_callback(0.9, "Synthesizing final report...")
            synthesis_prompt = PROMPT_FINAL_SYNTHESIS.format(full_analysis_json=json.dumps(multimodal_analysis, indent=2))
            final_summary = self.vertex_manager.run_synthesis(synthesis_prompt)
            
            return {"quantitative_metrics": quant_metrics, "multimodal_analysis": multimodal_analysis, "executive_summary": final_summary}
        except Exception as e:
            logging.error(f"Analysis pipeline failed: {e}", exc_info=True)
            return {"error": str(e)}
        finally:
            if video_gcs_uri:
                self._delete_from_gcs(video_gcs_uri)
            # Only delete the local file if it was downloaded from YouTube
            if is_youtube_download and local_video_path and os.path.exists(local_video_path):
                os.remove(local_video_path)
                logging.info(f"Deleted temporary YouTube download: {local_video_path}")

# ==============================================================================
# UI FORMATTING HELPER
# ==============================================================================
def format_feedback_markdown(analysis: dict) -> str:
    if not analysis or "error" in analysis:
        error_message = analysis.get('error', 'Unknown error.')
        logging.error(f"Displaying analysis failure to user: {error_message}")
        return f"## Analysis Failed 😞\n\n**Reason:** {error_message}"
    
    summary = analysis.get('executive_summary', {})
    metrics = analysis.get('quantitative_metrics', {})
    ai_analysis = analysis.get('multimodal_analysis', {})
    
    def get_pace_rating(wpm):
        if wpm == 0: return "N/A (No speech detected)"
        if wpm < 120: return "Slow / Deliberate"
        if wpm <= 160: return "Conversational"
        return "Fast-Paced"

    def get_energy_rating(variation):
        if variation == 0: return "N/A"
        if variation < 0.02: return "Consistent / Monotonous"
        if variation <= 0.05: return "Moderately Dynamic"
        return "Highly Dynamic & Engaging"
        
    wpm = metrics.get('speaking_pace_wpm', 0)
    energy_var = metrics.get('vocal_energy_variation', 0)
    pace_rating = get_pace_rating(wpm)
    energy_rating = get_energy_rating(energy_var)

    metrics_md = f"""
- **Speaking Pace:** **{wpm} WPM** *(Rating: {pace_rating})*
  - *This measures the number of words spoken per minute. A typical conversational pace is between 120-160 WPM.*
- **Vocal Energy Variation:** **{energy_var:.4f}** *(Rating: {energy_rating})*
  - *This measures the standard deviation of your vocal loudness. A higher value indicates a more dynamic and engaging vocal range, while a very low value suggests a monotonous delivery.*
    """

    def format_ai_item(title, data):
        if not data or "score" not in data: return f"**{title}:**\n> Analysis not available.\n\n"
        raw_score = data.get('score', 0); score = max(1, min(10, raw_score))
        stars = "🟢" * score + "⚪️" * (10 - score)
        feedback = data.get('feedback', 'No feedback.').replace('\n', '\n> ')
        return f"**{title}:** `{stars} [{score}/10]`\n\n> {feedback}\n\n"
    
    content = ai_analysis.get('content_analysis', {}); vocal = ai_analysis.get('vocal_analysis', {}); visual = ai_analysis.get('visual_analysis', {})

    return f"""
# PitchPerfect AI Analysis Report 📊
## 🏆 Executive Summary
### Key Strengths
{summary.get('key_strengths', '- N/A')}
### High-Leverage Growth Opportunities
{summary.get('growth_opportunities', '- N/A')}
### Final Verdict
> {summary.get('executive_summary', 'N/A')}
---
## 📈 Quantitative Metrics Explained (via JAX & Whisper)
{metrics_md}
---
## 🧠 AI Multimodal Analysis (via Gemini 1.5 Pro)
### I. Content & Structure
{format_ai_item("Clarity", content.get('clarity'))}
{format_ai_item("Structure & Flow", content.get('structure'))}
{format_ai_item("Value Proposition", content.get('value_proposition'))}
{format_ai_item("Call to Action (CTA)", content.get('cta'))}
<hr style="border:1px solid #ddd">

### II. Vocal Delivery
{format_ai_item("Pacing", vocal.get('pacing'))}
{format_ai_item("Vocal Variety", vocal.get('vocal_variety'))}
{format_ai_item("Confidence & Energy", vocal.get('confidence_energy'))}
{format_ai_item("Clarity & Enunciation", vocal.get('clarity_enunciation'))}
<hr style="border:1px solid #ddd">

### III. Visual Delivery
{format_ai_item("Eye Contact", visual.get('eye_contact'))}
{format_ai_item("Body Language", visual.get('body_language'))}
{format_ai_item("Facial Expressions", visual.get('facial_expressions'))}
"""

# ==============================================================================
# GRADIO APPLICATION
# ==============================================================================
if __name__ == "__main__":
    pitch_agent = None
    try:
        # Initialize the agent only if the script is run directly.
        pitch_agent = PitchAnalyzerAgent()
    except Exception as e:
        logging.fatal(f"Failed to initialize agent during startup: {e}", exc_info=True)
        # Display the error in a simplified Gradio interface if initialization fails
        with gr.Blocks(theme=gr.themes.Soft()) as demo:
            gr.Markdown(f"""
            # ## 🔴 FATAL ERROR
            Could not initialize the PitchPerfect AI Agent. This is likely due to a configuration issue.
            Please check the following:
            1.  You have authenticated with `gcloud auth application-default login`.
            2.  The GCP Project ID (`{GCP_PROJECT_ID}`) and GCS Bucket (`{GCS_BUCKET_NAME}`) are correct and accessible.
            3.  The necessary APIs (Vertex AI, Cloud Storage) are enabled in your project.
            
            **Error Details:**
            ```
            {e}
            ```
            """)
        demo.launch()
        sys.exit(1)


    def run_analysis_pipeline_interface(video_path, url_path, progress=gr.Progress(track_tqdm=True)):
        """Interface function for Gradio to call the agent's pipeline."""
        if not pitch_agent:
            return "## FATAL ERROR: Application not initialized. Check logs and configuration."
        
        input_path = url_path if url_path and url_path.strip() else video_path
        if not input_path:
            return "## No Video Provided\nPlease upload a video file or enter a valid YouTube URL to begin."
        
        # Clear the other input field to avoid confusion
        if url_path:
            video_path = None
        else:
            url_path = None
        
        analysis_result = pitch_agent.run_analysis_pipeline(input_path, progress.update)
        return format_feedback_markdown(analysis_result)

    # Define the Gradio UI
    with gr.Blocks(theme=gr.themes.Soft(primary_hue="teal", secondary_hue="orange")) as demo:
        gr.Markdown("# **PitchPerfect AI**: Your Enterprise-Grade Sales Coach 🚀")
        with gr.Row():
            with gr.Column(scale=1):
                video_uploader = gr.Video(label="Upload Your Pitch", sources=["upload"])
                gr.Markdown("<center>--- **OR** ---</center>")
                youtube_url = gr.Textbox(label="Enter YouTube URL", placeholder="e.g., https://www.youtube.com/watch?v=...")
                analyze_button = gr.Button("Analyze My Pitch 🧠", variant="primary")
                gr.Examples(examples=EXAMPLE_VIDEOS, inputs=youtube_url, label="Example Pitches (Click to Use)")
            with gr.Column(scale=2):
                analysis_output = gr.Markdown(label="Your Feedback Report", value="### Your detailed report will appear here...")
        
        # Connect the button click to the analysis function
        analyze_button.click(
            fn=run_analysis_pipeline_interface,
            inputs=[video_uploader, youtube_url],
            outputs=analysis_output
        )

    # Launch the Gradio application
    demo.launch(debug=True, share=True)