Initial Commit

.gitignore

@@ -0,0 +1,3 @@
+__pycache__
+.venv
+.env

.python-version

@@ -0,0 +1 @@
+3.12

agents/info_extractor_agent.py

@@ -0,0 +1,32 @@
+from agno.agent import Agent
+from agno.models.google import Gemini
+from textwrap import dedent
+
+info_extractor = Agent(
+    model=Gemini(id="gemini-2.0-flash-001", response_modalities=["text"]),
+    description=dedent(
+        """\
+        Extracts key information from product demo call transcriptions.
+        Analyzes conversation context to identify product details, prospect pain points,
+        demonstrated features with timestamps, and actionable next steps, structuring
+        the output for microsite generation."""
+    ),
+    instructions=dedent(
+        """\
+        Given a timestamped product demo call transcription, extract the following information.
+        Format your response strictly as a JSON object validated by the `DemoSummary` Pydantic model.
+
+        **Extraction Rules:**
+        1. **Product Name:** Identify the primary product or solution discussed.
+        2. **Prospect Company:** Determine the name of the prospective customer's organization.
+        3. **Sales Rep:** Identify the name of the sales representative.
+        4. **Summary Points:** Provide 3-5 concise, high-level bullet points summarizing the entire demo.
+        5. **Pain Points Discussed:** List specific challenges or problems the prospect mentioned.
+        6. **Features Demonstrated:** For each feature explicitly shown or discussed in detail, provide a dictionary with 'name' (the feature name), 'timestamp_start' (start time, e.g., '00:05:10'), and 'timestamp_end' (end time, e.g., '00:08:45'). If a feature is mentioned but not demonstrated, do not include timestamps.
+        7. **Next Steps:** List any clear action items or agreed-upon follow-ups for either party.
+        8. **Unanswered Questions:** List any specific questions posed by the prospect that were not fully resolved during the call.
+        9. **Strict JSON Output:** Ensure the output is valid JSON and perfectly matches the structure defined by the `DemoSummary` model. Do not include any extra text or conversational filler outside the JSON.
+        """
+    ),
+    # response_model=DemoSummary,
+)

agents/site_builder_agent.py

