Building a Paid Service with the Gemini Multimodal API: Image, Audio, and Video Processing

# file_upload.py
import google.generativeai as genai
import time
from pathlib import Path
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
def upload_file(file_path: str) -> genai.types.File:
    """Upload a file to the Gemini File API and wait for processing to complete."""
    
    path = Path(file_path)
    
    mime_map = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".webp": "image/webp",
        ".mp3": "audio/mp3",
        ".wav": "audio/wav",
        ".m4a": "audio/m4a",
        ".mp4": "video/mp4",
        ".mov": "video/quicktime",
        ".pdf": "application/pdf",
    }
    mime_type = mime_map.get(path.suffix.lower(), "application/octet-stream")
    
    print(f"Uploading {path.name} ({mime_type})...")
    
    uploaded_file = genai.upload_file(
        path=str(path),
        mime_type=mime_type,
        display_name=path.name,
    )
    
    # Video and audio files may require server-side processing time
    while uploaded_file.state.name == "PROCESSING":
        time.sleep(2)
        uploaded_file = genai.get_file(uploaded_file.name)
    
    if uploaded_file.state.name == "FAILED":
        raise ValueError(f"File upload failed: {uploaded_file.name}")
    
    print(f"Upload complete: {uploaded_file.uri}")
    return uploaded_file
 
def delete_file_after_processing(file: genai.types.File):
    """Delete the file from File API after processing (storage hygiene)."""
    genai.delete_file(file.name)
    print(f"Deleted: {file.name}")

# real_estate_scorer.py
from dataclasses import dataclass
from typing import List
import json
 
@dataclass
class PhotoScore:
    score: int            # 1-100
    strengths: List[str]
    improvements: List[str]
    estimated_impact: str
 
@handle_gemini_errors
def score_real_estate_photo(image_path: str) -> PhotoScore:
    """Score a real estate photo and suggest improvements."""
    
    uploaded_file = upload_file(image_path)
    model = genai.GenerativeModel("gemini-2.5-pro")
    
    prompt = """
    Evaluate this real estate listing photo from the perspective of a professional photographer and real estate marketer.
 
    Respond in the following JSON format:
    {
      "score": integer from 1 to 100,
      "strengths": ["up to 3 strengths"],
      "improvements": ["up to 3 specific improvements"],
      "estimated_impact": "one sentence on how improvements would affect listing performance"
    }
 
    Scoring criteria:
    - Brightness and contrast (20 points)
    - Composition and angle (20 points)
    - Tidiness and presentation (20 points)
    - Features and finishes showcase (20 points)
    - Overall visual appeal (20 points)
    """
    
    response = model.generate_content([prompt, uploaded_file])
    delete_file_after_processing(uploaded_file)
    
    text = response.text
    start = text.find("{")
    end = text.rfind("}") + 1
    data = json.loads(text[start:end])
    
    return PhotoScore(**data)
 
# Example output
score = score_real_estate_photo("living_room.jpg")
print(f"Score: {score.score}/100")
print(f"Strengths: {', '.join(score.strengths)}")
print(f"Improvements: {', '.join(score.improvements)}")

# meeting_analyzer.py
from dataclasses import dataclass
from typing import List
 
@dataclass
class MeetingAnalysis:
    transcript: str
    summary: str
    action_items: List[str]
    key_decisions: List[str]
    billed_minutes: int
 
