Update app.py

app.py

@@ -9,26 +9,18 @@ from datetime import datetime
 from io import BytesIO
 from typing import Dict, List, Optional, Any
 from contextlib import asynccontextmanager
+from pathlib import Path
 
-import gradio as gr
 import uvicorn
 from fastapi import FastAPI, HTTPException, BackgroundTasks
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from fastapi.responses import FileResponse, HTMLResponse
-from fastapi.staticfiles import StaticFiles
-from pathlib import Path
 from pydantic import BaseModel
 from playwright.async_api import async_playwright, Browser, BrowserContext, Page
 from selenium import webdriver
 from selenium.webdriver.chrome.options import Options
-from selenium.webdriver.chrome.service import Service
-from selenium.webdriver.common.by import By
-from selenium.webdriver.support.ui import WebDriverWait
-from selenium.webdriver.support import expected_conditions as EC
-from selenium.common.exceptions import TimeoutException, WebDriverException
 from bs4 import BeautifulSoup
-from PIL import Image
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -103,8 +95,6 @@ app = FastAPI(
 )
 
 # ---------- serve the single-page UI at “/” ----------
-#BASE_DIR = pathlib.Path(__file__).resolve().parent
-#UI_FILE  = BASE_DIR / "browser_automation_ui.html"   # adjust if you stored it elsewhere
 BASE_DIR = Path(__file__).resolve().parent
 UI_FILE  = BASE_DIR / "browser_automation_ui.html"
 
@@ -126,17 +116,16 @@ app.add_middleware(
 )
 
 # Utility functions
-def get_chrome_options():
-    """Get Chrome options for Selenium"""
-    options = Options()
-    options.add_argument('--headless')
-    options.add_argument('--no-sandbox')
-    options.add_argument('--disable-dev-shm-usage')
-    options.add_argument('--disable-gpu')
-    options.add_argument('--window-size=1920,1080')
-    return options
+def _chrome_opts():
+    o = Options()
+    o.add_argument("--headless")
+    o.add_argument("--no-sandbox")
+    o.add_argument("--disable-dev-shm-usage")
+    o.add_argument("--disable-gpu")
+    o.add_argument("--window-size=1920,1080")
+    return o
 
-def cleanup_old_sessions():
+def _purge_idle():
     """Clean up sessions older than 1 hour"""
     current_time = time.time()
     expired_sessions = []
@@ -148,191 +137,159 @@ def cleanup_old_sessions():
     for session_id in expired_sessions:
         asyncio.create_task(close_browser_session(session_id))
 
-async def close_browser_session(session_id: str):
-    """Close a specific browser session"""
-    if session_id in browser_instances:
-        session = browser_instances[session_id]
-        
-        # Close Playwright session
-        if 'playwright_page' in session:
-            try:
-                await session['playwright_page'].close()
-                await session['playwright_context'].close()
-            except Exception as e:
-                logger.error(f"Error closing Playwright session {session_id}: {e}")
-        
-        # Close Selenium session
-        if 'selenium_driver' in session:
-            try:
-                session['selenium_driver'].quit()
-            except Exception as e:
-                logger.error(f"Error closing Selenium session {session_id}: {e}")
-        
-        del browser_instances[session_id]
-        logger.info(f"Closed browser session: {session_id}")
+async def _close_session(session_id: str):
+    if session_id not in browser_instances:
+        return
+    sess = browser_instances[session_id]
+    # Playwright
+    try:
+        await sess['playwright_page'].close()
+        await sess['playwright_context'].close()
+    except Exception:
+        pass
+
+    del browser_instances[session_id]
+    logger.info("Closed session %s", session_id)
+
+@app.on_event("startup")
+async def _startup():
+    async def _cleaner():
+        while True:
+            _purge_idle()
+            await asyncio.sleep(300)
+    asyncio.create_task(_cleaner())
 
 # API Endpoints
 
 @app.get("/health")
-async def health_check():
-    """Health check endpoint"""
-    return {"status": "healthy", "timestamp": datetime.now().isoformat()}
+async def health():
+    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
 
 @app.post("/api/browser/launch")
-async def launch_browser(request: BrowserLaunchRequest):
+async def launch(request: BrowserLaunchRequest):
     """Launch a new browser instance"""
     session_id = str(uuid.uuid4())
     
     try:
         # Launch Playwright browser