@@ -0,0 +1,73 @@
+from agno.agent import Agent
+from agno.models.google import Gemini
+from textwrap import dedent
+from pydantic import BaseModel, Field
+
+
+class HtmlContent(BaseModel):
+    content: str = Field(
+        ..., description="The generated HTML content for the microsite."
+    )
+
+
+microsite_builder_agent = Agent(
+    model=Gemini(id="gemini-2.0-flash-001", response_modalities=["text"]),
+    description=dedent(
+        """\
+                Generates a personalized, interactive HTML microsite from demo call data.
+                It combines structured extracted information with raw transcription details
+                to create a visually appealing and informative recap page for prospects."""
+    ),
+    instructions=dedent(
+        f"""\
+                You are an expert web developer specializing in creating concise, engaging, and personalized microsites for product demo recaps.
+
+                **Your Task:**
+                Generate a complete, single-page HTML document for a product demo recap microsite.
+                The HTML should be fully self-contained (no external CSS files, use Tailwind CSS CDN).
+                It must be responsive, visually appealing, and **have clean, minimal formatting (avoid excessive newlines or unnecessary whitespace)**.
+
+                **Inputs:**
+                -   `extracted_info_json`: A JSON string containing structured data about the demo (product, prospect, features, pain points, next steps, etc.).
+                -   `raw_transcription`: The full, verbatim transcription of the demo call, including timestamps and speaker identification. This is crucial for creating "Watch this moment" links.
+
+                **Microsite Structure & Content Requirements:**
+
+                1.  **HTML Boilerplate:** Include `<!DOCTYPE html>`, `<html>`, `<head>`, `<body>`.
+                2.  **Meta Tags:** Include `viewport` for responsiveness.
+                3.  **Title:** Use the `product_name` and `prospect_company` for the page title.
+                4.  **Tailwind CSS:** Load from CDN: `<script src="https://cdn.tailwindcss.com"></script>`.
+                5.  **Font:** Load Inter font via Google Fonts CDN in `<head>` and apply `font-family: 'Inter', sans-serif;` via a `<style>` block.
+                6.  **Overall Styling:**
+                    * Use a clean, modern design with `bg-gray-100` for the body.
+                    * Content should be in a white card (`bg-white rounded-lg shadow-md`) with good padding.
+                    * Apply rounded corners to elements.
+                    * Ensure appropriate spacing (padding, margin classes).
+                    * Center text for headers and CTAs.
+                7.  **Header Section:**
+                    * Prominent `<h1>` for the recap title (e.g., "Recap for [Prospect Company] - [Product Name] Demo").
+                    * `<p>` tag for "Presented by [Sales Rep's Name] ([Product Name])".
+                8.  **Summary Section (`<section>`):**
+                    * `<h2>` title: "Key Summary Points".
+                    * Unordered list (`<ul>`) with `list-disc list-inside` for `summary_points`.
+                9.  **Pain Points Discussed Section (`<section>`):**
+                    * `<h2>` title: "Pain Points Discussed".
+                    * Unordered list (`<ul>`) with `list-disc list-inside` for `pain_points_discussed`.
+                10. **Features Demonstrated Section (`<section>`):**
+                    * `<h2>` title: "Features Demonstrated".
+                    * If `features_demonstrated` is empty, use a `<p>` tag: "No features were explicitly demonstrated in this call."
+                    * If features exist, use an unordered list (`<ul>`). For each feature:
+                        * Display `name`.
+                        * Create a button/link `<a>` with Tailwind classes (e.g., `inline-block bg-blue-500 hover:bg-blue-600 text-white text-xs font-semibold py-1 px-2 rounded ml-2`) labeled "Watch this moment".
+                        * The `href` for this link MUST be `{{demo_recording_url}}#t={{timestamp_start_in_seconds}}`. Convert `HH:MM:SS` to total seconds for the hash (e.g., 00:00:30 becomes 30).
+                11. **Next Steps Section (`<section>`):**
+                    * `<h2>` title: "Next Steps".
+                    * Unordered list (`<ul>`) with `list-disc list-inside` for `next_steps`.
+                12. **Call to Action (CTA) (`<div>`):**
+                    * Centered `<div>`.
+                    * A prominent button `<a>` with Tailwind classes (e.g., `bg-blue-500 hover:bg-blue-700 text-white font-bold py-2 px-4 rounded`) labeled "Schedule a Follow-Up". This can point to a placeholder link (`#`).
+                13. **Strict HTML Output:** Output ONLY the complete HTML document. Do not include any other text, preambles, explanations, or conversational filler outside the HTML. **Ensure minimal newlines and whitespace within the HTML for a compact output.**
+                """
+    ),
+    response_model=HtmlContent,  # Agent will return an HtmlContent object containing the raw HTML string
+)

agents/transcription_agent.py

@@ -0,0 +1,44 @@
+from agno.agent import Agent
+from agno.models.google import Gemini
+from textwrap import dedent
+from pydantic import BaseModel
+
+
+class Transcription(BaseModel):
+    transcription: str
+
+
+transcription_agent = Agent(
+    model=Gemini(id="gemini-2.0-flash-lite", response_modalities=["text"]),
+    description=dedent(
+        """\
+                Highly accurate, verbatim audio-to-text transcription service.
+                Converts spoken words into a detailed textual record, preserving crucial temporal context and speaker identification."""
+    ),
+    instructions=dedent(
+        """\
+                Strictly follow these rules for verbatim transcription with timestamps and speaker identification.
+                Output the transcription as a continuous string, with each segment on a new line.
+
+                **Output Format:**
+                [HH:MM:SS - HH:MM:SS] Speaker Name: Transcribed verbatim speech
+
+                **Transcription Rules (Strictly Adhere to All):**
+
+                1.  **Verbatim Accuracy:** Transcribe every single word exactly as heard.
+                2.  **No Interpretation/Summarization:** Do not summarize, interpret, or rephrase speech. Transcribe only what is explicitly said.
+                3.  **Unclear Speech:** Use '[inaudible]' for any speech that cannot be clearly understood.
+                4.  **Pauses:** Indicate pauses longer than 2 seconds with '...' (three periods) directly within the transcribed text.
+                5.  **No Punctuation/Formatting:** Do not add any punctuation (commas, periods, question marks, etc.) or apply any text formatting (bold, italics).
+                6.  **Preserve Filler Words:** Include all filler words (e.g., 'um', 'uh', 'like', 'you know').
+
+                **Example of Desired Output:**
+                [00:00:00 - 00:00:05] Sales Rep: Good morning Jane thanks for joining the call
+                [00:00:05 - 00:00:12] Prospect: Hi Alice excited to learn more about the Microsite Pilot
+                [00:00:12 - 00:00:25] Sales Rep: Great today we're going to focus on how we automate post-demo follow-ups
+                [00:00:25 - 00:00:30] Prospect: My biggest pain point is the time spent summarizing
+                [00:00:30 - 00:00:45] Sales Rep: Exactly our key feature is the 'Instant Microsite Generation' let me show you that
+                """
+    ),
+    response_model=Transcription,
+)

