MicrositePilot / app.py
devesh1011
code changes
4b0553d
import gradio as gr
from workflow import (
MicroSiteGenerator,
) # Make sure workflow.py is in the same directory or accessible
from agno.workflow import RunEvent
import os
from dotenv import load_dotenv
import traceback # Import traceback for detailed error logging
# Load environment variables from .env file, if present
load_dotenv()
# Instantiate the workflow
# This will also initialize the agents defined in the workflow.
microsite_workflow: MicroSiteGenerator = None
try:
microsite_workflow = MicroSiteGenerator()
except Exception as e:
print(f"Error initializing MicroSiteGenerator: {e}")
traceback.print_exc()
# Gradio UI will show a message if microsite_workflow is None
def generate_microsite_app(audio_file_obj, audio_format_str, use_cache_bool):
"""
Gradio function to process audio and generate a microsite using the local workflow.
audio_file_obj: Output from gr.Audio (type="filepath").
audio_format_str: The user-selected audio format.
use_cache_bool: Boolean indicating whether to use the transcription cache.
"""
if microsite_workflow is None:
return (
"Critical Error: Workflow failed to initialize. Check console logs.",
"App is not functional. Please ensure all configurations (like API keys) are correctly set.",
)
if audio_file_obj is None:
return "Status: Idle", "Please upload an audio file to begin. 🎀"
audio_source_path = audio_file_obj # This is already the path string
processing_log_entries = ["πŸš€ Starting microsite generation..."]
final_result_markdown = "⏳ Processing... please wait."
try:
# The run method is a generator. We iterate to get the final result.
for response in microsite_workflow.run(
audio_source=audio_source_path,
audio_format=audio_format_str.lower(), # Ensure format is lowercase
):
processing_log_entries.append(f"πŸ”„ Workflow event: {response.event.value}")
if response.event == RunEvent.workflow_completed:
content = response.content # This is site_details from the workflow
if isinstance(content, dict): # Expected site_details dictionary
if content.get("success"):
site_url = content.get("site", {}).get("url")
site_name = content.get("site", {}).get("name", "N/A")
admin_url = content.get("site", {}).get("admin_url", "#")
if site_url:
final_result_markdown = (
f"πŸŽ‰ **Microsite '{site_name}' Deployed!** πŸŽ‰\n\n"
f"πŸ”— **Access it here:** [{site_url}]({site_url})\n\n"
f"<details><summary>ℹ️ Deployment Details (Admin Link)</summary>"
f"<p>Admin URL: <a href='{admin_url}' target='_blank' rel='noopener noreferrer'>{admin_url}</a></p>"
f"</details>"
)
processing_log_entries.append("βœ… Deployment successful.")
else:
final_result_markdown = "⚠️ Microsite generated, but deployment URL was not found in the response."
processing_log_entries.append("❌ Deployment URL missing.")
else: # Deployment failed or error reported by deploy_html_file_with_digest
error_msg = content.get("message", "Unknown deployment error.")
final_result_markdown = (
f"❌ **Microsite Generation Failed:** {error_msg}"
)
processing_log_entries.append(
f"❌ Deployment/Generation failed: {error_msg}"
)
elif isinstance(content, str) and "Site was not generated" in content: # Workflow specific error string
final_result_markdown = f"❌ **Site Not Generated:** {content}. This often indicates a transcription failure or issue with the audio."
processing_log_entries.append(
f"❌ Site generation aborted: {content}"
)
else:
final_result_markdown = f"πŸ€” Workflow completed with unexpected content: {str(content)[:200]}..." # Truncate long content
processing_log_entries.append(
f"⚠️ Unexpected content: {str(content)[:200]}..."
)
break # Critical final event processed
else:
# Handle other potential intermediate events if the workflow is updated to yield more
# For example, if the workflow yields progress updates:
if hasattr(response.content, 'get') and response.content.get('progress_message'):
processing_log_entries.append(f"⏳ {response.content['progress_message']}")
elif isinstance(response.content, str):
processing_log_entries.append(f"ℹ️ {response.content[:100]}...")
except Exception as e:
tb_str = traceback.format_exc()
processing_log_entries.append(
f"πŸ’₯ Critical error during workflow execution: {str(e)}"
)
print(f"Error in generate_microsite_app: {e}\n{tb_str}") # Log to console for debugging
final_result_markdown = (
f"πŸ’₯ **An Unexpected Error Occurred!** πŸ’₯\n\n"
f"Details: `{str(e)}`\n\n"
f"Please check the console logs for more information or try again. "
f"If the problem persists, ensure all configurations and API keys are correctly set."
)
return "\n".join(processing_log_entries), final_result_markdown
# Define common audio formats
COMMON_AUDIO_FORMATS = [
"wav", "mp3", "m4a", "flac", "ogg", "aac", "opus", "amr", "webm",
]
# Prepare environment variable warning message for the UI
missing_env_vars_messages = []
if not os.getenv("NETLIFY_PERSONAL_ACCESS_TOKEN"):
missing_env_vars_messages.append("`NETLIFY_PERSONAL_ACCESS_TOKEN` (for deploying the site to Netlify)")
if not os.getenv("GOOGLE_API_KEY"): # Common for Gemini models
missing_env_vars_messages.append("`GOOGLE_API_KEY` (for AI models like Google Gemini)")
# Add other critical env vars checks here if your workflow agents need them
env_warning_html = ""
if missing_env_vars_messages:
vars_list_html = "".join([f"<li>{var}</li>" for var in missing_env_vars_messages])
env_warning_html = (
f"<div style='background-color: #332200; color: #FFDDAA; border: 1px solid #553300; padding: 15px; margin-bottom:20px; border-radius: 5px;'>"
f"<strong>⚠️ Heads up!</strong> The application might be missing the following environment variable(s):"
f"<ul style='margin-top: 10px; margin-bottom: 0; padding-left: 20px;'>{vars_list_html}</ul>"
f"This could affect its functionality. Please ensure they are set in your environment or `.env` file."
f"</div>"
)
# Workflow description for the UI
workflow_desc_html = ""
if microsite_workflow and hasattr(microsite_workflow, "description"):
escaped_description = microsite_workflow.description.replace("\n", "<br>")
workflow_desc_html = f"""
<details style="margin-top:15px; margin-bottom:15px; padding:10px; background-color:#2a2a2a; border-radius:5px; border: 1px solid #444;">
<summary style="font-weight:bold; cursor:pointer;">πŸ“– Click to see Workflow Details</summary>
<p style="margin-top:10px;"><em>{escaped_description}</em></p>
</details>
"""
elif microsite_workflow is None:
workflow_desc_html = "<p style='color:red; font-weight:bold;'>WORKFLOW INITIALIZATION FAILED. Please check console logs for errors. API keys might be missing or other configurations might be incorrect.</p>"
else:
workflow_desc_html = "<p style='color:orange;'>Workflow description not available.</p>"
app_title = "MicrositePilot πŸŽ™οΈβž‘οΈπŸŒ"
app_intro_markdown = f"""
{env_warning_html}
Welcome to **MicrositePilot**! Upload a product demo call recording (audio file).
The AI will transcribe it, extract key information, and generate a personalized recap microsite, automatically deployed to Netlify.
{workflow_desc_html}
"""
custom_css = """
body { font-family: 'Inter', sans-serif; }
.gradio-container { max-width: 900px !important; margin: auto !important; }
footer { display: none !important; } /* Hide default Gradio footer */
h1 { text-align: center; }
.gr-button { box-shadow: 0 1px 3px 0 rgba(0,0,0,.1), 0 1px 2px 0 rgba(0,0,0,.06); }
"""
with gr.Blocks(theme="dark", css=custom_css) as demo:
gr.Markdown(f"<h1>{app_title}</h1>")
gr.HTML(app_intro_markdown)
with gr.Row(equal_height=False):
with gr.Column(scale=1, min_width=300):
gr.Markdown("### πŸ“€ Step 1: Upload Audio")
audio_input = gr.Audio(
type="filepath",
label="Product Demo Audio File (WAV, MP3, M4A, etc.)",
)
gr.Markdown("### βš™οΈ Step 2: Configure Options")
audio_format_input = gr.Dropdown(
choices=COMMON_AUDIO_FORMATS,
label="Original Audio Format (Crucial)",
value="mp3", # Default common format
)
cache_checkbox = gr.Checkbox(
label="Use Transcription Cache ⚑ (Speeds up re-runs)",
value=True, # Default to using cache
)
gr.Markdown("### ✨ Step 3: Generate!")
submit_button = gr.Button(
"Generate Microsite", variant="primary", elem_id="submit_button_custom"
)
with gr.Column(scale=2, min_width=400):
gr.Markdown("### πŸ“Š Results")
log_output = gr.Textbox(
label="βš™οΈ Processing Log",
lines=12,
interactive=False,
placeholder="Workflow updates and logs will appear here...",
)
microsite_link_output = gr.Markdown(
label="πŸ”— Microsite Output",
value="Your deployed microsite link and details will appear here once generated.",
)
example_audio_file = "Listen to an A.I. sales rep cold call (and close) a prospect. #ai #sales.mp3"
if os.path.exists(example_audio_file):
gr.Examples(
examples=[[example_audio_file, "mp3", True]],
inputs=[audio_input, audio_format_input, cache_checkbox],
outputs=[log_output, microsite_link_output],
fn=generate_microsite_app,
cache_examples=False,
label="πŸ“‹ Example (click to run)",
)
else:
gr.Markdown(
"<p style='text-align:center; font-style:italic; color:grey;'>Note: Example audio file 'Listen to an A.I. sales rep cold call (and close) a prospect. #ai #sales.mp3' not found. Examples disabled.</p>"
)
submit_button.click(
fn=generate_microsite_app,
inputs=[audio_input, audio_format_input, cache_checkbox],
outputs=[log_output, microsite_link_output],
api_name="generate_microsite",
)
if __name__ == "__main__":
if microsite_workflow is None:
print("CRITICAL: MicroSiteGenerator workflow failed to initialize. The Gradio app might not function correctly.")
print("Please check for errors above, ensure API keys (e.g., GOOGLE_API_KEY, NETLIFY_PERSONAL_ACCESS_TOKEN) are set in your .env file or environment, and all dependencies are installed.")
else:
print("MicroSiteGenerator workflow initialized successfully.")
if not os.getenv("NETLIFY_PERSONAL_ACCESS_TOKEN"):
print("CONSOLE REMINDER: NETLIFY_PERSONAL_ACCESS_TOKEN is not set. Deployment to Netlify will fail.")
if not os.getenv("GOOGLE_API_KEY"):
print("CONSOLE REMINDER: GOOGLE_API_KEY is not set. AI agent calls may fail.")
print("Gradio app starting...")
demo.launch()