357 lines
12 KiB
Python
357 lines
12 KiB
Python
"""Worker activity router — exposes recent background task status."""
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
from pydantic import BaseModel
|
|
|
|
from app.database import get_db
|
|
from app.models.cad_file import CadFile, ProcessingStatus
|
|
from app.models.order_item import OrderItem
|
|
from app.models.order import Order
|
|
from app.models.order_line import OrderLine
|
|
from app.models.product import Product
|
|
from app.models.user import User
|
|
from app.utils.auth import get_current_user, require_admin_or_pm
|
|
|
|
router = APIRouter(prefix="/worker", tags=["worker"])
|
|
|
|
|
|
class CadActivityEntry(BaseModel):
|
|
cad_file_id: str
|
|
original_name: str
|
|
file_size: int | None
|
|
processing_status: str
|
|
error_message: str | None
|
|
updated_at: str
|
|
created_at: str
|
|
order_numbers: list[str]
|
|
render_log: dict | None
|
|
|
|
|
|
class RenderJobEntry(BaseModel):
|
|
order_line_id: str
|
|
order_number: str | None
|
|
product_name: str | None
|
|
output_type_name: str | None
|
|
render_status: str
|
|
render_backend_used: str | None
|
|
flamenco_job_id: str | None
|
|
render_started_at: str | None
|
|
render_completed_at: str | None
|
|
updated_at: str
|
|
|
|
|
|
class WorkerActivity(BaseModel):
|
|
cad_processing: list[CadActivityEntry]
|
|
active_count: int # files currently in "processing" state
|
|
failed_count: int # files in "failed" state (recent 50)
|
|
render_jobs: list[RenderJobEntry] = []
|
|
render_active_count: int = 0
|
|
render_failed_count: int = 0
|
|
|
|
|
|
@router.get("/activity", response_model=WorkerActivity)
|
|
async def get_worker_activity(
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""
|
|
Return recent CAD file processing activity.
|
|
|
|
Shows the last 30 processed/failed/processing CAD files so the user can
|
|
see what the worker is doing without needing Flower or Celery logs.
|
|
"""
|
|
# Recent CadFiles ordered by last update, with order_items to resolve order numbers
|
|
result = await db.execute(
|
|
select(CadFile)
|
|
.order_by(CadFile.updated_at.desc())
|
|
.limit(30)
|
|
)
|
|
cad_files = result.scalars().all()
|
|
|
|
if not cad_files:
|
|
return WorkerActivity(cad_processing=[], active_count=0, failed_count=0)
|
|
|
|
# Fetch order items referencing these CAD files in one query
|
|
cad_ids = [cf.id for cf in cad_files]
|
|
items_result = await db.execute(
|
|
select(OrderItem)
|
|
.options(selectinload(OrderItem.order))
|
|
.where(OrderItem.cad_file_id.in_(cad_ids))
|
|
)
|
|
items = items_result.scalars().all()
|
|
|
|
# Build cad_file_id → list[order_number] mapping
|
|
from collections import defaultdict
|
|
cad_to_orders: dict[str, list[str]] = defaultdict(list)
|
|
for item in items:
|
|
if item.order and item.order.order_number:
|
|
key = str(item.cad_file_id)
|
|
if item.order.order_number not in cad_to_orders[key]:
|
|
cad_to_orders[key].append(item.order.order_number)
|
|
|
|
entries = []
|
|
for cf in cad_files:
|
|
entries.append(CadActivityEntry(
|
|
cad_file_id=str(cf.id),
|
|
original_name=cf.original_name or "unknown",
|
|
file_size=getattr(cf, "file_size", None),
|
|
processing_status=cf.processing_status.value if cf.processing_status else "unknown",
|
|
error_message=getattr(cf, "error_message", None),
|
|
updated_at=cf.updated_at.isoformat() if cf.updated_at else datetime.utcnow().isoformat(),
|
|
created_at=cf.created_at.isoformat() if cf.created_at else datetime.utcnow().isoformat(),
|
|
order_numbers=cad_to_orders.get(str(cf.id), []),
|
|
render_log=getattr(cf, "render_log", None),
|
|
))
|
|
|
|
active_count = sum(
|
|
1 for cf in cad_files
|
|
if cf.processing_status == ProcessingStatus.processing
|
|
)
|
|
failed_count = sum(
|
|
1 for cf in cad_files
|
|
if cf.processing_status == ProcessingStatus.failed
|
|
)
|
|
|
|
# ── Render job activity ──────────────────────────────────────────────
|
|
render_result = await db.execute(
|
|
select(OrderLine)
|
|
.options(
|
|
selectinload(OrderLine.product),
|
|
selectinload(OrderLine.output_type),
|
|
selectinload(OrderLine.order),
|
|
)
|
|
.where(OrderLine.output_type_id.isnot(None))
|
|
.where(OrderLine.render_status != "pending")
|
|
.order_by(OrderLine.updated_at.desc())
|
|
.limit(30)
|
|
)
|
|
render_lines = render_result.scalars().all()
|
|
|
|
render_entries = []
|
|
for rl in render_lines:
|
|
render_entries.append(RenderJobEntry(
|
|
order_line_id=str(rl.id),
|
|
order_number=rl.order.order_number if rl.order else None,
|
|
product_name=rl.product.name if rl.product else None,
|
|
output_type_name=rl.output_type.name if rl.output_type else None,
|
|
render_status=rl.render_status,
|
|
render_backend_used=rl.render_backend_used,
|
|
flamenco_job_id=rl.flamenco_job_id,
|
|
render_started_at=rl.render_started_at.isoformat() if rl.render_started_at else None,
|
|
render_completed_at=rl.render_completed_at.isoformat() if rl.render_completed_at else None,
|
|
updated_at=rl.updated_at.isoformat(),
|
|
))
|
|
|
|
render_active = sum(1 for rl in render_lines if rl.render_status == "processing")
|
|
render_failed = sum(1 for rl in render_lines if rl.render_status == "failed")
|
|
|
|
return WorkerActivity(
|
|
cad_processing=entries,
|
|
active_count=active_count,
|
|
failed_count=failed_count,
|
|
render_jobs=render_entries,
|
|
render_active_count=render_active,
|
|
render_failed_count=render_failed,
|
|
)
|
|
|
|
|
|
@router.get("/render-log/{order_line_id}")
|
|
async def get_render_log(
|
|
order_line_id: str,
|
|
after: int = 0,
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""Return render log entries for an order line (polling fallback)."""
|
|
from app.services.render_log import get_entries, count
|
|
entries = get_entries(order_line_id, after_index=after)
|
|
total = count(order_line_id)
|
|
return {"entries": entries, "total": total, "next_after": total}
|
|
|
|
|
|
@router.get("/render-log/{order_line_id}/stream")
|
|
async def stream_render_log(
|
|
order_line_id: str,
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""SSE stream of render log entries for an order line."""
|
|
import asyncio
|
|
import json
|
|
from fastapi.responses import StreamingResponse
|
|
from app.services.render_log import get_entries, count
|
|
|
|
async def event_generator():
|
|
cursor = 0
|
|
idle_ticks = 0
|
|
max_idle = 120 # stop after 2 minutes of no new entries
|
|
while idle_ticks < max_idle:
|
|
entries = get_entries(order_line_id, after_index=cursor)
|
|
if entries:
|
|
idle_ticks = 0
|
|
for entry in entries:
|
|
yield f"data: {json.dumps(entry)}\n\n"
|
|
cursor += len(entries)
|
|
else:
|
|
idle_ticks += 1
|
|
await asyncio.sleep(1)
|
|
yield f"data: {json.dumps({'level': 'info', 'msg': 'Stream ended (idle timeout)', 't': ''})}\n\n"
|
|
|
|
return StreamingResponse(
|
|
event_generator(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|
|
|
|
|
|
from fastapi import status as http_status
|
|
|
|
|
|
@router.post("/activity/{cad_file_id}/reprocess", status_code=http_status.HTTP_202_ACCEPTED)
|
|
async def reprocess_cad_file(
|
|
cad_file_id: str,
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Re-queue a CAD file for full processing (STEP extraction + thumbnail + glTF)."""
|
|
result = await db.execute(select(CadFile).where(CadFile.id == cad_file_id))
|
|
cad_file = result.scalar_one_or_none()
|
|
if not cad_file:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(404, detail="CAD file not found")
|
|
|
|
cad_file.processing_status = ProcessingStatus.pending
|
|
await db.commit()
|
|
|
|
from app.tasks.step_tasks import process_step_file
|
|
process_step_file.delay(cad_file_id)
|
|
return {"queued": cad_file_id, "task": "process_step_file"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Queue inspection + control
|
|
# ---------------------------------------------------------------------------
|
|
|
|
MONITORED_QUEUES = ["step_processing", "thumbnail_rendering", "ai_validation"]
|
|
|
|
|
|
def _parse_redis_task(raw: str) -> dict | None:
|
|
"""Parse a raw Redis Celery message into a simplified dict."""
|
|
import json, base64
|
|
try:
|
|
msg = json.loads(raw)
|
|
headers = msg.get("headers", {})
|
|
task_name = headers.get("task", "unknown")
|
|
task_id = headers.get("id", "unknown")
|
|
argsrepr = headers.get("argsrepr", "")
|
|
args: list = []
|
|
try:
|
|
body = json.loads(base64.b64decode(msg.get("body", "")))
|
|
if isinstance(body, list) and body:
|
|
args = list(body[0])
|
|
except Exception:
|
|
pass
|
|
return {
|
|
"task_id": task_id,
|
|
"task_name": task_name,
|
|
"args": args,
|
|
"argsrepr": argsrepr,
|
|
"status": "pending",
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
@router.get("/queue")
|
|
async def get_queue_status(user: User = Depends(get_current_user)):
|
|
"""Return Celery queue depths, pending tasks, and active/reserved tasks."""
|
|
import asyncio
|
|
import redis as redis_lib
|
|
from app.config import settings as app_settings
|
|
from app.tasks.celery_app import celery_app
|
|
|
|
r = redis_lib.from_url(app_settings.redis_url, decode_responses=True)
|
|
|
|
# Pending tasks per queue from Redis
|
|
queue_depths: dict[str, int] = {}
|
|
pending: list[dict] = []
|
|
for q in MONITORED_QUEUES:
|
|
depth = r.llen(q) or 0
|
|
queue_depths[q] = depth
|
|
if depth > 0:
|
|
raw_items = r.lrange(q, 0, 99)
|
|
for raw in raw_items:
|
|
task = _parse_redis_task(raw)
|
|
if task:
|
|
task["queue"] = q
|
|
pending.append(task)
|
|
|
|
# Active / reserved from Celery inspect (runs in thread, 1.5 s timeout)
|
|
active: list[dict] = []
|
|
reserved: list[dict] = []
|
|
|
|
def _inspect() -> tuple[dict, dict]:
|
|
try:
|
|
insp = celery_app.control.inspect(timeout=1.5)
|
|
return (insp.active() or {}), (insp.reserved() or {})
|
|
except Exception:
|
|
return {}, {}
|
|
|
|
act_raw, rsv_raw = await asyncio.to_thread(_inspect)
|
|
|
|
for worker, tasks in act_raw.items():
|
|
for t in (tasks or []):
|
|
active.append({
|
|
"task_id": t.get("id", ""),
|
|
"task_name": t.get("name", ""),
|
|
"args": list(t.get("args") or []),
|
|
"argsrepr": t.get("kwargs", {}).get("argsrepr", ""),
|
|
"status": "active",
|
|
"worker": worker,
|
|
})
|
|
|
|
for worker, tasks in rsv_raw.items():
|
|
for t in (tasks or []):
|
|
reserved.append({
|
|
"task_id": t.get("id", ""),
|
|
"task_name": t.get("name", ""),
|
|
"args": list(t.get("args") or []),
|
|
"argsrepr": "",
|
|
"status": "reserved",
|
|
"worker": worker,
|
|
})
|
|
|
|
return {
|
|
"queue_depths": queue_depths,
|
|
"pending_count": sum(queue_depths.values()),
|
|
"active": active,
|
|
"reserved": reserved,
|
|
"pending": pending,
|
|
}
|
|
|
|
|
|
@router.post("/queue/purge", status_code=http_status.HTTP_202_ACCEPTED)
|
|
async def purge_queue(user: User = Depends(require_admin_or_pm)):
|
|
"""Delete all pending tasks from all monitored queues."""
|
|
import redis as redis_lib
|
|
from app.config import settings as app_settings
|
|
|
|
r = redis_lib.from_url(app_settings.redis_url, decode_responses=True)
|
|
total = 0
|
|
for q in MONITORED_QUEUES:
|
|
count = r.llen(q) or 0
|
|
if count:
|
|
r.delete(q)
|
|
total += count
|
|
return {"purged": total, "message": f"Removed {total} pending task(s) from queue"}
|
|
|
|
|
|
@router.post("/queue/cancel/{task_id}", status_code=http_status.HTTP_202_ACCEPTED)
|
|
async def cancel_task(task_id: str, user: User = Depends(require_admin_or_pm)):
|
|
"""Revoke a task by ID. Terminates it if running, skips it if still pending."""
|
|
from app.tasks.celery_app import celery_app
|
|
celery_app.control.revoke(task_id, terminate=True, signal="SIGTERM")
|
|
return {"revoked": task_id}
|