From 7c4197000f0fede022478fc32ef921cd44cd7ec8 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Mon, 27 Apr 2026 19:04:01 +0300 Subject: [PATCH] add monitor scrappers --- backend/app/api/v1/admin_scrape.py | 90 +++++++++ frontend/src/app/admin/scrape/page.tsx | 267 +++++++++++++++++++++++++ frontend/tsconfig.tsbuildinfo | 2 +- 3 files changed, 358 insertions(+), 1 deletion(-) diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 8d3b6c9b..3badc6e4 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -67,6 +67,96 @@ def trigger_kn_sweep( } +@router.get("/queue") +def queue_status( + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Snapshot of Celery worker(s) state: active tasks, reserved (queued on + worker, not yet started), scheduled (with ETA), beat-schedule, and queue + depth in Redis. If no worker is running, all worker-side fields are empty. + """ + _check_token(x_admin_token) + + from app.workers.celery_app import celery_app + + inspect = celery_app.control.inspect(timeout=2.0) + + def _flatten(per_worker: dict[str, list[dict[str, Any]]] | None) -> list[dict[str, Any]]: + if not per_worker: + return [] + out: list[dict[str, Any]] = [] + for worker, tasks in per_worker.items(): + for t in tasks or []: + out.append( + { + "worker": worker, + "task_id": t.get("id"), + "name": t.get("name"), + "args": t.get("args"), + "kwargs": t.get("kwargs"), + "time_start": t.get("time_start"), + "eta": t.get("eta"), + } + ) + return out + + try: + active = _flatten(inspect.active()) + except Exception: + active = [] + try: + reserved = _flatten(inspect.reserved()) + except Exception: + reserved = [] + try: + scheduled = _flatten(inspect.scheduled()) + except Exception: + scheduled = [] + + # Pending in broker queue (not yet picked up by any worker). + queue_depth: int | None = None + try: + with celery_app.connection_or_acquire() as conn: + with conn.channel() as channel: + # Default queue name is 'celery'. + queue_depth = channel.client.llen("celery") + except Exception: + queue_depth = None + + workers: list[str] = [] + try: + ping_resp = inspect.ping() or {} + workers = list(ping_resp.keys()) + except Exception: + pass + + return { + "workers": workers, + "queue_depth": queue_depth, + "active": active, + "reserved": reserved, + "scheduled": scheduled, + } + + +class RevokeRequest(BaseModel): + task_id: str = Field(..., min_length=1, max_length=255) + terminate: bool = False # SIGTERM running task vs just unmark from queue + + +@router.post("/revoke") +def revoke_task( + payload: RevokeRequest, + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Cancel a queued or running task by id.""" + _check_token(x_admin_token) + from app.workers.celery_app import celery_app + + celery_app.control.revoke(payload.task_id, terminate=payload.terminate, signal="SIGTERM") + return {"task_id": payload.task_id, "revoked": True, "terminate": payload.terminate} + + @router.get("/failures") def list_failures( db: Annotated[Session, Depends(get_db)], diff --git a/frontend/src/app/admin/scrape/page.tsx b/frontend/src/app/admin/scrape/page.tsx index 6b47c2e0..c344f87a 100644 --- a/frontend/src/app/admin/scrape/page.tsx +++ b/frontend/src/app/admin/scrape/page.tsx @@ -25,6 +25,24 @@ interface TriggerResp { developers: string[] | null; } +interface QueueTask { + worker: string; + task_id: string; + name: string | null; + args: unknown[] | null; + kwargs: Record | null; + time_start: number | null; + eta: string | null; +} + +interface QueueStatus { + workers: string[]; + queue_depth: number | null; + active: QueueTask[]; + reserved: QueueTask[]; + scheduled: QueueTask[]; +} + interface FailureRow { failure_id: number; run_id: number | null; @@ -89,6 +107,27 @@ export default function ScrapeAdminPage() { }, }); + const queue = useQuery({ + queryKey: ["scrape-queue", token], + queryFn: () => + apiFetch("/api/v1/admin/scrape/queue", { + headers: { "X-Admin-Token": token }, + }), + enabled: !!token, + refetchInterval: 3000, + }); + + const revokeMutation = useMutation({ + mutationFn: (task_id: string) => + apiFetch<{ revoked: boolean }>("/api/v1/admin/scrape/revoke", { + method: "POST", + headers: { "X-Admin-Token": token }, + body: JSON.stringify({ task_id, terminate: false }), + }), + onSuccess: () => + queryClient.invalidateQueries({ queryKey: ["scrape-queue"] }), + }); + const failures = useQuery({ queryKey: ["scrape-failures", failuresRunId, token], queryFn: () => { @@ -269,6 +308,219 @@ export default function ScrapeAdminPage() { ) : null} +
+
+

+ Очередь и активные{" "} + {queue.data ? ( + + · workers: {queue.data.workers.length || "0"} · в очереди:{" "} + {queue.data.queue_depth ?? "?"} · активно:{" "} + {queue.data.active.length} · reserved:{" "} + {queue.data.reserved.length} + + ) : null} +

+
+ {!token ? ( +

+ Введите Admin Token чтобы видеть состояние очереди. +

+ ) : queue.isLoading ? ( +

Загрузка…

+ ) : queue.isError ? ( +

Ошибка: {String(queue.error)}

+ ) : queue.data && + queue.data.workers.length === 0 && + (queue.data.queue_depth ?? 0) === 0 ? ( +

+ Нет активных worker-ов и очередь пуста. Запусти{" "} + + docker compose up -d worker beat + {" "} + если ожидаешь scheduled-прогон. +

+ ) : queue.data ? ( + <> + {queue.data.active.length > 0 ? ( + <> +

+ ▶ Активные ({queue.data.active.length}) +

+ + + + {["Task", "Worker", "Аргументы", "Старт", "Действие"].map( + (h) => ( + + ), + )} + + + + {queue.data.active.map((t) => ( + + + + + + + + ))} + +
+ {h} +
+
+ {t.name?.split(".").pop() ?? "—"} +
+
+ {t.task_id?.slice(0, 8)}… +
+ 		{t.worker} + {JSON.stringify(t.args)}{" "} + {JSON.stringify(t.kwargs).slice(0, 80)} + + {t.time_start + ? new Date(t.time_start * 1000).toLocaleTimeString( + "ru", + ) + : "—"} + + +
+ + ) : null} + {queue.data.reserved.length > 0 ? ( + <> +

+ ⏸ Reserved (приняты worker-ом, не стартовали) ( + {queue.data.reserved.length}) +

+ + + + {["Task", "Worker", "Аргументы", "Действие"].map((h) => ( + + ))} + + + + {queue.data.reserved.map((t) => ( + + + + + + + ))} + +
+ {h} +
+
+ {t.name?.split(".").pop() ?? "—"} +
+
+ {t.task_id?.slice(0, 8)}… +
+ 		{t.worker} + {JSON.stringify(t.args)}{" "} + {JSON.stringify(t.kwargs).slice(0, 80)} + + +
+ + ) : null} + {queue.data.scheduled.length > 0 ? ( + <> +

+ 📅 Scheduled (с ETA) ({queue.data.scheduled.length}) +

+
    + {queue.data.scheduled.map((t) => ( +
  • + {t.name} → ETA {t.eta} — {JSON.stringify(t.args)} +
    • + ))} +
+ + ) : null} + + ) : null} +
+