app.py

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

# ==============================================================================
# File: README.md (Instructions)
# ==============================================================================
"""
# 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.

## 🔑 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.  **Install Dependencies:**
    Create a `requirements.txt` file with the content below and run `pip install -r requirements.txt`.
    ```
    gradio
    google-cloud-aiplatform
    google-cloud-storage
    moviepy
    # For JAX and Quantitative Analysis
    jax
    jaxlib
    librosa
    speechrecognition
    openai-whisper
    # For YouTube support
    yt-dlp
    ```

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

5.  **Run the Application:**
    ```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 logging
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-002"

# --- 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()
        self.whisper_model = whisper.load_model("base.en")

    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)
            
            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):
        self.storage_client.get_bucket(GCS_BUCKET_NAME)

    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)
        return f"gs://{GCS_BUCKET_NAME}/{blob_name}"

    def _delete_from_gcs(self, gcs_uri: str):
        bucket_name, blob_name = gcs_uri.replace("gs://", "").split("/", 1)
        self.storage_client.bucket(bucket_name).blob(blob_name).delete()

    def run_analysis_pipeline(self, video_path_or_url: str, progress_callback):
        local_video_path = None
        video_gcs_uri = None
        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)
            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:
                try: self._delete_from_gcs(video_gcs_uri)
                except Exception as e: logging.warning(f"Failed to delete GCS object {video_gcs_uri}: {e}")
            if local_video_path and video_path_or_url != local_video_path:
                if os.path.exists(local_video_path): os.remove(local_video_path)

# ==============================================================================
# UI FORMATTING HELPER
# ==============================================================================
def format_feedback_markdown(analysis: dict) -> str:
    if not analysis or "error" in analysis:
        return f"## Analysis Failed 😞\n\n**Reason:** {analysis.get('error', 'Unknown error.')}"
    
    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.*
    """

    # --- FIX: Revert to using bold text instead of headers for consistency ---
    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', {})

    # --- FIX: Use a more consistent structure for the final report ---
    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)
{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__":
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    pitch_agent = None
    try:
        pitch_agent = PitchAnalyzerAgent()
    except Exception as e:
        logging.fatal(f"Failed to initialize agent during startup: {e}", exc_info=True)

    def run_analysis_pipeline(video_path, url_path, progress=gr.Progress(track_tqdm=True)):
        if not pitch_agent: return "## FATAL ERROR: Application not initialized. Check logs and config."
        input_path = url_path if url_path else video_path
        if not input_path: return "## No Video Provided. Please upload a video or enter a YouTube URL."
        
        analysis_result = pitch_agent.run_analysis_pipeline(input_path, progress)
        return format_feedback_markdown(analysis_result)

    with gr.Blocks(theme=gr.themes.Soft(primary_hue="teal", secondary_hue="orange")) as demo:
        gr.Markdown("# **Video Analysis 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("--- **OR** ---")
                youtube_url = gr.Textbox(label="Enter YouTube URL")
                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...")
        analyze_button.click(fn=run_analysis_pipeline, inputs=[video_uploader, youtube_url], outputs=analysis_output)

    if pitch_agent:
        demo.launch(debug=True, share=True)
    else:
        print("\n" + "="*80 + "\nCOULD NOT START GRADIO APP: Agent failed to initialize.\n" + "="*80)