Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| import modal | |
| import os | |
| import json | |
| # --- Configuration --- | |
| MODAL_APP_NAME = "sitegeist-ai-app" | |
| # Import the Modal app from the package | |
| try: | |
| from sitegeist_core import app as modal_app | |
| except ImportError: | |
| # Fallback for when package is not available locally | |
| modal_app = None | |
| print("Warning: sitegeist_core package not found. Modal functions may not be available locally.") | |
| # --- Gradio MCP Tool Definitions --- | |
| def analyze_specific_urls(urls_json: str, analysis_prompt: str = "Summarize the content and identify key themes."): | |
| """ | |
| MCP Tool: Analyzes web content from a specific list of URLs using swarm analysis. Is useful for finding content related to a specific topic and compare the content with other URLs. | |
| Args: | |
| urls_json (str): A JSON string representing a list of URLs. e.g., '["http://example.com", "http://another.com"]' | |
| analysis_prompt (str): The specific analysis to perform on the content. | |
| """ | |
| print(f"Tool 'analyze_specific_urls' received: prompt='{analysis_prompt}', urls_json='{urls_json}'") | |
| try: | |
| urls = json.loads(urls_json) | |
| if not isinstance(urls, list) or not all(isinstance(url, str) for url in urls): | |
| raise ValueError("Input must be a JSON string of a list of URLs.") | |
| if not urls: | |
| return json.dumps({"status": "error", "message": "URL list cannot be empty."}) | |
| except json.JSONDecodeError: | |
| return json.dumps({"status": "error", "message": "Invalid JSON format for URLs."}) | |
| except ValueError as ve: | |
| return json.dumps({"status": "error", "message": str(ve)}) | |
| result = None | |
| try: | |
| print(f"Calling Modal: swarm_analyze_urls for {len(urls)} URLs") | |
| modal_swarm_analyze = modal.Function.lookup(MODAL_APP_NAME, "swarm_analyze_urls") | |
| if modal_swarm_analyze is None: | |
| return json.dumps({"status": "error", "message": f"Could not find Modal function 'swarm_analyze_urls' in app '{MODAL_APP_NAME}'."}) | |
| result = modal_swarm_analyze.remote(urls=urls, analysis_prompt=analysis_prompt) | |
| return json.dumps(result, indent=2) | |
| except modal.exception.NotFoundError as e: | |
| print(f"Modal function not found: {e}") | |
| return json.dumps({"status": "error", "message": f"Modal function lookup failed. Ensure '{MODAL_APP_NAME}' is deployed and functions are correctly named. Details: {e}"}) | |
| except Exception as e: | |
| print(f"An unexpected error occurred in 'analyze_specific_urls': {e}") | |
| return json.dumps({"status": "error", "message": f"An unexpected error occurred: {str(e)}"}) | |
| def discover_and_analyze_web(search_prompt: str, num_urls: int = 100, analysis_prompt_override: str = None): | |
| """ | |
| MCP Tool: Discovers relevant URLs based on a search prompt using web search search and then analyzes as Marketing & Content Intelligence Engine assistant. | |
| The analysis uses the swarm mode and is useful for finding content related to a specific topic including SEO keywords. | |
| Args: | |
| search_prompt (str): The query to search for. | |
| num_urls (int): The number of URLs to discover (max 200, uses pagination for >50 results). | |
| analysis_prompt_override (str, optional): Specific prompt for the analysis phase. | |
| If None, a default based on search_prompt is used. | |
| """ | |
| print(f"Tool 'discover_and_analyze_web' received: search_prompt='{search_prompt}', num_urls={num_urls}, analysis_prompt_override='{analysis_prompt_override}'") | |
| if not search_prompt: | |
| return json.dumps({"status": "error", "message": "Search prompt cannot be empty."}) | |
| # Ensure num_urls is within a reasonable range for Tavily search | |
| num_urls = max(1, min(int(num_urls), 200)) # Cap at 200, uses pagination for >50 results | |
| # 1. Discover URLs (Using Tavily via Modal) | |
| try: | |
| print(f"Calling Modal: tavily_search_engine for query '{search_prompt}'") | |
| modal_tavily_search = modal.Function.lookup(MODAL_APP_NAME, "tavily_search_engine") | |
| if modal_tavily_search is None: | |
| return json.dumps({"status": "error", "message": f"Could not find Modal function 'tavily_search_engine' in app '{MODAL_APP_NAME}'."}) | |
| discovered_urls = modal_tavily_search.remote(query=search_prompt, num_results=num_urls) | |
| if not discovered_urls: | |
| return json.dumps({"status": "error", "message": "Search returned no URLs."}) | |
| except modal.exception.NotFoundError as e: | |
| print(f"Modal function not found: {e}") | |
| return json.dumps({"status": "error", "message": f"Modal function lookup failed. Ensure '{MODAL_APP_NAME}' is deployed and tavily_search_engine function exists. Details: {e}"}) | |
| except Exception as e: | |
| print(f"An unexpected error occurred in tavily search: {e}") | |
| return json.dumps({"status": "error", "message": f"An unexpected error occurred during search: {str(e)}"}) | |
| # 2. Determine the analysis prompt for the swarm analysis | |
| current_analysis_prompt = analysis_prompt_override if analysis_prompt_override else f"Analyze content related to the search query: '{search_prompt}'" | |
| print(f"Using analysis prompt for swarm: '{current_analysis_prompt}'") | |
| # 3. Call swarm_analyze_urls Modal function | |
| try: | |
| print(f"Calling Modal: swarm_analyze_urls for {len(discovered_urls)} discovered URLs") | |
| modal_swarm_analyze = modal.Function.lookup(MODAL_APP_NAME, "swarm_analyze_urls") | |
| if modal_swarm_analyze is None: | |
| return json.dumps({"status": "error", "message": f"Could not find Modal function 'swarm_analyze_urls' in app '{MODAL_APP_NAME}'."}) | |
| result = modal_swarm_analyze.remote(urls=discovered_urls, analysis_prompt=current_analysis_prompt) | |
| return json.dumps(result, indent=2) | |
| except modal.exception.NotFoundError as e: | |
| print(f"Modal function not found: {e}") | |
| return json.dumps({"status": "error", "message": f"Modal function lookup failed. Ensure '{MODAL_APP_NAME}' is deployed. Details: {e}"}) | |
| except Exception as e: | |
| print(f"An unexpected error occurred in 'discover_and_analyze_web': {e}") | |
| return json.dumps({"status": "error", "message": f"An unexpected error occurred during discovery/analysis: {str(e)}"}) | |
| # --- Gradio Interface --- | |
| with gr.Blocks(theme=gr.themes.Soft()) as demo: | |
| gr.Markdown("# 👻 SitegeistAI - Your On-Demand Marketing Strategist") | |
| gr.Markdown("Stop drowning in data. SitegeistAI transforms the web into your competitive advantage. " | |
| "Instantly analyze competitor pages to uncover their strategy, or unleash our AI to discover and distill key insights from top-ranking content on any topic. " | |
| "Go from question to actionable intelligence in seconds.") | |
| with gr.Tabs(): | |
| with gr.TabItem("Deep-Dive Analysis"): | |
| gr.Markdown("## Analyze Your Competition") | |
| gr.Markdown( | |
| "Paste in competitor URLs, product pages, or key articles to instantly extract their core strategies, themes, and talking points." | |
| ) | |
| specific_urls_input = gr.Textbox(label="Target URLs", placeholder='["https://competitor.com/pricing", "https://competitor.com/blog/big-announcement"]') | |
| specific_analysis_prompt_input = gr.Textbox(label="What insights do you need?", value="Identify the primary calls-to-action and target audience.") | |
| specific_submit_button = gr.Button("Run Deep-Dive Analysis") | |
| specific_output_json = gr.JSON(label="Your Intelligence Report") | |
| specific_submit_button.click( | |
| analyze_specific_urls, | |
| inputs=[specific_urls_input, specific_analysis_prompt_input], | |
| outputs=specific_output_json | |
| ) | |
| with gr.TabItem("Market & Trend Discovery"): | |
| gr.Markdown("## Master Any Topic") | |
| gr.Markdown( | |
| "Want to dominate a new market or understand the latest trends? " | |
| "Enter any topic, and SitegeistAI will find the most relevant online content and deliver a complete strategic summary." | |
| ) | |
| discover_search_prompt_input = gr.Textbox(label="Topic to Research", placeholder="e.g., emerging strategies for B2B SaaS marketing") | |
| discover_num_urls_input = gr.Number(label="Number of Sources to Analyze", value=10, minimum=1, maximum=200, step=1, info="We'll find and analyze the top results for your topic.") | |
| discover_analysis_prompt_input = gr.Textbox( | |
| label="Refine Your Analysis (Optional)", | |
| placeholder="Focus on summarizing the common advice for beginners." | |
| ) | |
| discover_submit_button = gr.Button("Discover & Generate Insights") | |
| discover_output_json = gr.JSON(label="Your Market Overview") | |
| discover_submit_button.click( | |
| discover_and_analyze_web, | |
| inputs=[discover_search_prompt_input, discover_num_urls_input, discover_analysis_prompt_input], | |
| outputs=discover_output_json | |
| ) | |
| if __name__ == "__main__": | |
| print("Attempting to launch Gradio hackaton demo ...") | |
| demo.launch(mcp_server=True) | |