app.py

# === Gradio Demo App: gradio_app.py ===
# This script creates a user-friendly web interface to demonstrate the
# multimodal moderation capabilities of the main FastAPI server.
#
# It interacts with the /v3/moderations endpoint.
# --------------------------------------------------------------------

import base64
import os
import json
import logging

import gradio as gr
import httpx
from dotenv import load_dotenv

# --- Configuration ---
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
load_dotenv()

# The URL of your running FastAPI server.
# It's crucial to set this in your .env file for deployment.
API_BASE_URL = os.environ.get("API_BASE_URL", "")
MODERATION_ENDPOINT = f"{API_BASE_URL}/v3/moderations"

# --- Full list of Whisper V3 supported languages ---
# Mapping user-friendly names to ISO 639-1 codes
WHISPER_LANGUAGES = {
    "English": "en", "Chinese": "zh", "German": "de", "Spanish": "es", "Russian": "ru",
    "Korean": "ko", "French": "fr", "Japanese": "ja", "Portuguese": "pt", "Turkish": "tr",
    "Polish": "pl", "Catalan": "ca", "Dutch": "nl", "Arabic": "ar", "Swedish": "sv",
    "Italian": "it", "Indonesian": "id", "Hindi": "hi", "Finnish": "fi", "Vietnamese": "vi",
    "Hebrew": "he", "Ukrainian": "uk", "Greek": "el", "Malay": "ms", "Czech": "cs",
    "Romanian": "ro", "Danish": "da", "Hungarian": "hu", "Tamil": "ta", "Norwegian": "no",
    "Thai": "th", "Urdu": "ur", "Croatian": "hr", "Bulgarian": "bg", "Lithuanian": "lt",
    "Latin": "la", "Maori": "mi", "Malayalam": "ml", "Welsh": "cy", "Slovak": "sk",
    "Telugu": "te", "Persian": "fa", "Latvian": "lv", "Bengali": "bn", "Serbian": "sr",
    "Azerbaijani": "az", "Slovenian": "sl", "Kannada": "kn", "Estonian": "et", "Macedonian": "mk",
    "Breton": "br", "Basque": "eu", "Icelandic": "is", "Armenian": "hy", "Nepali": "ne",
    "Mongolian": "mn", "Bosnian": "bs", "Kazakh": "kk", "Albanian": "sq", "Swahili": "sw",
    "Galician": "gl", "Marathi": "mr", "Punjabi": "pa", "Sinhala": "si", "Khmer": "km",
    "Shona": "sn", "Yoruba": "yo", "Somali": "so", "Afrikaans": "af", "Occitan": "oc",
    "Georgian": "ka", "Belarusian": "be", "Tajik": "tg", "Sindhi": "sd", "Gujarati": "gu",
    "Amharic": "am", "Yiddish": "yi", "Lao": "lo", "Uzbek": "uz", "Faroese": "fo",
    "Haitian Creole": "ht", "Pashto": "ps", "Turkmen": "tk", "Nynorsk": "nn", "Maltese": "mt",
    "Sanskrit": "sa", "Luxembourgish": "lb", "Myanmar (Burmese)": "my", "Tibetan": "bo",
    "Tagalog": "tl", "Malagasy": "mg", "Assamese": "as", "Tatar": "tt", "Hawaiian": "haw",
    "Lingala": "ln", "Hausa": "ha", "Bashkir": "ba", "Javanese": "jw", "Sundanese": "su",
}
# Sort languages alphabetically for the dropdown
SORTED_LANGUAGES = dict(sorted(WHISPER_LANGUAGES.items()))


# --- Helper Function ---
def file_to_base64(filepath: str) -> str:
    """Reads a file and converts it to a base64 encoded string."""
    if not filepath:
        return None
    try:
        with open(filepath, "rb") as f:
            encoded_string = base64.b64encode(f.read()).decode("utf-8")
        return encoded_string
    except Exception as e:
        logging.error(f"Failed to convert file {filepath} to base64: {e}")
        return None

