|
# retoor <retoor@molodetz.nl>
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse, Response
|
|
|
|
from devplacepy.responses import respond
|
|
from devplacepy.schemas import AdminBotsOut
|
|
from devplacepy.seo import base_seo_context, site_url, website_schema
|
|
from devplacepy.services.bot.monitor import monitor
|
|
from devplacepy.services.manager import service_manager
|
|
from devplacepy.utils import not_found, require_admin
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _service_describe() -> dict:
|
|
service = service_manager.get_service("bots")
|
|
if service is None:
|
|
return {}
|
|
try:
|
|
return service.describe()
|
|
except Exception as e:
|
|
logger.warning("bots describe failed: %s", e)
|
|
return {}
|
|
|
|
|
|
def _frames_payload() -> dict:
|
|
described = _service_describe()
|
|
metrics = described.get("metrics") or {}
|
|
frames = []
|
|
for frame in metrics.get("frames", []):
|
|
slot = frame.get("slot", 0)
|
|
captured_at = frame.get("captured_at", 0)
|
|
frames.append(
|
|
{
|
|
**frame,
|
|
"frame_url": f"/admin/bots/{slot}/frame.jpg?t={captured_at}",
|
|
}
|
|
)
|
|
return {
|
|
"frames": frames,
|
|
"enabled": bool(described.get("enabled")),
|
|
"service_status": described.get("status", ""),
|
|
}
|
|
|
|
|
|
@router.get("/bots", response_class=HTMLResponse)
|
|
async def bots_monitor(request: Request):
|
|
admin = require_admin(request)
|
|
payload = _frames_payload()
|
|
base = site_url(request)
|
|
seo_ctx = base_seo_context(
|
|
request,
|
|
title="Bot Monitor - Admin",
|
|
description="Live low-quality screenshots of every running bot persona.",
|
|
robots="noindex,nofollow",
|
|
breadcrumbs=[
|
|
{"name": "Home", "url": "/feed"},
|
|
{"name": "Admin", "url": "/admin"},
|
|
{"name": "Bot Monitor", "url": "/admin/bots"},
|
|
],
|
|
schemas=[website_schema(base)],
|
|
)
|
|
|
|
return respond(
|
|
request,
|
|
"admin_bots.html",
|
|
{
|
|
**seo_ctx,
|
|
"request": request,
|
|
"user": admin,
|
|
**payload,
|
|
"admin_section": "bots",
|
|
},
|
|
model=AdminBotsOut,
|
|
)
|
|
|
|
|
|
@router.get("/bots/data")
|
|
async def bots_monitor_data(request: Request):
|
|
require_admin(request)
|
|
return JSONResponse(_frames_payload())
|
|
|
|
|
|
@router.get("/bots/{slot}/frame.jpg")
|
|
async def bots_monitor_frame(request: Request, slot: int):
|
|
require_admin(request)
|
|
image = monitor.read_image(slot)
|
|
if image is None:
|
|
raise not_found("Frame not available")
|
|
return Response(
|
|
content=image,
|
|
media_type="image/jpeg",
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|