import os import assemblyai as aai import gradio as gr import requests import time from supabase import create_client, Client # -------------------------------------------------- # Load secrets from environment # -------------------------------------------------- ASSEMBLYAI_API_KEY = os.environ.get("ASSEMBLYAI_API_KEY") SUPABASE_URL = os.environ.get("SUPABASE_URL") SUPABASE_KEY = os.environ.get("SUPABASE_KEY") WEBHOOK_URL = os.environ.get( "N8N_WEBHOOK_URL", "http://13.48.132.18:5678/webhook-test/report-writing" ) if not ASSEMBLYAI_API_KEY: raise RuntimeError("ASSEMBLYAI_API_KEY not set") if not SUPABASE_URL or not SUPABASE_KEY: raise RuntimeError("SUPABASE_URL or SUPABASE_KEY not set") # -------------------------------------------------- # Initialize clients # -------------------------------------------------- aai.settings.api_key = ASSEMBLYAI_API_KEY supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY) # -------------------------------------------------- # In-memory session # -------------------------------------------------- INACTIVITY_TIMEOUT = 30 * 60 # 30 minutes in seconds session = {"user": None, "user_id": None, "last_activity": None} # -------------------------------------------------- # Check for existing Supabase session on startup # -------------------------------------------------- def restore_session(): try: user = supabase.auth.get_user() if user and user.user: session["user"] = user.user.email session["user_id"] = user.user.id return True except Exception as e: print(f"⚠️ Session restore error: {e}") return False # Restore session on app startup restore_session() # -------------------------------------------------- # Inactivity timeout check # -------------------------------------------------- def check_inactivity(): """Check if user has been inactive for 30 mins, auto-logout if true""" if session["user_id"] is None: return False if session["last_activity"] is None: return False elapsed = time.time() - session["last_activity"] if elapsed > INACTIVITY_TIMEOUT: # Auto logout try: supabase.auth.sign_out() except: pass session["user"] = None session["user_id"] = None session["last_activity"] = None return True # Timed out return False def update_activity(): """Update last activity timestamp""" session["last_activity"] = time.time() # -------------------------------------------------- # Auth functions # -------------------------------------------------- def signup(email, password): try: res = supabase.auth.sign_up({"email": email, "password": password}) if res.user: session["user"] = res.user.email session["user_id"] = res.user.id session["last_activity"] = time.time() return ( "✅ Signup successful.", gr.update(visible=False), # hide auth gr.update(visible=True) # show transcription ) return "❌ Signup failed.", gr.update(), gr.update() except Exception as e: return f"❌ Signup error: {str(e)}", gr.update(), gr.update() def login(email, password): try: res = supabase.auth.sign_in_with_password({"email": email, "password": password}) if res.user: session["user"] = res.user.email session["user_id"] = res.user.id session["last_activity"] = time.time() return ( "✅ Login successful.", gr.update(visible=False), # hide auth gr.update(visible=True) # show transcription ) return "❌ Invalid credentials.", gr.update(), gr.update() except Exception as e: return f"❌ Login error: {str(e)}", gr.update(), gr.update() def logout(): try: supabase.auth.sign_out() session["user"] = None session["user_id"] = None return ( "✅ Logged out successfully.", gr.update(visible=True), # show auth gr.update(visible=False) # hide transcription ) except Exception as e: return f"❌ Logout error: {str(e)}", gr.update(), gr.update() # -------------------------------------------------- # Transcription logic # -------------------------------------------------- def transcribe(audio_path: str): transcriber = aai.Transcriber() transcript = transcriber.transcribe(audio_path) if transcript.status == aai.TranscriptStatus.error: raise RuntimeError(transcript.error) return transcript.text, transcript.audio_duration def transcribe_audio(audio_file): # Check for inactivity timeout if check_inactivity(): return "❌ Session expired due to inactivity. Please log in again." if session["user_id"] is None: return "❌ Please log in." if audio_file is None: return "⚠️ Please upload an audio file." try: # Update activity timestamp update_activity() # Transcribe audio text, duration_seconds = transcribe(audio_file) minutes_used = max(1, int(duration_seconds // 60)) # Save transcription supabase.table("transcriptions").insert({ "user_id": session["user_id"], "user_email": session["user"], "audio_duration": minutes_used, "transcript": text }).execute() # Update users_usage existing = supabase.table("users_usage").select("minutes_used").eq("user_id", session["user_id"]).execute() if existing.data: supabase.table("users_usage").update({ "minutes_used": existing.data[0]["minutes_used"] + minutes_used }).eq("user_id", session["user_id"]).execute() else: supabase.table("users_usage").insert({ "user_id": session["user_id"], "minutes_used": minutes_used }).execute() # Send to N8N webhook using GET try: payload = { "transcriptions": text, "user_email": session["user"] } resp = requests.get(WEBHOOK_URL, params=payload) if not resp.ok: print(f"⚠️ Webhook failed: {resp.status_code} {resp.text}") except Exception as e: print(f"⚠️ Webhook error: {e}") return text except Exception as e: return f"❌ Error: {str(e)}" # -------------------------------------------------- # Gradio UI # -------------------------------------------------- with gr.Blocks() as app: # 🔐 AUTH SECTION with gr.Column(visible=not restore_session()) as auth_section: gr.Markdown("## 🔐 Login or Signup") email = gr.Textbox(label="Email") password = gr.Textbox(label="Password", type="password") auth_status = gr.Textbox(label="Status") with gr.Row(): login_btn = gr.Button("Login") signup_btn = gr.Button("Signup") # 🎙️ TRANSCRIPTION SECTION (hidden initially) with gr.Column(visible=restore_session()) as transcription_section: gr.Markdown("## 🎙️ AI Transcription") gr.Markdown("Upload audio → get transcript") with gr.Row(): logout_btn = gr.Button("🚪 Logout", scale=1) audio_input = gr.Audio(type="filepath", label="Upload Audio") output_text = gr.Textbox(label="Transcript", lines=10) transcribe_btn = gr.Button("Transcribe") transcribe_btn.click(transcribe_audio, audio_input, output_text) # 📋 EMBEDDED N8N FORM gr.Markdown("## 📋 Site Report Form") gr.HTML( '' ) logout_status = gr.Textbox(label="Status", visible=False) # -------------------------------------------------- # Button bindings login_btn.click( login, [email, password], [auth_status, auth_section, transcription_section] ) signup_btn.click( signup, [email, password], [auth_status, auth_section, transcription_section] ) logout_btn.click( logout, None, [logout_status, transcription_section, auth_section] ) # -------------------------------------------------- # Launch # -------------------------------------------------- app.launch( server_name="0.0.0.0", server_port=7860, show_error=True )