-        context = await browser_pool.new_context(
+        ctx = await browser_pool.new_context(
             viewport={'width': request.width, 'height': request.height},
             user_agent=request.user_agent
         )
-        page = await context.new_page()
-        
-        # Launch Selenium browser
-        chrome_options = get_chrome_options()
-        if request.user_agent:
-            chrome_options.add_argument(f'--user-agent={request.user_agent}')
-        
-        selenium_driver = webdriver.Chrome(options=chrome_options)
-        selenium_driver.set_window_size(request.width, request.height)
+        page = await ctx.new_page()
         
         # Store session
         browser_instances[session_id] = {
-            'playwright_context': context,
-            'playwright_page': page,
-            'selenium_driver': selenium_driver,
-            'created_at': time.time(),
-            'config': request.dict()
+            "playwright_context": ctx,
+            "playwright_page": page,
+            "selenium_driver": driver,
+            "created_at": time.time(),
+            "config": request.dict()
         }
-        
-        logger.info(f"Launched browser session: {session_id}")
-        return {"session_id": session_id, "status": "launched"}
-        
+        logger.info("Launched session %s", session_id)
+        return {"session_id": session_id, "status": "launched"} 
     except Exception as e:
         logger.error(f"Error launching browser: {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
 @app.post("/api/browser/navigate")
-async def navigate_to_url(request: NavigateRequest):
+async def navigate(req: NavigateRequest):
     """Navigate to a URL"""
-    if request.session_id not in browser_instances:
+    if req.session_id not in browser_instances:
         raise HTTPException(status_code=404, detail="Session not found")
-    
-    session = browser_instances[request.session_id]
-    
+    sess = browser_instances[req.session_id]
     try:
         # Navigate with Playwright
-        await session['playwright_page'].goto(request.url, wait_until=request.wait_until)
-        
-        # Navigate with Selenium
-        session['selenium_driver'].get(request.url)
-        
-        return {"status": "navigated", "url": request.url}
-        
+        await sess["playwright_page"].goto(req.url, wait_until=req.wait_until)
+        return {"status": "navigated", "url": req.url}
     except Exception as e:
-        logger.error(f"Error navigating to {request.url}: {e}")
+        logger.error(f"Error navigating to {req.url}: {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
 @app.post("/api/browser/screenshot")
-async def take_screenshot(request: ScreenshotRequest):
+async def screenshot(req: ScreenshotRequest):
     """Take a screenshot"""
-    if request.session_id not in browser_instances:
+    if req.session_id not in browser_instances:
         raise HTTPException(status_code=404, detail="Session not found")
-    
-    session = browser_instances[request.session_id]
+    sess = browser_instances[req.session_id]
     
     try:
-        if request.selector:
-            # Screenshot specific element with Playwright
-            element = await session['playwright_page'].locator(request.selector).first
-            screenshot_bytes = await element.screenshot()
+        if req.selector:
+            el = sess["playwright_page"].locator(req.selector).first
+            png = await el.screenshot()
         else:
             # Full page screenshot with Playwright
-            screenshot_bytes = await session['playwright_page'].screenshot(
-                full_page=request.full_page
-            )
-        
+            png = await sess["playwright_page"].screenshot(full_page=req.full_page)
         # Convert to base64
-        screenshot_b64 = base64.b64encode(screenshot_bytes).decode()
-        
-        return {
-            "screenshot": screenshot_b64,
-            "format": "png",
-            "timestamp": datetime.now().isoformat()
-        }
-        
+        b64 = base64.b64encode(png).decode()
+        return {"screenshot": b64, "format": "png", "timestamp": datetime.utcnow().isoformat()}
     except Exception as e:
         logger.error(f"Error taking screenshot: {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
+@app.delete("/api/browser/close/{session_id}")
+async def close(session_id: str):
+    if session_id not in browser_instances:
+        raise HTTPException(status_code=404, detail="Session not found")
+    await _close_session(session_id)
+    return {"status": "closed", "session_id": session_id}
+
+# --------------------------------------------------------------------------- #
+#  API – Element-level
+# --------------------------------------------------------------------------- #
+    
 @app.post("/api/elements/action")
-async def perform_element_action(request: ElementActionRequest):
+async def element_action(req: ElementActionRequest):
     """Perform action on an element"""
-    if request.session_id not in browser_instances:
+    if req.session_id not in browser_instances:
         raise HTTPException(status_code=404, detail="Session not found")
-    
-    session = browser_instances[request.session_id]
+    sess = browser_instances[req.session_id]
+    page = sess["playwright_page"]
+    el = page.locator(req.selector).first
     
     try:
-        page = session['playwright_page']
-        element = page.locator(request.selector).first
-        
-        if request.action == "click":
-            await element.click()
-        elif request.action == "type":
-            await element.fill(request.value or "")
-        elif request.action == "scroll":
-            await element.scroll_into_view_if_needed()
-        elif request.action == "hover":
-            await element.hover()
+        if req.action == "click":
+            await el.click()
+        elif req.action == "type":
+            await el.fill(req.value or "")
+        elif req.action == "scroll":
+            await el.scroll_into_view_if_needed()
+        elif req.action == "hover":
+            await el.hover()
+        elif req.action == "textContent":
+            text = await el.text_content() or ""
+            return {"status": "completed", "action": "textContent", "text": text}
         else:
-            raise HTTPException(status_code=400, detail="Invalid action")
-        
-        return {"status": "completed", "action": request.action}
-        
+            raise ValueError("unknown action")
+        return {"status": "completed", "action": req.action}   
     except Exception as e:
-        logger.error(f"Error performing action {request.action}: {e}")
+        logger.error(f"Error performing action {req.action}: {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
 @app.get("/api/elements/inspect/{session_id}")
-async def inspect_page_elements(session_id: str):
+async def inspect(session_id: str):
     """Get all interactive elements on the page"""
     if session_id not in browser_instances:
         raise HTTPException(status_code=404, detail="Session not found")
-    
-    session = browser_instances[session_id]
+    page = browser_instances[session_id]["playwright_page"]
+    soup = BeautifulSoup(await page.content(), "html.parser")
+    selectors = [
+        'a','button','input','select','textarea',
+        '[onclick]','[href]','[role="button"]'
+    ]
+    out = []
     
     try:
-        page = session['playwright_page']
-        
-        # Get page content
-        content = await page.content()
-        soup = BeautifulSoup(content, 'html.parser')
-        
-        # Find interactive elements
-        interactive_selectors = [
-            'a', 'button', 'input', 'select', 'textarea', 
-            '[onclick]', '[href]', '[role="button"]'
-        ]
-        
-        elements = []
-        for selector in interactive_selectors:
-            found_elements = soup.select(selector)
-            for i, elem in enumerate(found_elements):
-                element_info = {
-                    'tag': elem.name,
-                    'selector': f"{selector}:nth-of-type({i+1})",
-                    'text': elem.get_text(strip=True)[:100],
-                    'attributes': dict(elem.attrs),
-                    'type': elem.get('type', 'N/A')
-                }
-                elements.append(element_info)
-        
-        return {"elements": elements, "total_count": len(elements)}
-        
+        for sel in selectors:
+        for idx, elem in enumerate(soup.select(sel)):
+            out.append({
+                "tag": elem.name,
+                "selector": f"{sel}:nth-of-type({idx+1})",
+                "text": elem.get_text(strip=True)[:100],
+                "attributes": dict(elem.attrs),
+                "type": elem.get("type", "N/A")
+            })
+        return {"elements": out, "total_count": len(out)}
     except Exception as e:
         logger.error(f"Error inspecting elements: {e}")
         raise HTTPException(status_code=500, detail=str(e))
@@ -374,15 +331,6 @@ async def scrape_content(request: ScrapeRequest):
         logger.error(f"Error scraping content: {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
-@app.delete("/api/browser/close/{session_id}")
-async def close_browser(session_id: str):
-    """Close a browser session"""
-    if session_id not in browser_instances:
-        raise HTTPException(status_code=404, detail="Session not found")
-    
-    await close_browser_session(session_id)
-    return {"status": "closed", "session_id": session_id}
-
 @app.get("/api/sessions")
 async def list_sessions():
     """List all active browser sessions"""
@@ -396,21 +344,5 @@ async def list_sessions():
     
     return {"sessions": sessions, "total_count": len(sessions)}
 
-# Background task to cleanup old sessions
-@app.on_event("startup")
-async def startup_event():
-    async def cleanup_task():
-        while True:
-            cleanup_old_sessions()
-            await asyncio.sleep(300)  # Clean up every 5 minutes
-    
-    asyncio.create_task(cleanup_task())
-
 if __name__ == "__main__":
-    uvicorn.run(
-        "app:app",
-        host="0.0.0.0",
-        port=7860,
-        reload=False,
-        workers=1
-    )
+    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)