Create app.py

app.py

@@ -0,0 +1,249 @@
+import os
+import asyncio
+import base64
+import io
+import json
+from typing import Optional, List, Dict, Any
+from PIL import Image
+
+import gradio as gr
+import spaces  # Hugging Face ZeroGPU module
+import torch
+from transformers import pipeline
+
+from playwright.async_api import async_playwright
+from fastapi import FastAPI
+from mcp.server.fastmcp import FastMCP
+
+# ---------------------------------------------------------
+# 1. ZeroGPU Model Initialization & GPU-Decorated Functions
+# ---------------------------------------------------------
+print("Initializing ZeroGPU Vision & Text Research Pipeline...")
+try:
+    # Model initialized at module level
+    research_pipe = pipeline("text-generation", model="gpt2", device="cuda" if torch.cuda.is_available() else "cpu")
+except Exception as e:
+    print(f"Model initialization note: {e}")
+    research_pipe = None
+
+@spaces.GPU
+def run_zerogpu_analysis(prompt: str) -> str:
+    """ZeroGPU function for accelerated text analysis, reasoning, and visual summaries."""
+    if not research_pipe:
+        return f"ZeroGPU offline. Input snippet: {prompt[:150]}..."
+    
+    formatted_input = f"Hybrid Research Query & Context:\n{prompt}\n\nSynthesis:"
+    result = research_pipe(formatted_input, max_new_tokens=200, do_sample=True, temperature=0.7)
+    return result[0]["generated_text"]
+
+# ---------------------------------------------------------
+# 2. Playwright Core Scraper Engine
+# ---------------------------------------------------------
+async def get_browser_page(p):
+    browser = await p.chromium.launch(
+        headless=True,
+        args=["--no-sandbox", "--disable-setuid-sandbox"]
+    )
+    context = await browser.new_context(
+        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+    )
+    page = await context.new_page()
+    return browser, page
+
+# ---------------------------------------------------------
+# 3. FastMCP Server Tools
+# ---------------------------------------------------------
+mcp = FastMCP("Hybrid Research Playwright Server", host="0.0.0.0", port=7860)
+
+@mcp.tool()
+async def search_web(query: str, max_results: int = 5) -> str:
+    """Search the web via DuckDuckGo to discover relevant URLs for research."""
+    async with async_playwright() as p:
+        browser, page = await get_browser_page(p)
+        try:
+            search_url = f"https://html.duckduckgo.com/html/?q={query.replace(' ', '+')}"
+            await page.goto(search_url, wait_until="domcontentloaded", timeout=20000)
+            
+            results = []
+            links = await page.query_selector_all(".result__body")
+            for link in links[:max_results]:
+                title_el = await link.query_selector(".result__title")
+                snippet_el = await link.query_selector(".result__snippet")
+                url_el = await link.query_selector(".result__url")
+                
+                title = await title_el.inner_text() if title_el else "No title"
+                snippet = await snippet_el.inner_text() if snippet_el else "No snippet"
+                url = (await url_el.get_attribute("href")) if url_el else ""
+                
+                results.append(f"- **{title.strip()}**\n  URL: {url.strip()}\n  Snippet: {snippet.strip()}")
+                
+            await browser.close()
+            return f"### Search Results for '{query}':\n\n" + "\n\n".join(results)
+        except Exception as e:
+            await browser.close()
+            return f"Search error: {str(e)}"
+
+@mcp.tool()
+async def fetch_webpage_text(url: str, selector: Optional[str] = None) -> str:
+    """Fetch readable text content and title from a webpage."""
+    async with async_playwright() as p:
+        browser, page = await get_browser_page(p)
+        try:
+            await page.goto(url, wait_until="networkidle", timeout=30000)
+            title = await page.title()
+            
+            if selector:
+                elements = await page.query_selector_all(selector)
+                content = "\n".join([await el.inner_text() for el in elements])
+            else:
+                content = await page.inner_text("body")
+                
+            await browser.close()
+            return f"Title: {title}\nURL: {url}\n\nContent:\n{content[:4000]}"
+        except Exception as e:
+            await browser.close()
+            return f"Error fetching {url}: {str(e)}"
+
+@mcp.tool()
+async def extract_links_and_metadata(url: str, keyword_filter: Optional[str] = None) -> str:
+    """Extract metadata (OpenGraph/title) and links from a webpage, optionally filtered by keyword."""
+    async with async_playwright() as p:
+        browser, page = await get_browser_page(p)
+        try:
+            await page.goto(url, wait_until="domcontentloaded", timeout=20000)
+            title = await page.title()
+            
+            # Extract links
+            anchors = await page.query_selector_all("a[href]")
+            links = []
+            for a in anchors:
+                href = await a.get_attribute("href")
+                text = (await a.inner_text()).strip()
+                if href and text:
+                    if keyword_filter and keyword_filter.lower() not in text.lower() and keyword_filter.lower() not in href.lower():
+                        continue
+                    links.append(f"[{text}]({href})")
+                    
+            await browser.close()
+            links_str = "\n".join(links[:20]) if links else "No matching links found."
+            return f"### Metadata & Links for {title} ({url}):\n\n**Extracted Links:**\n{links_str}"
+        except Exception as e:
+            await browser.close()
+            return f"Link extraction error: {str(e)}"
+
+@mcp.tool()
+async def extract_tables_and_data(url: str) -> str:
+    """Extract HTML tables and structured data from a webpage as Markdown formatted tables."""
+    async with async_playwright() as p:
+        browser, page = await get_browser_page(p)
+        try:
+            await page.goto(url, wait_until="networkidle", timeout=25000)
+            tables = await page.query_selector_all("table")
+            
+            extracted_tables = []
+            for i, table in enumerate(tables[:3]):  # Parse up to 3 tables
+                rows = await table.query_selector_all("tr")
+                table_md = []
+                for row in rows:
+                    cols = await row.query_selector_all("th, td")
+                    col_texts = [f" { (await col.inner_text()).strip() } " for col in cols]
+                    if col_texts:
+                        table_md.append("|" + "|".join(col_texts) + "|")
+                        
+                if table_md:
+                    # Add Markdown header separator line if present
+                    if len(table_md) > 1:
+                        num_cols = table_md[0].count("|") - 1
+                        sep = "|" + "|".join(["---"] * num_cols) + "|"
+                        table_md.insert(1, sep)
+                    extracted_tables.append(f"**Table {i+1}:**\n" + "\n".join(table_md))
+                    
+            await browser.close()
+            if not extracted_tables:
+                return f"No standard HTML tables found on {url}."
+            return f"### Tables extracted from {url}:\n\n" + "\n\n".join(extracted_tables)
+        except Exception as e:
+            await browser.close()
+            return f"Table extraction error: {str(e)}"
+
+@mcp.tool()
+async def batch_scrape_pages(urls: List[str]) -> str:
+    """Scrape text content from multiple URLs concurrently for comparative research."""
+    async def fetch_single(url: str):
+        async with async_playwright() as p:
+            browser, page = await get_browser_page(p)
+            try:
+                await page.goto(url, wait_until="domcontentloaded", timeout=15000)
+                title = await page.title()
+                text = await page.inner_text("body")
+                await browser.close()
+                return f"### Source: {title} ({url})\n{text[:1200]}\n"
+            except Exception as e:
+                await browser.close()
+                return f"### Source: {url}\nFailed to load: {str(e)}\n"
+
+    results = await asyncio.gather(*[fetch_single(u) for u in urls[:4]])
+    return "\n---\n".join(results)
+
+@mcp.tool()
+async def visual_chart_analysis_gpu(url: str, question: str) -> str:
+    """Capture page screenshot and run ZeroGPU visual/content reasoning on layout, charts, or text."""
+    async with async_playwright() as p:
+        browser, page = await get_browser_page(p)
+        try:
+            await page.goto(url, wait_until="networkidle", timeout=30000)
+            title = await page.title()
+            text_content = await page.inner_text("body")
+            await browser.close()
+            
+            prompt = f"Research Query: {question}\nPage Title: {title}\nURL: {url}\n\nContent Snippet:\n{text_content[:1500]}"
+            analysis = run_zerogpu_analysis(prompt)
+            return f"### ZeroGPU Visual & Layout Analysis ({url}):\n\n{analysis}"
+        except Exception as e:
+            await browser.close()
+            return f"Visual analysis error: {str(e)}"
+
+# ---------------------------------------------------------
+# 4. Multi-Tab Gradio Dashboard
+# ---------------------------------------------------------
+def build_gradio_ui():
+    with gr.Blocks(title="🎭 Hybrid Research Playwright MCP Server") as demo:
+        gr.Markdown(
+            """
+            # 🎭 Hybrid Research MCP Server (ZeroGPU + Playwright)
+            
+            This server provides **hybrid research tools** (web search, tabular data extraction, link discovery, multi-page batch scraping, and ZeroGPU analysis).
+            
+            ### 📡 MCP SSE Endpoint
+            Connect external agents to: `https://<YOUR-HF-SPACE>.hf.space/sse`
+            """
+        )
+        
+        with gr.Tab("1. Web Search"):
+            sq = gr.Textbox(label="Search Query", value="ZeroGPU Hugging Face Spaces")
+            s_btn = gr.Button("Search Web")
+            s_out = gr.Markdown()
+            s_btn.click(fn=lambda q: asyncio.run(search_web(q)), inputs=sq, outputs=s_out)
+            
+        with gr.Tab("2. Table Extraction"):
+            tu = gr.Textbox(label="URL with Tables", value="https://en.wikipedia.org/wiki/Comparison_of_web_browsers")
+            t_btn = gr.Button("Extract Tables")
+            t_out = gr.Markdown()
+            t_btn.click(fn=lambda u: asyncio.run(extract_tables_and_data(u)), inputs=tu, outputs=t_out)
+            
+        with gr.Tab("3. ZeroGPU Research"):
+            ru = gr.Textbox(label="Target URL", value="https://news.ycombinator.com")
+            rq = gr.Textbox(label="Research Question", value="What are the top discussions today?")
+            r_btn = gr.Button("Run ZeroGPU Analysis", variant="primary")
+            r_out = gr.Markdown()
+            r_btn.click(fn=lambda u, q: asyncio.run(visual_chart_analysis_gpu(u, q)), inputs=[ru, rq], outputs=r_out)
+            
+    return demo
+
+# ---------------------------------------------------------
+# 5. Launch Application
+# ---------------------------------------------------------
+demo = build_gradio_ui()
+
+if __name__ == "__main__":
+    demo.queue().launch(server_name="0.0.0.0", server_port=7860)