# --- Core Logic ---
def moderate_content(text_input, image_input, video_input, audio_input, language_full_name):
    """
    Prepares the payload, calls the moderation API, and formats the response.
    """
    if not any([text_input, image_input, video_input, audio_input]):
        return "Please provide at least one input (text, image, video, or audio).", None

    logging.info("Preparing payload for moderation API...")
    payload = {
        "model": "nai-moderation-latest" # This is the model name expected by our API
    }

    if text_input:
        payload["input"] = text_input
    
    image_b64 = file_to_base64(image_input)
    if image_b64:
        payload["image"] = image_b64

    video_b64 = file_to_base64(video_input)
    if video_b64:
        payload["video"] = video_b64

    audio_b64 = file_to_base64(audio_input)
    if audio_b64:
        payload["voice"] = audio_b64
        language_code = SORTED_LANGUAGES.get(language_full_name, "en") # Default to 'en'
        payload["language"] = language_code
        logging.info(f"Audio detected. Using language: {language_full_name} ({language_code})")

    logging.info(f"Sending request to {MODERATION_ENDPOINT} with inputs: {list(payload.keys())}")
    
    summary_output = "An error occurred. Please check the logs."
    full_response_output = {}

    try:
        with httpx.Client(timeout=180.0) as client:
            response = client.post(MODERATION_ENDPOINT, json=payload)
            response.raise_for_status() # Raises HTTPStatusError for 4xx/5xx responses
            
            data = response.json()
            full_response_output = data
            
            if not data.get("results"):
                summary_output = "API returned an empty result. This might happen if media processing fails (e.g., a video with no frames)."
                return summary_output, full_response_output

            result = data["results"][0]
            
            status = "🚨 FLAGGED 🚨" if result["flagged"] else "✅ SAFE ✅"
            reason = result.get("reason") or "N/A"
            transcribed = result.get("transcribed_text") or "N/A"
            flagged_categories = [cat for cat, flagged in result.get("categories", {}).items() if flagged]
            categories_str = ", ".join(flagged_categories) if flagged_categories else "None"

            summary_output = f"""
            **Moderation Status:** {status}
            ---
            **Reason:** {reason}
            ---
            **Flagged Categories:** {categories_str}
            ---
            **Transcribed Text (from audio):**
            {transcribed}
            """
            logging.info("Successfully received and parsed moderation response.")

    # --- MODIFIED: Enhanced Error Handling ---
    except httpx.HTTPStatusError as e:
        # Catches errors returned by the backend API (e.g., 422, 500)
        user_message = "The moderation service returned an error."
        error_details = ""
        try:
            # Try to parse the JSON error response from the server
            error_json = e.response.json()
            detail = error_json.get("detail", "No specific error detail provided.")
            error_details = f"**Reason:** {detail}"
            full_response_output = {"error": "Backend API Error", "status_code": e.response.status_code, "details": error_json}
        except (json.JSONDecodeError, AttributeError):
            # Fallback for non-JSON or unexpected error formats
            error_details = f"**Raw Server Response:**\n```\n{e.response.text}\n```"
            full_response_output = {"error": "Backend API Error", "status_code": e.response.status_code, "details": e.response.text}

        summary_output = f"""
        **🚫 Error from Moderation Service (HTTP {e.response.status_code})**
        ---
        {user_message}
        
        {error_details}
        """
        logging.error(f"HTTP Status Error: {e.response.status_code} - Response: {e.response.text}")
    
    except httpx.RequestError as e:
        # Catches network errors (e.g., server is down, DNS issues)
        summary_output = f"""
        **🔌 Connection Error**
        ---
        Could not connect to the API server at `{API_BASE_URL}`.
        
        Please ensure the backend server is running and the URL is configured correctly in your `.env` file.
        """
        full_response_output = {"error": "Connection Error", "url": API_BASE_URL, "details": str(e)}
        logging.error(f"Request Error: Could not connect to {API_BASE_URL}. Details: {e}")

    except Exception as e:
        # A catch-all for any other unexpected errors in this Gradio script
        summary_output = f"""
        **💥 An Unexpected Application Error Occurred**
        ---
        An error happened within the Gradio application itself. 
        Please check the application logs for more details.
        
        **Error Type:** `{type(e).__name__}`
        """
        full_response_output = {"error": "Gradio App Internal Error", "type": type(e).__name__, "details": str(e)}
        logging.error(f"Unexpected Error in Gradio App: {e}", exc_info=True)
        
    return summary_output, full_response_output

# --- Gradio Interface ---
with gr.Blocks(theme=gr.themes.Soft(), css="footer {display: none !important}") as demo:
    gr.Markdown(
        """
        # 🤖 Multimodal Content Moderation Demo
        This demo uses a custom API server to perform advanced content moderation.
        You can provide any combination of text, image, video, and audio. The system will analyze all inputs together.
        """
    )

    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### 1. Provide Your Content")
            text_input = gr.Textbox(label="Text Input", lines=4, placeholder="Enter any text here...")
            image_input = gr.Image(label="Image Input", type="filepath")
            video_input = gr.Video(label="Video Input")
            audio_input = gr.Audio(label="Voice/Audio Input", type="filepath")
            
            language_input = gr.Dropdown(
                label="Audio Language (if providing audio)",
                choices=list(SORTED_LANGUAGES.keys()),
                value="English",
                interactive=True
            )

            submit_button = gr.Button("Moderate Content", variant="primary")
        
        with gr.Column(scale=2):
            gr.Markdown("### 2. See the Results")
            result_output = gr.Markdown(label="Moderation Summary")
            full_response_output = gr.JSON(label="Full API Response")

    submit_button.click(
        fn=moderate_content,
        inputs=[text_input, image_input, video_input, audio_input, language_input],
        outputs=[result_output, full_response_output]
    )
    
    gr.Examples(
        examples=[
            ["This is a test of the system with safe text.", None, None, None, "English"],
            ["I am going to kill the process on my computer.", None, None, None, "English"],
        ],
        inputs=[text_input, image_input, video_input, audio_input, language_input],
        outputs=[result_output, full_response_output],
        fn=moderate_content
    )

if __name__ == "__main__":
    logging.info(f"Connecting to API server at: {API_BASE_URL}")
    if API_BASE_URL == "http://127.0.0.1:8000":
        logging.warning("API_BASE_URL is set to the default local address. Make sure this is correct or set it in your .env file.")
    demo.launch(server_name="0.0.0.0", server_port=7860)