app.py

import gradio as gr
import requests
import base64
from PIL import Image
import io
import logging
from typing import List, Dict, Optional, Tuple

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

MAX_PAIRS = 20
SERVER_URL = "https://c73b-163-1-222-182.ngrok-free.app"

def create_qa_boxes(num_pairs: int) -> Tuple[List[gr.Textbox], List[gr.Textbox]]:
    """Dynamically create question and answer textboxes"""
    questions = []
    answers = []
    for i in range(num_pairs):
        questions.append(gr.Textbox(
            lines=2,
            label=f"Question {i+1}",
            placeholder="Enter question here...",
            visible=True
        ))
        answers.append(gr.Textbox(
            lines=2,
            label=f"Answer {i+1}",
            placeholder="Enter answer here...",
            visible=True
        ))
    return questions, answers

def validate_password(password: str) -> Tuple[bool, str, Optional[str]]:
    """Validate the password with the server and return the API key if correct"""
    try:
        response = requests.post(f"{SERVER_URL}/validate", json={"password": password})
        if response.status_code == 200:
            data = response.json()
            if data.get("valid"):
                return True, "Password correct. Loading data...", data.get("api_key")
        return False, "Incorrect password", None
    except Exception as e:
        print(f"Error validating password: {e}")
        return False, "Server error during validation", None

def base64_to_image(base64_str: str) -> Image.Image:
    """Convert base64 string to PIL Image"""
    image_bytes = base64.b64decode(base64_str)
    return Image.open(io.BytesIO(image_bytes))

def get_next_item(api_key: str) -> Optional[Dict]:
    """Get next item from the server"""
    try:
        headers = {"X-API-Key": api_key} if api_key else {}
        response = requests.get(f"{SERVER_URL}/get_data", headers=headers)
        if response.status_code == 200:
            return response.json()
        return None
    except Exception as e:
        print(f"Error getting next item: {e}")
        return None

def save_item(data: Dict, api_key: str) -> bool:
    """Save item to the server"""
    try:
        # Remove image_base64 if present
        data_to_save = {k: v for k, v in data.items() if k != "image_base64"}
        headers = {"X-API-Key": api_key} if api_key else {}
        response = requests.post(f"{SERVER_URL}/save_data", json=data_to_save, headers=headers)
        return response.status_code == 200
    except Exception as e:
        print(f"Error saving item: {e}")
        return False

def update_interface(data: Optional[Dict]) -> Tuple[Optional[Image.Image], str, List[str], List[str], List[bool], List[bool]]:
    """Update interface with new data"""
    if data is None:
        logging.warning("update_interface: No data provided")
        return None, "No data available", [], [], [], []
    
    logging.info(f"update_interface: Received data with {len(data['questions'])} questions")
    
    # Convert base64 to image
    image = base64_to_image(data.get("image_base64")) if data.get("image_base64") else None
    
    # Extract questions and answers
    questions = [qa["question"] for qa in data["questions"]]
    answers = [qa["answer"] for qa in data["questions"]]

    updates = []
    
    # First add all question updates
    for i in range(MAX_PAIRS):
        if i < len(questions):
            logging.info(f"Making question {i+1} visible with: {questions[i]}")
            updates.append(gr.update(value=questions[i], visible=True))
            updates.append(gr.update(value=answers[i], visible=True))
        else:
            updates.append(gr.update(value="", visible=False))
            updates.append(gr.update(value="", visible=False))

    return image, "Data loaded successfully", updates

def login(password: str, state: Dict):
    """Handle login and load first item"""
    logging.info("Attempting login")
    success, message, api_key = validate_password(password)
    if not success:
        logging.warning(f"Login failed: {message}")
        # Return empty updates for both questions and answers
        empty_updates = [gr.update(value="", visible=False)] * MAX_PAIRS * 2
        return [None, message, gr.update(visible=True), gr.update(visible=False), {"api_key": None, "current_data": None}] + empty_updates
    
    logging.info("Login successful, fetching first item")
    # Get first item
    current_data = get_next_item(api_key)
    if current_data is None:
        logging.warning("No items available after login")
        empty_updates = [gr.update(value="", visible=False)] * MAX_PAIRS * 2
        return [None, "No items available", gr.update(visible=True), gr.update(visible=False), {"api_key": api_key, "current_data": None}] + empty_updates
    
    # Update interface with new data
    image, status, question_answers = update_interface(current_data)
    
    logging.info("Returning login updates")
    return [image, status, gr.update(visible=False), gr.update(visible=True), {"api_key": api_key, "current_data": current_data}] + question_answers