def analyze_meeting_audio(audio_path: str, duration_minutes: float) -> MeetingAnalysis:
    """Transcribe meeting audio and extract summary, action items, and decisions."""
    
    uploaded_file = upload_file(audio_path)
    
    # Step 1: Transcription with Flash (fast and cheap)
    flash_model = genai.GenerativeModel("gemini-2.5-flash")
    transcript_response = flash_model.generate_content([
        "Transcribe this audio accurately. If multiple speakers are present, label them as [Speaker A]:, [Speaker B]:, etc.",
        uploaded_file,
    ])
    transcript = transcript_response.text
    
    # Step 2: Analysis with Pro (higher quality summarization)
    pro_model = genai.GenerativeModel("gemini-2.5-pro")
    analysis_response = pro_model.generate_content(f"""
    Analyze this meeting transcript and extract structured information.
 
    ## Transcript
    {transcript}
 
    Respond with:
 
    ## Looking back (3 sentences max)
    [concise summary]
 
    ## Action Items
    - [Owner if identifiable]: [task]
 
    ## Key Decisions
    - [decision made]
    """)
    
    delete_file_after_processing(uploaded_file)
    
    analysis_text = analysis_response.text
    
    return MeetingAnalysis(
        transcript=transcript,
        summary=extract_section(analysis_text, "Summary"),
        action_items=extract_list(analysis_text, "Action Items"),
        key_decisions=extract_list(analysis_text, "Key Decisions"),
        billed_minutes=max(1, round(duration_minutes)),
    )
 
def extract_section(text: str, header: str) -> str:
    lines = text.split("\n")
    in_section, result = False, []
    for line in lines:
        if header in line and line.startswith("##"):
            in_section = True
            continue
        if in_section and line.startswith("##"):
            break
        if in_section:
            result.append(line)
    return "\n".join(result).strip()
 
def extract_list(text: str, header: str) -> list:
    section = extract_section(text, header)
    return [l.lstrip("- ").strip() for l in section.split("\n") if l.strip().startswith("-")]

# pdf_qa.py
from dataclasses import dataclass
 
@dataclass
class QAResult:
    answer: str
    confidence: str       # "high" / "medium" / "low"
    source_hints: list[str]
 
class PDFQAService:
    def __init__(self):
        self.model = genai.GenerativeModel("gemini-2.5-pro")
        self._cached_files: dict[str, genai.types.File] = {}
    
    def load_pdf(self, pdf_path: str, doc_id: str):
        """Upload PDF and cache the reference for subsequent questions."""
        if doc_id not in self._cached_files:
            self._cached_files[doc_id] = upload_file(pdf_path)
        return self._cached_files[doc_id]
    
    @handle_gemini_errors
    def ask(self, doc_id: str, question: str) -> QAResult:
        """Ask a natural language question about an uploaded PDF."""
        
        if doc_id not in self._cached_files:
            raise ValueError(f"Document {doc_id} not loaded. Call load_pdf() first.")
        
        pdf_file = self._cached_files[doc_id]
        
        response = self.model.generate_content([
            f"""Answer the following question based solely on the content of this PDF document.
 
            Important instructions:
            - If the answer is not in the document, say "This information is not in the document" — do not guess
            - Rate your confidence as "high", "medium", or "low"
            - Include hints about where in the document you found the answer (page numbers, section headings)
 
            Question: {question}
 
            Respond in JSON:
            {{
              "answer": "your answer",
              "confidence": "high/medium/low",
              "source_hints": ["location hints"]
            }}
            """,
            pdf_file,
        ])
        
        import json
        text = response.text
        data = json.loads(text[text.find("{"):text.rfind("}")+1])
        return QAResult(**data)
    
    def cleanup(self, doc_id: str):
        if doc_id in self._cached_files:
            delete_file_after_processing(self._cached_files[doc_id])
            del self._cached_files[doc_id]

# stripe_metered_billing.py
import stripe
from dataclasses import dataclass
import time
 
stripe.api_key = "YOUR_STRIPE_SECRET_KEY"
 
@dataclass
class UsageRecord:
    customer_id: str
    subscription_item_id: str
    quantity: int
    unit: str  # e.g., "image_analysis", "audio_minutes", "pdf_qa"
 
def report_usage_to_stripe(record: UsageRecord) -> stripe.UsageRecord:
    """Report usage to Stripe Metered Billing for end-of-period invoicing."""
    
    usage_record = stripe.SubscriptionItem.create_usage_record(
        record.subscription_item_id,
        quantity=record.quantity,
        timestamp=int(time.time()),
        action="increment",
    )
    
    print(f"Billed: {record.quantity} {record.unit} for {record.customer_id}")
    return usage_record
 
