add swagger to puppeteer
Build and Push Docker Images / build-and-push (push) Successful in 4m10s

This commit is contained in:
2026-04-23 15:21:50 +02:00
parent 2a3cafc128
commit 09ecad1e7c
5 changed files with 132 additions and 16 deletions
+61 -10
View File
@@ -14,7 +14,7 @@ from app.services.browser import (
get_resulting_url_service
)
router = APIRouter()
router = APIRouter(tags=["Browser"])
class ClickInteraction(BaseModel):
action: Literal["click"]
@@ -43,14 +43,21 @@ class Viewport(BaseModel):
class InteractionsRequest(BaseModel):
url: str
returnType: Optional[Literal["screenshot", "html"]] = "screenshot"
url: str = Field(..., description="Target URL (will be URL-decoded server-side).")
returnType: Optional[Literal["screenshot", "html"]] = Field(
"screenshot",
description="Return a PNG screenshot or HTML content.",
)
viewport: Optional[Viewport] = None
scrollX: Optional[int] = None
scrollY: Optional[int] = None
interactions: List[Interaction]
@router.get("/")
@router.get(
"/",
summary="Visit URL",
description="Navigate to `url` and return HTML content (cached unless `skipCache=true`).",
)
async def visit_url(url: str, skipCache: bool = False, x_api_key: Optional[str] = Header(None)):
# Validate API key
if not x_api_key or x_api_key != API_KEY:
@@ -78,7 +85,16 @@ async def visit_url(url: str, skipCache: bool = False, x_api_key: Optional[str]
print(f"Error visiting URL {decoded_url}: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/screenshot")
@router.get(
"/screenshot",
summary="Capture screenshot",
description="Capture a PNG screenshot for `url`.",
responses={
200: {"content": {"image/png": {}}},
401: {"description": "Invalid API key"},
500: {"description": "Screenshot capture failed"},
},
)
async def screenshot_url(url: str, fullPage: bool = True, x_api_key: Optional[str] = Header(None)):
"""Capture a screenshot of a website and return it as PNG"""
if not x_api_key or x_api_key != API_KEY:
@@ -102,7 +118,26 @@ async def screenshot_url(url: str, fullPage: bool = True, x_api_key: Optional[st
print(f"Error taking screenshot for URL {decoded_url}: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/interactions")
@router.post(
"/interactions",
summary="Run interactions",
description=(
"Run a sequence of interactions (click/type) on a page.\n\n"
"- `returnType=screenshot` returns `image/png`\n"
"- `returnType=html` returns `text/html`"
),
responses={
200: {
"content": {
"image/png": {},
"text/html": {},
"application/json": {},
}
},
401: {"description": "Invalid API key"},
500: {"description": "Interaction failed"},
},
)
async def run_interactions(payload: InteractionsRequest, x_api_key: Optional[str] = Header(None)):
if not x_api_key or x_api_key != API_KEY:
raise HTTPException(status_code=401, detail="Invalid API key")
@@ -137,7 +172,11 @@ async def run_interactions(payload: InteractionsRequest, x_api_key: Optional[str
print(f"Error running interactions on URL {decoded_url}: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/seo")
@router.get(
"/seo",
summary="Extract SEO information",
description="Extract SEO information from `url` (cached unless `skipCache=true`).",
)
async def extract_seo(url: str, skipCache: bool = False, x_api_key: Optional[str] = Header(None)):
"""Extract SEO information from a website"""
# Validate API key
@@ -165,7 +204,11 @@ async def extract_seo(url: str, skipCache: bool = False, x_api_key: Optional[str
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/meta")
@router.get(
"/meta",
summary="Extract meta tags",
description="Extract meta tags / Open Graph / Twitter card data from `url` (cached unless `skipCache=true`).",
)
async def extract_meta_tags(url: str, skipCache: bool = False, x_api_key: Optional[str] = Header(None)):
"""Extract meta tags from a website"""
# Validate API key
@@ -194,7 +237,11 @@ async def extract_meta_tags(url: str, skipCache: bool = False, x_api_key: Option
raise HTTPException(status_code=500, detail=str(e))
@router.get("/outgoing-calls")
@router.get(
"/outgoing-calls",
summary="Capture outgoing calls",
description="Capture outgoing API calls from `url` (cached unless `skipCache=true`).",
)
async def capture_outgoing_calls(url: str, skipCache: bool = False, x_api_key: Optional[str] = Header(None)):
"""Capture outgoing API calls from a website"""
# Validate API key
@@ -224,7 +271,11 @@ async def capture_outgoing_calls(url: str, skipCache: bool = False, x_api_key: O
raise HTTPException(status_code=500, detail=str(e))
@router.get("/resulting-url")
@router.get(
"/resulting-url",
summary="Get resulting URL",
description="Get the final URL after navigation/redirects for `url` (cached unless `skipCache=true`).",
)
async def get_resulting_url(url: str, skipCache: bool = False, x_api_key: Optional[str] = Header(None)):
"""Get the resulting URL after navigation (handles redirects)"""
# Validate API key