Facebook Ads Library MCP Server

import os
import sys
import logging
import google.generativeai as genai
from google.generativeai.types import File
from typing import Optional, List, Dict, Any

# Set up logger
logger = logging.getLogger(__name__)

GEMINI_API_KEY = None

def get_gemini_api_key() -> str:
    """
    Get Gemini API key from command line arguments or environment variable.
    Caches the key in memory after first read.
    Priority: command line argument > environment variable

    Returns:
        str: The Gemini API key.

    Raises:
        Exception: If no key is provided in command line arguments or environment.
    """
    global GEMINI_API_KEY
    if GEMINI_API_KEY is None:
        # Try command line argument first
        if "--gemini-api-key" in sys.argv:
            token_index = sys.argv.index("--gemini-api-key") + 1
            if token_index < len(sys.argv):
                GEMINI_API_KEY = sys.argv[token_index]
                logger.info(f"Using Gemini API key from command line arguments")
            else:
                raise Exception("--gemini-api-key argument provided but no key value followed it")
        # Try environment variable
        elif os.getenv("GEMINI_API_KEY"):
            GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
            logger.info(f"Using Gemini API key from environment variable")
        else:
            raise Exception("Gemini API key must be provided via '--gemini-api-key' command line argument or 'GEMINI_API_KEY' environment variable")

    return GEMINI_API_KEY


def configure_gemini() -> genai.GenerativeModel:
    """
    Configure Gemini API with the API key and return a model instance.
    
    Returns:
        genai.GenerativeModel: Configured Gemini model instance for video analysis
    """
    api_key = get_gemini_api_key()
    genai.configure(api_key=api_key)
    
    # Use Gemini 2.0 Flash for video analysis (more cost-effective than Pro)
    model = genai.GenerativeModel('gemini-2.0-flash-exp')
    
    logger.info("Gemini API configured successfully")
    return model


def upload_video_to_gemini(video_path: str) -> File:
    """
    Upload a video file to Gemini File API for analysis.
    
    Args:
        video_path: Path to the video file to upload
        
    Returns:
        genai.File: The uploaded file object for use in analysis
        
    Raises:
        Exception: If upload fails
    """
    try:
        # Upload video file
        video_file = genai.upload_file(path=video_path)
        
        # Wait for processing to complete
        import time
        while video_file.state.name == "PROCESSING":
            time.sleep(2)
            video_file = genai.get_file(video_file.name)
        
        if video_file.state.name == "FAILED":
            raise Exception(f"Video processing failed: {video_file.state}")
            
        logger.info(f"Video uploaded successfully: {video_file.name}")
        return video_file
        
    except Exception as e:
        logger.error(f"Failed to upload video to Gemini: {str(e)}")
        raise


def analyze_video_with_gemini(model: genai.GenerativeModel, video_file: File, prompt: str) -> str:
    """
    Analyze a video using Gemini with a custom prompt.
    
    Args:
        model: Configured Gemini model instance
        video_file: Uploaded video file from Gemini File API
        prompt: Analysis prompt for the video
        
    Returns:
        str: Analysis results from Gemini
        
    Raises:
        Exception: If analysis fails
    """
    try:
        # Generate analysis
        response = model.generate_content([video_file, prompt])
        
        if not response.text:
            raise Exception("Gemini returned empty response")
            
        logger.info("Video analysis completed successfully")
        return response.text
        
    except Exception as e:
        logger.error(f"Video analysis failed: {str(e)}")
        raise


