# retoor <retoor@molodetz.nl>
import logging
from pathlib import Path
from typing import Annotated
from fastapi import Depends, APIRouter, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from devplacepy.config import SEO_REPORTS_DIR
from devplacepy.models import SeoRunForm
from devplacepy.responses import respond
from devplacepy.schemas import SeoJobOut, SeoReportOut
from devplacepy.seo import base_seo_context, site_url, web_application_schema, website_schema
from devplacepy.services.jobs import queue
from devplacepy.services.jobs.seo.progress import hub
from devplacepy.services.manager import service_manager
from devplacepy.templating import templates
from devplacepy.utils import get_current_user, not_found, track_action
from devplacepy.dependencies import json_or_form
from ._shared import owner_for as _owner
logger = logging.getLogger(__name__)
router = APIRouter()
ACTIVE_STATES = (queue.PENDING, queue.RUNNING)
TERMINAL_STATES = (queue.DONE, queue.FAILED)
def _status_payload(job: dict) -> dict:
result = job.get("result", {})
done = job.get("status") == queue.DONE
uid = job.get("uid", "")
payload = job.get("payload", {})
return {
"uid": uid,
"kind": job.get("kind", ""),
"status": job.get("status", ""),
"target": result.get("target") or payload.get("url"),
"mode": result.get("mode") or payload.get("mode"),
"ws_url": f"/tools/seo/{uid}/ws",
"report_url": f"/tools/seo/{uid}/report" if done else None,
"score": result.get("score") if done else None,
"grade": result.get("grade") if done else None,
"page_count": int(result.get("page_count") or 0),
"error": job.get("error") or None,
"created_at": job.get("created_at"),
"completed_at": job.get("completed_at") or None,
}
@router.get("", response_class=HTMLResponse)
async def seo_page(request: Request):
user = get_current_user(request)
base = site_url(request)
description = (
"Run a deep SEO audit on any URL or sitemap: technical, on-page, structured "
"data, Core Web Vitals, accessibility and AI-readiness checks with live progress."
)
seo_ctx = base_seo_context(
request,
title="SEO Diagnostics",
description=description,
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Tools", "url": "/tools"},
{"name": "SEO Diagnostics", "url": "/tools/seo"},
],
schemas=[
website_schema(base),
web_application_schema(
"SEO Diagnostics", description, "/tools/seo", base
),
],
)
return templates.TemplateResponse(
request,
"tools/seo.html",
{**seo_ctx, "request": request, "user": user},
)
@router.post("/run")
async def seo_run(request: Request, data: Annotated[SeoRunForm, Depends(json_or_form(SeoRunForm))]):
owner_kind, owner_id = _owner(request)
active = [
job
for job in queue.list_jobs(kind="seo", owner=(owner_kind, owner_id))
if job.get("status") in ACTIVE_STATES
]
from devplacepy.services.audit import record as audit
if active:
audit.record(
request,
"seo.run.request",
result="denied",
summary=f"SEO diagnostics denied for {data.url}: audit already running",
metadata={"target": data.url, "reason": "active_job", "uid": active[0]["uid"]},
links=[audit.job(active[0]["uid"])],
)
return JSONResponse(
{
"error": {
"status": 429,
"message": "You already have an SEO audit running. Wait for it to finish.",
"uid": active[0]["uid"],
}
},
status_code=429,
)
uid = queue.enqueue(
"seo",
{"url": data.url, "mode": data.mode, "max_pages": data.max_pages, "allow_private": False},
owner_kind,
owner_id,
f"SEO: {data.url}"[:64],
)
audit.record(
request,
"seo.run.request",
summary=f"requested SEO diagnostics for {data.url}",
metadata={"target": data.url, "mode": data.mode, "max_pages": data.max_pages},
links=[audit.job(uid)],
)
if owner_kind == "user":
track_action(owner_id, "seo")
return JSONResponse(
{
"uid": uid,
"status_url": f"/tools/seo/{uid}",
"ws_url": f"/tools/seo/{uid}/ws",
}
)
@router.get("/{uid}")
async def seo_status(request: Request, uid: str):
job = queue.get_job(uid)
if not job or job.get("kind") != "seo":
raise not_found("Audit not found")
return JSONResponse(
SeoJobOut.model_validate(_status_payload(job)).model_dump(mode="json")
)
@router.get("/{uid}/report")
async def seo_report(request: Request, uid: str):
job = queue.get_job(uid)
if not job or job.get("kind") != "seo":
raise not_found("Audit not found")
result = job.get("result", {})
report = result.get("report", {}) if job.get("status") == queue.DONE else {}
context = {
"uid": uid,
"status": job.get("status", ""),
"target": report.get("target") or job.get("payload", {}).get("url"),
"mode": report.get("mode") or job.get("payload", {}).get("mode"),
"score": report.get("score"),
"grade": report.get("grade"),
"page_count": report.get("page_count", 0),
"counts": report.get("counts", {}),
"categories": report.get("categories", {}),
"pages": report.get("pages", []),
"checks": report.get("checks", []),
"site": report.get("site", {}),
"generated_at": report.get("generated_at"),
"request": request,
"user": get_current_user(request),
"meta_robots": "noindex,nofollow",
}
return respond(request, "tools/seo_report.html", context, model=SeoReportOut)
@router.get("/{uid}/screenshot/{index}")
async def seo_screenshot(request: Request, uid: str, index: int):
job = queue.get_job(uid)
if not job or job.get("kind") != "seo":
raise not_found("Screenshot not available")
path = (SEO_REPORTS_DIR / uid / f"page-{index}.png").resolve()
if not path.is_relative_to(SEO_REPORTS_DIR.resolve()) or not path.is_file():
raise not_found("Screenshot not available")
return FileResponse(path, media_type="image/png")
@router.websocket("/{uid}/ws")
async def seo_ws(websocket: WebSocket, uid: str):
await websocket.accept()
if not service_manager.owns_lock():
await websocket.close(code=4013)
return
job = queue.get_job(uid)
if not job or job.get("kind") != "seo":
await websocket.close(code=1008)
return
listener = hub.subscribe(uid)
try:
for frame in hub.snapshot(uid):
await websocket.send_json(frame)
current = queue.get_job(uid)
if current and current.get("status") in TERMINAL_STATES:
done = current.get("status") == queue.DONE
report = current.get("result", {}).get("report", {}) if done else {}
await websocket.send_json(
{
"type": "done" if done else "failed",
"status": current.get("status"),
"report_url": f"/tools/seo/{uid}/report",
"error": current.get("error") or None,
"report": report,
}
)
return
while True:
frame = await listener.get()
await websocket.send_json(frame)
if frame.get("type") in ("done", "failed"):
break
except WebSocketDisconnect:
pass
except Exception: # noqa: BLE001 - never crash the worker on a socket error
logger.exception("SEO websocket loop failed for %s", uid)
finally:
hub.unsubscribe(uid, listener)
Reference in New Issue View Git Blame Copy Permalink