app.py

@@ -0,0 +1,248 @@
+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
+            use_transcription_cache=use_cache_bool,
+        ):
+            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_default", 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()

pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "micrositepilot"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "agno>=1.5.10",
+    "fastapi>=0.115.12",
+    "google>=3.0.0",
+    "google-genai>=1.19.0",
+    "gradio>=5.33.0",
+    "netlify-python>=0.3.2",
+    "python-dotenv>=1.1.0",
+]

utils/netlify_deployment.py

@@ -0,0 +1,134 @@
+import os
+import uuid
+import requests
+import hashlib
+
+
+def deploy_html_file_with_digest(title, html_file_path, access_token=None):
+    """
+    Deploy a single HTML file to Netlify using the file digest method.
+
+    Args:
+        title (str): The title/name for the site
+        html_file_path (str): Path to the HTML file to deploy
+        access_token (str): Netlify personal access token (optional, will use env var if not provided)
+
+    Returns:
+        dict: Response containing site information and deploy details
+    """
+    # Use provided token or get from environment
+    token = access_token or os.getenv("NETLIFY_PERSONAL_ACCESS_TOKEN")
+    if not token:
+        raise ValueError("No Netlify access token provided")
+
+    # Generate a random site name
+    site_name = f"{title.lower().replace(' ', '-')}-{str(uuid.uuid4())[:8]}"
+
+    # Netlify API base URL
+    api_base = "https://api.netlify.com/api/v1"
+
+    # Headers for authentication
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "User-Agent": "MicrositePilot-Deployer",
+        "Content-Type": "application/json",
+    }
+
+    try:
+        # Step 1: Create a new site
+        site_data = {
+            "name": site_name,
+            "processing_settings": {"html": {"pretty_urls": True}},
+        }
+
+        site_response = requests.post(
+            f"{api_base}/sites", headers=headers, json=site_data
+        )
+        site_response.raise_for_status()
+        site_info = site_response.json()
+
+        site_id = site_info["id"]
+        site_url = site_info["url"]
+        admin_url = site_info["admin_url"]
+
+        # Step 2: Read the HTML file and calculate SHA1
+        with open(html_file_path, "rb") as f:
+            html_content = f.read()
+
+        # Calculate SHA1 hash
+        sha1_hash = hashlib.sha1(html_content).hexdigest()
+
+        # Step 3: Create deployment with file digest
+
+        deploy_data = {"files": {"/index.html": sha1_hash}}
+
+        deploy_response = requests.post(
+            f"{api_base}/sites/{site_id}/deploys",
+            headers=headers,
+            json=deploy_data,
+        )
+        deploy_response.raise_for_status()
+        deploy_info = deploy_response.json()
+
+        deploy_id = deploy_info["id"]
+        required_files = deploy_info.get("required", [])
+        deploy_url = deploy_info.get("deploy_url", "")
+        deploy_state = deploy_info.get("state", "unknown")
+
+        # Step 4: Upload required files
+        if sha1_hash in required_files:
+            file_headers = {
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "text/html",
+                "User-Agent": "MicrositePilot-Deployer",
+            }
+
+            upload_response = requests.put(
+                f"{api_base}/deploys/{deploy_id}/files/index.html",
+                headers=file_headers,
+                data=html_content,
+            )
+            upload_response.raise_for_status()
+            print(f"✅ File uploaded successfully!")
+        else:
+            print(f"ℹ️  File already exists on Netlify, no upload needed")
+
+        # Step 5: Check final deployment status
+        status_response = requests.get(
+            f"{api_base}/deploys/{deploy_id}", headers=headers
+        )
+        status_response.raise_for_status()
+        status_info = status_response.json()
+
+        final_state = status_info.get("state", "unknown")
+        final_url = status_info.get("deploy_url", deploy_url)
+
+        # Return comprehensive information
+        return {
+            "success": True,
+            "site": {
+                "id": site_id,
+                "name": site_name,
+                "url": site_url,
+                "admin_url": admin_url,
+            },
+        }
+
+    except requests.exceptions.RequestException as e:
+        return {
+            "success": False,
+            "error": str(e),
+            "message": f"Failed to deploy {title}",
+        }
+    except FileNotFoundError:
+        return {
+            "success": False,
+            "error": "File not found",
+            "message": f"HTML file {html_file_path} not found",
+        }
+    except Exception as e:
+        return {
+            "success": False,
+            "error": str(e),
+            "message": f"Unexpected error during deployment of {title}",
+        }

