|
|
|
|
|
from fastapi import FastAPI, HTTPException, Header |
|
|
from pydantic import BaseModel |
|
|
from playwright.async_api import async_playwright |
|
|
import os |
|
|
|
|
|
app = FastAPI() |
|
|
|
|
|
|
|
|
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") |
|
|
|
|
|
class CommandRequest(BaseModel): |
|
|
command: str |
|
|
|
|
|
@app.post("/run-command") |
|
|
async def run_command(req: CommandRequest, authorization: str = Header(None)): |
|
|
if authorization != f"Bearer {GEMINI_API_KEY}": |
|
|
raise HTTPException(status_code=401, detail="Unauthorized") |
|
|
|
|
|
|
|
|
|
|
|
if "example.com" in req.command.lower(): |
|
|
async with async_playwright() as p: |
|
|
browser = await p.chromium.launch(headless=True) |
|
|
page = await browser.new_page() |
|
|
await page.goto("https://example.com") |
|
|
screenshot = await page.screenshot() |
|
|
await browser.close() |
|
|
return {"status": "done", "screenshot_bytes": len(screenshot)} |
|
|
|
|
|
return {"status": "command not recognized"} |
|
|
|
|
|
@app.get("/") |
|
|
async def root(): |
|
|
return {"message": "Playwright + Gemini API running. POST /run-command to automate."} |
|
|
|