def save_and_next(*args):
    """Save current item and load next one"""
    # Extract state from the last argument
    qa_inputs = args[:-1]
    state = args[-1]
    current_data = state.get("current_data")
    api_key = state.get("api_key")
    
    logging.info("save_and_next called")
    
    if current_data is not None and api_key is not None:
        # Split inputs into questions and answers
        questions = qa_inputs[::2]
        answers = qa_inputs[1::2]
        
        # Filter out hidden inputs
        valid_qa_pairs = []
        for q, a in zip(questions, answers):
            if q is not None and q.strip() and a is not None and a.strip():  # If question exists
                valid_qa_pairs.append({"question": q, "answer": a})
        current_data["questions"] = valid_qa_pairs

        if not save_item(current_data, api_key):
            logging.error("Failed to save current item")
            return [None, "Failed to save item", {"api_key": api_key, "current_data": None}] + [gr.update(value="", visible=False)] * MAX_PAIRS * 2
    
        logging.info("Successfully saved current item, fetching next")
    
    # Get next item
    next_data = get_next_item(api_key)
    if next_data is None:
        logging.warning("No more items available after save")
        empty_updates = [gr.update(value="", visible=False)] * MAX_PAIRS * 2
        return [None, "No more items available", {"api_key": api_key, "current_data": None}] + empty_updates
    
    image, status, updates = update_interface(next_data)
    return [image, status, {"api_key": api_key, "current_data": next_data}] + updates

# Create the Gradio interface
with gr.Blocks() as app:
    # Initialize session state
    state = gr.State(value={"api_key": None, "current_data": None})
    
    logging.info("Creating Gradio interface")
    # Login section
    with gr.Row(visible=True) as login_row:
        password_input = gr.Textbox(type="password", label="Password")
        login_button = gr.Button("Login")
        login_message = gr.Textbox(label="Status", interactive=False)
    
    # Main interface (hidden initially)
    with gr.Row(visible=False) as main_interface:
        # Left column - Image
        with gr.Column(scale=1):
            image_output = gr.Image(type="pil", label="Image")
        
        # Right column - Q&A pairs and controls
        with gr.Column(scale=1):
            # Create maximum number of Q&A pairs (they'll be hidden/shown as needed)
            questions_list = []
            answers_list = []
            
            # Create containers for Q&A pairs
            
            # Create interleaved Q&A pairs
            for i in range(MAX_PAIRS):
                with gr.Group():
                    # Question box
                    questions_list.append(gr.Textbox(
                        lines=2,
                        label=f"Question {i+1}",
                        placeholder="Enter question here...",
                        visible=False
                    ))
                    # Answer box immediately after its question
                    answers_list.append(gr.Textbox(
                        lines=2,
                        label=f"Answer {i+1}",
                        placeholder="Enter answer here...",
                        visible=False
                    ))
            
            save_button = gr.Button("Save and Load Next")
            status_message = gr.Textbox(label="Status", interactive=False)
    
    logging.info("Setting up event handlers")
    # Set up event handlers
    all_components = []
    for i in range(MAX_PAIRS):
        all_components.extend([questions_list[i], answers_list[i]])
        
    login_button.click(
        login,
        inputs=[password_input, state],
        outputs=[image_output, login_message, login_row, main_interface, state] + all_components
    )
    
    save_button.click(
        save_and_next,
        inputs=all_components + [state],
        outputs=[image_output, status_message, state] + all_components
    )

if __name__ == "__main__":
    logging.info("Starting Gradio app")
    app.launch()