workflow.py

@@ -0,0 +1,375 @@
+from agno.workflow import Workflow, RunResponse, RunEvent
+from agents.transcription_agent import transcription_agent, Transcription
+from agents.site_builder_agent import microsite_builder_agent
+from agents.info_extractor_agent import info_extractor
+from utils.netlify_deployment import deploy_html_file_with_digest
+from textwrap import dedent
+from agno.agent import Agent
+from typing import Iterator, Union, Optional
+from logging import Logger
+from pathlib import Path
+from agno.media import Audio
+from dotenv import load_dotenv
+import requests
+import json
+from datetime import datetime
+
+load_dotenv()
+
+# It's good practice to get a logger instance here, though `logging` module needs configuration
+logger = Logger(__name__)
+
+
+class MicroSiteGenerator(Workflow):
+    description: str = dedent(
+        """\
+        An intelligent AI agent that seamlessly transforms product demo call recordings into personalized, interactive recap websites. This workflow orchestrates multiple AI agents to transcribe the demo, intelligently extract key discussion points and features, and dynamically assemble compelling, shareable microsites.
+    """
+    )
+
+    transcriber: Agent = transcription_agent
+    info_extractor: Agent = info_extractor
+    microsite_builder: Agent = microsite_builder_agent
+
+    def save_html_to_file(self, html_content: str) -> str:
+        """
+        Manually save HTML content to the microsites directory.
+
+        Args:
+            html_content: The HTML content to save
+
+        Returns:
+            str: The full path to the saved HTML file
+        """
+        # Create microsites directory if it doesn't exist
+        microsites_dir = Path(__file__).parent.parent / "microsites"
+        microsites_dir.mkdir(exist_ok=True)
+
+        # Generate filename with timestamp
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        filename = f"demo_{timestamp}.html"
+        file_path = microsites_dir / filename
+
+        try:
+            # Write HTML content to file
+            with open(file_path, "w", encoding="utf-8") as f:
+                f.write(html_content)
+
+            logger.info(f"HTML saved successfully to: {file_path}")
+            return str(file_path)
+
+        except Exception as e:
+            logger.error(f"Failed to save HTML file: {e}")
+            raise Exception(f"Could not save HTML file: {e}")
+
+    def run(
+        self,
+        audio_source: str,
+        audio_format: str,
+        use_transcription_cache: bool = True,
+    ) -> Iterator[RunResponse]:
+        logger.info("Microsite generation initiated.")
+
+        transcription_results: Optional[Transcription] = None
+        if use_transcription_cache:
+            transcription_results = self.get_cached_transcription(audio_source)
+            if transcription_results:
+                logger.info(f"Using cached transcription for {audio_source}")
+            else:
+                logger.info(
+                    f"No cached transcription found for {audio_source}, transcribing now."
+                )
+                transcription_results = self.transcribe_audio(
+                    audio_source, audio_format
+                )
+        if transcription_results:
+            self._add_transcription_to_cache(audio_source, transcription_results)
+            extracted_info: RunResponse = self.info_extractor.run(
+                message=transcription_results.transcription
+            )
+            extracted_info = self.remove_markdown_json_wrapper(extracted_info.content)
+            print(extracted_info)
+
+            microsite_builder_input = {
+                "extracted_info_json": extracted_info,
+                "raw_transcription": transcription_results.transcription,
+            }
+            site_html: RunResponse = microsite_builder_agent.run(
+                json.dumps(microsite_builder_input)
+            )
+
+            # Save HTML to filesystem using manual function
+            html_file_path = self.save_html_to_file(site_html.content.content)
+            logger.info(f"HTML saved to: {html_file_path}")
+
+            product_name = json.loads(extracted_info)["product_name"]
+
+            site_details = deploy_html_file_with_digest(
+                title=product_name,
+                html_file_path=html_file_path,
+            )
+
+            yield RunResponse(
+                content=site_details,
+                event=RunEvent.workflow_completed,
+            )
+        else:
+            yield RunResponse(
+                content="Site was not generated",
+                event=RunEvent.workflow_completed,
+            )
+
+        # transcription_results = self.transcribe_audio(audio_source, audio_format)
+        # if transcription_results:
+        #     yield RunResponse(
+        #         content=transcription_results.transcription,  # The transcription text
+        #         event=RunEvent.workflow_completed,
+        #     )
+        # else:
+        #     yield RunResponse(
+        #         content="Transcription failed.", event=RunEvent.workflow_completed
+        #     )
+        # extracted_info: RunResponse = self.info_extractor.run(
+        #     message=transcription_results.transcription
+        # )
+        # print(self.remove_markdown_json_wrapper(extracted_info.content))
+
+    def get_cached_transcription(
+        self, audio_source: Union[str, Path, bytes]
+    ) -> Optional[Transcription]:
+        """
+        Retrieves a cached transcription result for a given audio source.
+        """
+        # For caching, audio_source needs to be hashable. If it's bytes, convert to a string key.
+        cache_key = (
+            str(audio_source)
+            if isinstance(audio_source, (str, Path))
+            else f"bytes_hash_{hash(audio_source)}"
+        )
+        logger.info(f"Checking if cached transcription exists for {cache_key}.")
+        transcription_result = self.session_state.get("transcription_cache", {}).get(
+            cache_key
+        )
+        # Use model_validate to convert dict from cache back to Transcription object
+        return (
+            Transcription.model_validate(transcription_result)
+            if transcription_result and isinstance(transcription_result, dict)
+            else None
+        )
+
+    def _add_transcription_to_cache(
+        self, audio_source: Union[str, Path, bytes], transcription_result: Transcription
+    ):
+        """
+        Adds a transcription result to the session cache.
+        """
+        cache_key = (
+            str(audio_source)
+            if isinstance(audio_source, (str, Path))
+            else f"bytes_hash_{hash(audio_source)}"
+        )
+        logger.info(f"Saving transcription results for audio source: {cache_key}")
+        self.session_state.setdefault("transcription_cache", {})
+        # Store the Pydantic model as a dictionary
+        self.session_state["transcription_cache"][
+            cache_key
+        ] = transcription_result.model_dump()
+
+    def remove_markdown_json_wrapper(self, json_string_with_markdown: str) -> str:
+        """
+        Removes the '```json' prefix and '```' suffix from a string,
+        assuming the JSON content is wrapped in a Markdown code block.
+
+        Args:
+            json_string_with_markdown: The string containing the JSON wrapped in markdown.
+                                    Expected format: ```json\n{...json content...}\n```
+
+        Returns:
+            The cleaned JSON string without the markdown wrapper.
+        """
+        cleaned_string = json_string_with_markdown
+
+        # Remove '```json\n' from the start
+        if cleaned_string.startswith("```json\n"):
+            cleaned_string = cleaned_string[len("```json\n") :]
+
+        # Remove '\n```' from the end
+        if cleaned_string.endswith("\n```"):
+            cleaned_string = cleaned_string[: -len("\n```")]
+
+        return cleaned_string
+
+    # --- Caching Functions ---
+    def get_cached_transcription(
+        self, audio_source: Union[str, Path, bytes]
+    ) -> Optional[Transcription]:
+        """
+        Retrieves a cached transcription result for a given audio source.
+        """
+        # For caching, audio_source needs to be hashable. If it's bytes, convert to a string key.
+        cache_key = (
+            str(audio_source)
+            if isinstance(audio_source, (str, Path))
+            else f"bytes_hash_{hash(audio_source)}"
+        )
+        logger.info(f"Checking if cached transcription exists for {cache_key}.")
+        transcription_result = self.session_state.get("transcription_cache", {}).get(
+            cache_key
+        )
+        # Use model_validate to convert dict from cache back to Transcription object
+        return (
+            Transcription.model_validate(transcription_result)
+            if transcription_result and isinstance(transcription_result, dict)
+            else None
+        )
+
+    def _add_transcription_to_cache(
+        self, audio_source: Union[str, Path, bytes], transcription_result: Transcription
+    ):
+        """
+        Adds a transcription result to the session cache.
+        """
+        cache_key = (
+            str(audio_source)
+            if isinstance(audio_source, (str, Path))
+            else f"bytes_hash_{hash(audio_source)}"
+        )
+        logger.info(f"Saving transcription results for audio source: {cache_key}")
+        self.session_state.setdefault("transcription_cache", {})
+        # Store the Pydantic model as a dictionary
+        self.session_state["transcription_cache"][
+            cache_key
+        ] = transcription_result.model_dump()
+
+    # --- Audio Handling Function ---
+    def _download_audio(self, url: str) -> bytes:
+        """
+        Downloads audio from a given URL.
+        """
+        logger.info(f"Attempting to download audio from URL: {url}")
+        try:
+            response = requests.get(url, stream=True)
+            response.raise_for_status()  # Raise an exception for HTTP errors
+            return response.content
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Failed to download audio from {url}: {e}")
+            raise ValueError(f"Could not download audio from URL: {e}")
+
+    def _get_audio_bytes(self, source: Union[str, Path, bytes]) -> bytes:
+        """
+        Retrieves audio content as bytes from various sources (path, URL, or raw bytes).
+        """
+        if isinstance(source, bytes):
+            return source
+        elif isinstance(source, (str, Path)):
+            str_source = str(source)
+            if str_source.startswith(("http://", "https://")):
+                return self._download_audio(str_source)
+            return Path(str_source).read_bytes()
+        raise ValueError("Unsupported audio source type.")
+
+    # --- Transcription Execution Functions ---
+    def _run_transcription_agent(
+        self,
+        audio_source_bytes: bytes,
+        audio_format: str,
+    ):
+        """
+        Executes the transcription agent with the given audio bytes.
+        """
+        logger.info(f"Running transcription agent for audio format: {audio_format}")
+        try:
+            run_response: RunResponse = self.transcriber.run(
+                input="Transcribe this audio exactly as heard",
+                audio=[Audio(content=audio_source_bytes, format=audio_format)],
+            )
+            return run_response.content
+        except Exception as e:
+            logger.error(f"Transcription agent failed: {str(e)}")
+            return None
+
+    def transcribe_audio(
+        self,
+        audio_source: Union[str, Path, bytes],
+        audio_format: str = "wav",
+        num_attempts: int = 3,
+    ):
+        """
+        Manages the transcription process, including getting audio bytes and retrying the agent.
+        """
+        logger.info("Initiating audio transcription process.")
+        try:
+            audio_bytes = self._get_audio_bytes(audio_source)
+        except (ValueError, NotImplementedError) as e:
+            logger.error(f"Failed to get audio bytes: {str(e)}")
+            return None
+
+        for attempt in range(num_attempts):
+            transcription_response = self._run_transcription_agent(
+                audio_bytes, audio_format
+            )
+            if transcription_response:
+                logger.info(f"Transcription successful after {attempt + 1} attempt(s).")
+                return transcription_response
+            else:
+                logger.warning(
+                    f"Transcription attempt {attempt + 1}/{num_attempts} failed."
+                )
+        logger.error(
+            f"Transcription failed after {num_attempts} attempts for {audio_source}."
+        )
+        return None
+
+        # # --- Transcription Phase ---
+        # transcription_results: Optional[Transcription] = None
+        # if use_transcription_cache:
+        #     transcription_results = self.get_cached_transcription(audio_source)
+        #     if transcription_results:
+        #         logger.info(f"Using cached transcription for {audio_source}")
+        #         # Yield cached transcription as RunResponse
+        #         yield RunResponse(
+        #             content=f"Using cached transcription: {transcription_results.transcription}",
+        #             event=RunEvent.workflow_completed,
+        #         )
+        #         return
+        #     else:
+        #         logger.info(
+        #             f"No cached transcription found for {audio_source}, transcribing now."
+        #         )
+        #         transcription_results = self.transcribe_audio(
+        #             audio_source, audio_format
+        #         )
+        #         if transcription_results:
+        #             self._add_transcription_to_cache(
+        #                 audio_source, transcription_results
+        #             )
+        # else:
+        #     logger.info(
+        #         f"Transcription cache disabled, transcribing {audio_source} now."
+        #     )
+        #     transcription_results = self.transcribe_audio(audio_source, audio_format)
+        #     if transcription_results:
+        #         self._add_transcription_to_cache(audio_source, transcription_results)
+
+        # if transcription_results is None:
+        #     logger.error("Transcription was not successful. Workflow halted.")
+        #     yield RunResponse(
+        #         content="Transcription failed. Workflow halted.",
+        #         event=RunEvent.workflow_completed,
+        #     )
+        #     return
+
+        # # --- Information Extraction Phase ---
+        # logger.info("Transcription successful. Proceeding to information extraction.")
+        # # Run the info_extractor agent and yield its response
+        # yield from self.info_extractor.run(
+        #     input=transcription_results.transcription,  # Pass the raw string transcription to the extractor
+        #     stream=True,
+        # )
+
+        # # Cache the final result
+        # if (
+        #     self.info_extractor.run_response
+        #     and self.info_extractor.run_response.content
+        # ):
+        #     logger.info("Information extraction successful. Workflow completed.")