Blame for app.py

import os
import asyncio
import base64
import io
from typing import Optional, List, Dict, Any
from PIL import Image

import uvicorn
from fastapi import FastAPI
import gradio as gr
import spaces  # Hugging Face ZeroGPU module
import torch
from transformers import pipeline

from playwright.async_api import async_playwright
from mcp.server.fastmcp import FastMCP

# ---------------------------------------------------------
# 1. ZeroGPU Model Initialization & GPU-Decorated Functions
# ---------------------------------------------------------
print("Initializing ZeroGPU Pipeline...")
try:
    research_pipe = pipeline("text-generation", model="gpt2", device="cuda" if torch.cuda.is_available() else "cpu")
except Exception as e:
    print(f"Pipeline loading note: {e}")
    research_pipe = None

@spaces.GPU
def run_zerogpu_analysis(prompt: str) -> str:
    """ZeroGPU function decorated with @spaces.GPU."""
    if not research_pipe:
        return f"ZeroGPU offline. Input snippet: {prompt[:150]}..."
    
    formatted_input = f"Hybrid Research 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 Engine
# ---------------------------------------------------------
async def get_browser_page(p):
    browser = await p.chromium.launch(
        headless=True,
        args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage"]
    )
    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")

@mcp.tool()
async def search_web(query: str, max_results: int = 5) -> str:
    """Search the web via DuckDuckGo to discover candidate research URLs."""
    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 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_tables_and_data(url: str) -> str:
    """Extract HTML tables 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]):
                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:
                    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 visual_chart_analysis_gpu(url: str, question: str) -> str:
    """Run ZeroGPU visual/content reasoning on page layout 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"Query: {question}\nTitle: {title}\nURL: {url}\n\nContent:\n{text_content[:1500]}"
            analysis = run_zerogpu_analysis(prompt)
            return f"### ZeroGPU Analysis ({url}):\n\n{analysis}"
        except Exception as e:
            await browser.close()
            return f"Visual analysis error: {str(e)}"

# ---------------------------------------------------------
# 4. Gradio Interface
# ---------------------------------------------------------
def build_gradio_ui():
    with gr.Blocks(title="🎭 Playwright MCP Server on ZeroGPU") as demo:
        gr.Markdown(
            """
            # 🎭 Hybrid Research MCP Server (ZeroGPU + Docker)
            
            ### 📡 External MCP SSE Connection
            Point MCP clients to: `https://swroop1242-pywright-mcp-server.hf.space/sse`
            """
        )
        
        with gr.Tab("Web Search"):
            q_in = gr.Textbox(label="Search Query", value="ZeroGPU Hugging Face")
            s_btn = gr.Button("Search Web")
            s_out = gr.Markdown()
            s_btn.click(fn=lambda q: asyncio.run(search_web(q)), inputs=q_in, outputs=s_out)
            
        with gr.Tab("ZeroGPU Research"):
            url_in = gr.Textbox(label="Target URL", value="https://news.ycombinator.com")
            query_in = gr.Textbox(label="Research Question", value="Summarize top headlines")
            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=[url_in, query_in], outputs=r_out)
            
    return demo

# ---------------------------------------------------------
# 5. Unified FastAPI Application Setup
# ---------------------------------------------------------
app = FastAPI()

# Mount FastMCP SSE sub-application
sse_subapp = mcp.sse_app()
app.mount("/sse", sse_subapp)
app.mount("/messages", sse_subapp)

# Mount Gradio UI
demo_ui = build_gradio_ui()
app = gr.mount_gradio_app(app, demo_ui, path="/")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)