class BilledMultimodalService:
    """Multimodal processing with integrated Stripe billing."""
    
    def __init__(self):
        self.flash = genai.GenerativeModel("gemini-2.5-flash")
        self.pro = genai.GenerativeModel("gemini-2.5-pro")
    
    @handle_gemini_errors
    def analyze_image(
        self,
        image_path: str,
        prompt: str,
        customer_id: str,
        subscription_item_id: str,
        use_pro: bool = False,
    ) -> dict:
        """Analyze an image and record usage to Stripe."""
        
        uploaded = upload_file(image_path)
        model = self.pro if use_pro else self.flash
        response = model.generate_content([prompt, uploaded])
        delete_file_after_processing(uploaded)
        
        # Report 1 unit per image
        report_usage_to_stripe(UsageRecord(
            customer_id=customer_id,
            subscription_item_id=subscription_item_id,
            quantity=1,
            unit="image_analysis",
        ))
        
        return {
            "result": response.text,
            "model": "gemini-2.5-pro" if use_pro else "gemini-2.5-flash",
            "billed": True,
        }
    
    @handle_gemini_errors
    def transcribe_audio(
        self,
        audio_path: str,
        duration_minutes: float,
        customer_id: str,
        subscription_item_id: str,
    ) -> dict:
        """Transcribe audio and report per-minute usage to Stripe."""
        
        uploaded = upload_file(audio_path)
        response = self.flash.generate_content([
            "Transcribe this audio in full. Label multiple speakers as [Speaker A]:, [Speaker B]:, etc.",
            uploaded,
        ])
        delete_file_after_processing(uploaded)
        
        billed_minutes = max(1, round(duration_minutes))
        report_usage_to_stripe(UsageRecord(
            customer_id=customer_id,
            subscription_item_id=subscription_item_id,
            quantity=billed_minutes,
            unit="audio_minutes",
        ))
        
        return {
            "transcript": response.text,
            "billed_minutes": billed_minutes,
        }

# error_handling.py
import google.api_core.exceptions as google_exceptions
from functools import wraps
import logging
 
logger = logging.getLogger(__name__)
 
class MultimodalServiceError(Exception):
    def __init__(self, message: str, user_message: str, retryable: bool = False):
        super().__init__(message)
        self.user_message = user_message
        self.retryable = retryable
 
def handle_gemini_errors(func):
    """Decorator that translates Gemini API errors into service-level errors."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        
        except google_exceptions.InvalidArgument as e:
            error_str = str(e).lower()
            
            if "file type" in error_str or "mime" in error_str:
                raise MultimodalServiceError(
                    message=f"Unsupported file type: {e}",
                    user_message="This file format is not supported. Please use JPG, PNG, MP3, MP4, or PDF.",
                    retryable=False,
                )
            if "file size" in error_str or "too large" in error_str:
                raise MultimodalServiceError(
                    message=f"File too large: {e}",
                    user_message="File exceeds the 2GB size limit. Please split the file and try again.",
                    retryable=False,
                )
            raise MultimodalServiceError(
                message=f"Invalid request: {e}",
                user_message="The request format was invalid.",
                retryable=False,
            )
        
        except google_exceptions.ResourceExhausted as e:
            logger.warning(f"Rate limited: {e}")
            raise MultimodalServiceError(
                message=f"Rate limited: {e}",
                user_message="The service is temporarily busy. Please try again in a moment.",
                retryable=True,
            )
        
        except google_exceptions.DeadlineExceeded as e:
            logger.error(f"Timeout: {e}")
            raise MultimodalServiceError(
                message=f"Processing timeout: {e}",
                user_message="Processing timed out. For large files, try splitting into smaller segments.",
                retryable=True,
            )
        
        except Exception as e:
            logger.exception(f"Unexpected error in {func.__name__}: {e}")
            raise MultimodalServiceError(
                message=f"Unexpected error: {e}",
                user_message="An unexpected error occurred. Please try again later.",
                retryable=False,
            )
    
    return wrapper