def analyze_videos_batch_with_gemini(model: genai.GenerativeModel, video_files: List[File], prompt_template: str, video_contexts: List[Dict[str, Any]]) -> List[str]:
    """
    Analyze multiple videos using Gemini in a single request for token efficiency.
    
    Args:
        model: Configured Gemini model instance
        video_files: List of uploaded video files from Gemini File API
        prompt_template: Base analysis prompt template
        video_contexts: List of context dicts with brand_name, ad_id, etc. for each video
        
    Returns:
        List[str]: Analysis results for each video in order
        
    Raises:
        Exception: If batch analysis fails
    """
    try:
        if not video_files or len(video_files) != len(video_contexts):
            raise Exception("Video files and contexts must have matching lengths")
        
        # Create batch prompt with multiple videos
        batch_prompt = f"""Analyze the following {len(video_files)} Facebook ad videos. For each video, provide analysis following this format:

{prompt_template}

Please analyze each video separately and clearly label each analysis as "VIDEO 1:", "VIDEO 2:", etc.

"""
        
        # Add context information for each video
        for i, context in enumerate(video_contexts, 1):
            brand_info = f" (Brand: {context.get('brand_name', 'Unknown')})" if context.get('brand_name') else ""
            ad_info = f" (Ad ID: {context.get('ad_id', 'Unknown')})" if context.get('ad_id') else ""
            batch_prompt += f"VIDEO {i}{brand_info}{ad_info}:\n"
        
        # Combine all video files with the prompt
        content_parts = [batch_prompt] + video_files
        
        # Generate batch analysis
        response = model.generate_content(content_parts)
        
        if not response.text:
            raise Exception("Gemini returned empty response for batch analysis")
        
        # Split response by video markers
        analysis_text = response.text
        video_analyses = []
        
        # Parse individual video analyses
        for i in range(1, len(video_files) + 1):
            video_marker = f"VIDEO {i}:"
            next_marker = f"VIDEO {i + 1}:" if i < len(video_files) else None
            
            start_idx = analysis_text.find(video_marker)
            if start_idx == -1:
                logger.warning(f"Could not find analysis for VIDEO {i}")
                video_analyses.append(f"Analysis not found in batch response for video {i}")
                continue
                
            start_idx += len(video_marker)
            
            if next_marker:
                end_idx = analysis_text.find(next_marker)
                individual_analysis = analysis_text[start_idx:end_idx].strip() if end_idx != -1 else analysis_text[start_idx:].strip()
            else:
                individual_analysis = analysis_text[start_idx:].strip()
            
            video_analyses.append(individual_analysis)
        
        logger.info(f"Batch video analysis completed successfully for {len(video_files)} videos")
        return video_analyses
        
    except Exception as e:
        logger.error(f"Batch video analysis failed: {str(e)}")
        raise


def upload_videos_batch_to_gemini(video_paths: List[str]) -> List[File]:
    """
    Upload multiple video files to Gemini File API for batch analysis.
    
    Args:
        video_paths: List of paths to video files to upload
        
    Returns:
        List[genai.File]: List of uploaded file objects for use in analysis
        
    Raises:
        Exception: If any upload fails
    """
    uploaded_files = []
    failed_uploads = []
    
    try:
        for i, video_path in enumerate(video_paths):
            try:
                # Upload video file
                video_file = genai.upload_file(path=video_path)
                
                # Wait for processing to complete
                import time
                while video_file.state.name == "PROCESSING":
                    time.sleep(2)
                    video_file = genai.get_file(video_file.name)
                
                if video_file.state.name == "FAILED":
                    failed_uploads.append(f"Video {i+1}: {video_file.state}")
                    continue
                    
                uploaded_files.append(video_file)
                logger.info(f"Video {i+1} uploaded successfully: {video_file.name}")
                
            except Exception as e:
                failed_uploads.append(f"Video {i+1}: {str(e)}")
                logger.error(f"Failed to upload video {i+1} at {video_path}: {str(e)}")
        
        if failed_uploads:
            error_msg = f"Some video uploads failed: {'; '.join(failed_uploads)}"
            if not uploaded_files:  # All uploads failed
                raise Exception(error_msg)
            else:  # Partial failure
                logger.warning(error_msg)
        
        return uploaded_files
        
    except Exception as e:
        # Cleanup any successfully uploaded files on total failure
        for uploaded_file in uploaded_files:
            try:
                cleanup_gemini_file(uploaded_file.name)
            except:
                pass
        raise


def cleanup_gemini_files_batch(file_names: List[str]):
    """
    Delete multiple files from Gemini File API to free up storage.
    
    Args:
        file_names: List of file names to delete
    """
    for file_name in file_names:
        try:
            genai.delete_file(file_name)
            logger.info(f"Cleaned up Gemini file: {file_name}")
        except Exception as e:
            logger.warning(f"Failed to cleanup Gemini file {file_name}: {str(e)}")


def cleanup_gemini_file(file_name: str):
    """
    Delete a file from Gemini File API to free up storage.
    
    Args:
        file_name: Name of the file to delete
    """
    try:
        genai.delete_file(file_name)
        logger.info(f"Cleaned up Gemini file: {file_name}")
    except Exception as e:
        logger.warning(f"Failed to cleanup Gemini file {file_name}: {str(e)}")