app.py

from fastapi import FastAPI
from playwright.async_api import async_playwright
from pydantic import BaseModel
from typing import List, Dict, Any
import logging

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

class Step(BaseModel):
    tool: str
    parameters: Dict[str, Any]

class Payload(BaseModel):
    steps: List[Step]

@app.api_route("/", methods=["GET", "POST"])
async def automate(payload: Payload):
    logger.info(f"Received payload:")
    async with async_playwright() as p:
        try:
            browser = await p.chromium.launch(headless=True, timeout=180000)
            page = await browser.new_page()
            # デフォルトのナビゲーション／操作タイムアウトを180秒に設定
            page.set_default_navigation_timeout(180000)
            page.set_default_timeout(180000)
            for step in payload.steps:
                tool = step.tool
                params = step.parameters
                logger.info(f"Executing step: {tool}")
                if tool == "browser_navigate":
                    await page.goto(params["url"])
                elif tool == "browser_type":
                    await page.fill(params["element"], params["text"])
                    if params.get("submit", False):
                        await page.press(params["element"], "Enter")
                elif tool == "browser_click":
                    await page.click(params["element"])
                elif tool == "browser_snapshot":
                    content = await page.content()
                    return {"snapshot": content}
                else:
                    return {"error": f"Unknown tool: {tool}"}
            return {"status": "completed"}
        except Exception as e:
            logger.error(f"Error during execution: {str(e)}", exc_info=True)
            return {"error": str(e)}
        finally:
            if 'browser' in locals():
                await browser.close()