""" ScrAIbe Web GUI (Gradio) - Async Mode ------------------------------------- Runs the Web GUI that: - Accepts audio uploads - Enqueues transcription jobs asynchronously via Celery - Backend worker: - Transcribes (with diarization) - Optionally summarizes - Emails the user: - Immediately: confirmation + queue position - On success: transcript + JSON (+ summary if requested) - On error: error details This is the default entrypoint when running in Docker. """ import os import logging import shutil from datetime import datetime import gradio as gr from .misc import setup_logging logger = logging.getLogger("scraibe.webui") def load_config(): """ Load configuration from misc/config.yaml if present. Primary runtime configuration is via environment variables. """ config_path = os.getenv("SCRAIBE_CONFIG", "/app/src/misc/config.yaml") config = {} if os.path.exists(config_path): try: import yaml with open(config_path, "r", encoding="utf-8") as f: config = yaml.safe_load(f) or {} except Exception as e: logger.warning("Failed to load config from %s: %s", config_path, e) return config def load_html_template(path: str, **kwargs) -> str: """ Load an HTML template and fill placeholders. """ if not os.path.exists(path): return "" with open(path, "r", encoding="utf-8") as f: template = f.read() try: return template.format(**kwargs) except KeyError: return template def create_app(): """ Create and launch the Gradio Web GUI (async mode). """ # Logging log_level = os.getenv("LOG_LEVEL", "INFO") setup_logging(level=log_level) # Load config (branding, layout, etc.) config = load_config() layout_cfg = config.get("layout", {}) launch_cfg = config.get("launch", {}) logger.info("Starting ScrAIbe Web GUI (async mode).") # Ensure upload directory exists upload_dir = os.getenv("SCRAIBE_UPLOAD_DIR", "/tmp/scraibe_uploads") os.makedirs(upload_dir, exist_ok=True) # Paths for assets header_path = layout_cfg.get("header", "/app/src/misc/header.html") footer_path = layout_cfg.get("footer", "/app/src/misc/footer.html") # Configurable title, logo URL, and accent color via environment webui_title = os.getenv("WEBUI_TITLE", "A.P.Strom Transcription") logo_url = os.getenv("WEBUI_LOGO_URL", "https://apstrom.ca") accent_color = os.getenv("EMAIL_ACCENT_COLOR", "#7C6DA0") # Prepare header HTML with logo URL and accent color header_html = "" if os.path.exists(header_path): header_html = load_html_template( header_path, webui_title=webui_title, header_logo_url=logo_url, header_logo_src=logo_url, accent_color=accent_color, ) # Prepare footer HTML with accent color footer_html = "" if os.path.exists(footer_path): version = os.getenv("SCRABIE_VERSION", "0.1.1.dev") footer_html = load_html_template( footer_path, footer_scraibe_webui_version=version, accent_color=accent_color, ) # Build Gradio interface with gr.Blocks( title=webui_title, css=""" /* Responsive layout: stack columns on smaller screens */ @media (max-width: 850px) { .gradio-container { max-width: 100% !important; } #main-row .gr-row { flex-direction: column !important; } #main-row .gr-col { width: 100% !important; max-width: 100% !important; flex: none !important; } } """, ) as app: # Header if header_html: gr.HTML(header_html) with gr.Row(elem_id="main-row"): with gr.Column(): audio_input = gr.Audio( label="Upload or record audio", type="filepath", ) task_choice = gr.Radio( choices=[ ("Transcribe", "transcribe"), ("Transcribe & summarize", "transcript_and_summarize"), ], value="transcribe", label="Task", container=True, ) identify_speakers = gr.Checkbox( label="Identify speakers (best effort using AI)", value=True, info="If enabled, AI will attempt to infer real names for speakers and replace Speaker 1/2/etc. in the transcript.", ) with gr.Row(): language_input = gr.Textbox( label="Language (optional)", placeholder="e.g., english, german", ) num_speakers_input = gr.Number( label="Number of speakers (optional)", precision=0, ) email_to = gr.Textbox( label="Your email address (required)", placeholder="e.g. your.name@example.com", ) email_cc = gr.Textbox( label="CC (optional, comma-separated)", placeholder="e.g. manager@example.com", ) submit_btn = gr.Button("Submit for transcription", variant="primary") with gr.Column(): status_text = gr.Textbox( label="Status", lines=6, interactive=False, ) # Footer if footer_html: gr.HTML(footer_html) # Events def on_task_change(value): # No special UI changes needed; both modes handled in backend return task_choice.change( fn=on_task_change, inputs=[task_choice], outputs=[], ) def on_submit( audio, task, language, num_speakers, email_to_val, email_cc_val, identify_speakers_val, ): if not audio: return "Please upload or record audio." email_to_val = (email_to_val or "").strip() if not email_to_val: return "Please enter your email address." # Copy uploaded file to a stable location try: ext = os.path.splitext(audio)[1] or ".wav" ts = datetime.utcnow().strftime("%Y%m%d%H%M%S%f") new_name = f"upload_{ts}{ext}" dest_path = os.path.join(upload_dir, new_name) shutil.copy2(audio, dest_path) except Exception as e: logger.error("Error copying uploaded file: %s", e) return f"Error saving your file: {e}" # Import Celery task try: from .tasks import process_transcription_task except ImportError: return ( "Error: async processing is not available (Celery not configured)." ) # Enqueue transcription job try: task_result = process_transcription_task.delay( audio_path=dest_path, task_type=task, language=language or None, num_speakers=int(num_speakers) if num_speakers else None, email_to=email_to_val, email_cc=email_cc_val or None, include_summary=(task == "transcript_and_summarize"), identify_speakers=bool(identify_speakers_val), ) except Exception as e: logger.error("Error enqueuing job: %s", e) return f"Error submitting your file: {e}" return ( "Your audio file has been received and added to the queue.\n" "We have sent a confirmation email to you.\n" "You will receive another email with your transcript (and summary, if requested) " "once processing is complete.\n" f"Job ID: {task_result.id}" ) submit_btn.click( fn=on_submit, inputs=[ audio_input, task_choice, language_input, num_speakers_input, email_to, email_cc, identify_speakers, ], outputs=[status_text], ) # Launch options with accent color applied via CSS server_name = launch_cfg.get("server_name", os.getenv("GRADIO_SERVER_NAME", "0.0.0.0")) server_port = launch_cfg.get("server_port", 7860) accent_css = f""" :root {{ --primary-accent: {accent_color}; }} button.primary, .primary, .gradio-button-primary, .gradio-container button.primary {{ background-color: var(--primary-accent) !important; border-color: var(--primary-accent) !important; }} button.primary:hover, .primary:hover, .gradio-button-primary:hover {{ background-color: var(--primary-accent) !important; opacity: 0.95; }} .radio-item.selected, .radio-item.selected label {{ color: var(--primary-accent) !important; }} a, .gradio-container a {{ color: var(--primary-accent) !important; }} body {{ font-family: Arial, sans-serif; }} /* Increase main title font size */ h1, .webui-title, .header-title {{ font-size: 60px !important; }} /* Hide Gradio's "Use via API" link/button */ #share-btn, a[href*="/api"], a[href*="#/api"], a[href*="#api"], .gradio-container a[href*="api"] {{ display: none !important; }} /* Mobile-friendly adjustments */ @media (max-width: 700px) {{ .gradio-container {{ padding: 0 4px !important; }} .gradio-container .gr-row {{ flex-direction: column !important; gap: 8px !important; }} .gradio-container .gr-col {{ width: 100% !important; max-width: 100% !important; flex: none !important; }} .gradio-container button.primary {{ width: 100% !important; box-sizing: border-box; }} }} """ app.launch( server_name=str(server_name), server_port=int(server_port), css=accent_css, ) if __name__ == "__main__": create_app()