auth_interface.py

"""
Authentication Web Interface for OpenManus
Mobile number + password based authentication forms
"""

import asyncio
import sqlite3
from typing import Optional, Tuple

import gradio as gr

from app.auth import UserSignupRequest, UserLoginRequest
from app.auth_service import AuthService
from app.logger import logger


class AuthInterface:
    """Authentication interface with Gradio"""

    def __init__(self, db_path: str = "openmanus.db"):
        self.db_path = db_path
        self.auth_service = None
        self.current_session = None
        self.init_database()

    def init_database(self):
        """Initialize database with schema"""
        try:
            conn = sqlite3.connect(self.db_path)

            # Create users table with mobile auth
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS users (
                    id TEXT PRIMARY KEY,
                    mobile_number TEXT UNIQUE NOT NULL,
                    full_name TEXT NOT NULL,
                    password_hash TEXT NOT NULL,
                    avatar_url TEXT,
                    preferences TEXT,
                    is_active BOOLEAN DEFAULT TRUE,
                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
                )
            """
            )

            # Create sessions table
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS sessions (
                    id TEXT PRIMARY KEY,
                    user_id TEXT NOT NULL,
                    title TEXT,
                    metadata TEXT,
                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                    expires_at DATETIME,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                )
            """
            )

            conn.commit()
            conn.close()
            logger.info("Database initialized successfully")

        except Exception as e:
            logger.error(f"Database initialization error: {str(e)}")

    def get_db_connection(self):
        """Get database connection"""
        return sqlite3.connect(self.db_path)

    async def handle_signup(
        self, full_name: str, mobile_number: str, password: str, confirm_password: str
    ) -> Tuple[str, bool, dict]:
        """Handle user signup"""
        try:
            # Validate input
            if not all([full_name, mobile_number, password, confirm_password]):
                return "All fields are required", False, gr.update(visible=True)

            # Create signup request
            signup_data = UserSignupRequest(
                full_name=full_name,
                mobile_number=mobile_number,
                password=password,
                confirm_password=confirm_password,
            )

            # Process signup
            db_conn = self.get_db_connection()
            auth_service = AuthService(db_conn)

            result = await auth_service.register_user(signup_data)
            db_conn.close()

            if result.success:
                self.current_session = {
                    "session_id": result.session_id,
                    "user_id": result.user_id,
                    "full_name": result.full_name,
                }
                return (
                    f"Welcome {result.full_name}! Account created successfully.",
                    True,
                    gr.update(visible=False),
                )
            else:
                return result.message, False, gr.update(visible=True)

        except ValueError as e:
            return str(e), False, gr.update(visible=True)
        except Exception as e:
            logger.error(f"Signup error: {str(e)}")
            return "An error occurred during signup", False, gr.update(visible=True)

    async def handle_login(
        self, mobile_number: str, password: str
    ) -> Tuple[str, bool, dict]:
        """Handle user login"""
        try:
            # Validate input
            if not all([mobile_number, password]):
                return (
                    "Mobile number and password are required",
                    False,
                    gr.update(visible=True),
                )

            # Create login request
            login_data = UserLoginRequest(
                mobile_number=mobile_number, password=password
            )

            # Process login
            db_conn = self.get_db_connection()
            auth_service = AuthService(db_conn)

            result = await auth_service.login_user(login_data)
            db_conn.close()

            if result.success:
                self.current_session = {
                    "session_id": result.session_id,
                    "user_id": result.user_id,
                    "full_name": result.full_name,
                }
                return (
                    f"Welcome back, {result.full_name}!",
                    True,
                    gr.update(visible=False),
                )
            else:
                return result.message, False, gr.update(visible=True)

        except ValueError as e:
            return str(e), False, gr.update(visible=True)
        except Exception as e:
            logger.error(f"Login error: {str(e)}")
            return "An error occurred during login", False, gr.update(visible=True)

    def handle_logout(self) -> Tuple[str, bool, dict]:
        """Handle user logout"""
        if self.current_session:
            # In a real app, you'd delete the session from database
            self.current_session = None

        return "Logged out successfully", False, gr.update(visible=True)

    def create_interface(self) -> gr.Interface:
        """Create the authentication interface"""

        with gr.Blocks(
            title="OpenManus Authentication", theme=gr.themes.Soft()
        ) as auth_interface:
            gr.Markdown(
                """
            # 🔐 OpenManus Authentication
            ### Secure Mobile Number + Password Login System
            """
            )

            # Session status
            session_status = gr.Textbox(
                value="Not logged in", label="Status", interactive=False
            )

            # Auth forms container
            with gr.Column(visible=True) as auth_forms:

                with gr.Tabs():

                    # Login Tab
                    with gr.TabItem("🔑 Login"):
                        gr.Markdown("### Login with your mobile number and password")

                        login_mobile = gr.Textbox(
                            label="📱 Mobile Number",
                            placeholder="Enter your mobile number (e.g., +1234567890)",
                            lines=1,
                        )

                        login_password = gr.Textbox(
                            label="🔒 Password",
                            type="password",
                            placeholder="Enter your password",
                            lines=1,
                        )

                        login_btn = gr.Button("🔑 Login", variant="primary", size="lg")
                        login_result = gr.Textbox(label="Result", interactive=False)

                    # Signup Tab
                    with gr.TabItem("📝 Sign Up"):
                        gr.Markdown("### Create your new account")

                        signup_fullname = gr.Textbox(
                            label="👤 Full Name",
                            placeholder="Enter your full name",
                            lines=1,
                        )

                        signup_mobile = gr.Textbox(
                            label="📱 Mobile Number",
                            placeholder="Enter your mobile number (e.g., +1234567890)",
                            lines=1,
                        )

                        signup_password = gr.Textbox(
                            label="🔒 Password",
                            type="password",
                            placeholder="Create a strong password (min 8 chars, include uppercase, lowercase, digit)",
                            lines=1,
                        )

                        signup_confirm_password = gr.Textbox(
                            label="🔒 Confirm Password",
                            type="password",
                            placeholder="Confirm your password",
                            lines=1,
                        )

                        signup_btn = gr.Button(
                            "📝 Create Account", variant="primary", size="lg"
                        )
                        signup_result = gr.Textbox(label="Result", interactive=False)

            # Logged in section
            with gr.Column(visible=False) as logged_in_section:
                gr.Markdown("### ✅ You are logged in!")

                user_info = gr.Markdown("Welcome!")

                logout_btn = gr.Button("🚪 Logout", variant="secondary")
                logout_result = gr.Textbox(label="Result", interactive=False)

            # Password requirements info
            with gr.Accordion("📋 Password Requirements", open=False):
                gr.Markdown(
                    """
                **Password must contain:**
                - At least 8 characters
                - At least 1 uppercase letter (A-Z)
                - At least 1 lowercase letter (a-z)
                - At least 1 digit (0-9)
                - Maximum 128 characters

                **Mobile Number Format:**
                - 10-15 digits
                - Can include country code
                - Examples: +1234567890, 1234567890, +91987654321
                """
                )

            # Event handlers
            def sync_signup(*args):
                """Synchronous wrapper for signup"""
                return asyncio.run(self.handle_signup(*args))

            def sync_login(*args):
                """Synchronous wrapper for login"""
                return asyncio.run(self.handle_login(*args))

            def update_ui_after_auth(result_text, success, auth_forms_update):
                """Update UI after authentication"""
                if success:
                    return (
                        result_text,  # session_status
                        auth_forms_update,  # auth_forms visibility
                        gr.update(visible=True),  # logged_in_section visibility
                        f"### 👋 {self.current_session['full_name'] if self.current_session else 'User'}",  # user_info
                    )
                else:
                    return (
                        "Not logged in",  # session_status
                        auth_forms_update,  # auth_forms visibility
                        gr.update(visible=False),  # logged_in_section visibility
                        "Welcome!",  # user_info
                    )

            def update_ui_after_logout(result_text, success, auth_forms_update):
                """Update UI after logout"""
                return (
                    "Not logged in",  # session_status
                    auth_forms_update,  # auth_forms visibility
                    gr.update(visible=False),  # logged_in_section visibility
                    "Welcome!",  # user_info
                )

            # Login button click
            login_btn.click(
                fn=sync_login,
                inputs=[login_mobile, login_password],
                outputs=[login_result, gr.State(), gr.State()],
            ).then(
                fn=update_ui_after_auth,
                inputs=[login_result, gr.State(), gr.State()],
                outputs=[session_status, auth_forms, logged_in_section, user_info],
            )

            # Signup button click
            signup_btn.click(
                fn=sync_signup,
                inputs=[
                    signup_fullname,
                    signup_mobile,
                    signup_password,
                    signup_confirm_password,
                ],
                outputs=[signup_result, gr.State(), gr.State()],
            ).then(
                fn=update_ui_after_auth,
                inputs=[signup_result, gr.State(), gr.State()],
                outputs=[session_status, auth_forms, logged_in_section, user_info],
            )

            # Logout button click
            logout_btn.click(
                fn=self.handle_logout, outputs=[logout_result, gr.State(), gr.State()]
            ).then(
                fn=update_ui_after_logout,
                inputs=[logout_result, gr.State(), gr.State()],
                outputs=[session_status, auth_forms, logged_in_section, user_info],
            )

        return auth_interface


# Standalone authentication app
def create_auth_app(db_path: str = "openmanus.db") -> gr.Interface:
    """Create standalone authentication app"""
    auth_interface = AuthInterface(db_path)
    return auth_interface.create_interface()


if __name__ == "__main__":
    # Run standalone auth interface for testing
    auth_app = create_auth_app()
    auth_app.launch(server_name="0.0.0.0", server_port=7860, share=False, debug=True)