From de0a28ad6241299c3bb160d680bdd53ddcb38eba Mon Sep 17 00:00:00 2001 From: dv Date: Thu, 28 May 2026 17:29:51 +0300 Subject: [PATCH] =?UTF-8?q?=D1=80=D0=B5=D0=B2=D0=BE=D1=80=D0=BA=20=D0=B7?= =?UTF-8?q?=D0=B0=D0=B4=D0=B0=D1=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/services/task.py | 57 +++++++++++++++++++++++++++++++ frontend/src/hooks/useSSE.ts | 2 ++ frontend/src/pages/TaskDetail.tsx | 35 +++++++++++++++---- 3 files changed, 88 insertions(+), 6 deletions(-) diff --git a/backend/app/services/task.py b/backend/app/services/task.py index c1958cc..4ed8549 100644 --- a/backend/app/services/task.py +++ b/backend/app/services/task.py @@ -28,6 +28,23 @@ from app.services.sse import SSEEvent, sse_manager _TASK_DEVICE_CONCURRENCY = settings.TASK_DEVICE_CONCURRENCY +async def _quick_ping(ip: str, attempts: int = 4) -> bool: + """Quick reachability check: ping -c 1 -W 1, up to `attempts` times.""" + for _ in range(attempts): + try: + proc = await asyncio.create_subprocess_exec( + "ping", "-c", "1", "-W", "1", ip, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await asyncio.wait_for(proc.wait(), timeout=3) + if proc.returncode == 0: + return True + except (asyncio.TimeoutError, OSError): + continue + return False + + async def _emit(task_id: str, message: str, level: str = "info") -> None: """Write a progress line to DB and push to SSE subscribers.""" now = datetime.now(timezone.utc) @@ -126,6 +143,41 @@ async def _run_on_device( task_row = await db.get(Task, task_id) if task_row and task_row.cancel_requested: return None # None = skipped due to cancellation + + # Pre-ping reachability check (skip for ping action itself) + if action.name != "ping": + reachable = await _quick_ping(device.ip) + if not reachable: + await _emit(task_id, f"{device.ip}: unreachable, skipping", "warn") + async with AsyncSessionLocal() as db: + async with db.begin(): + tr = TaskResult( + task_id=task_id, + device_id=device.id, + status="skipped", + executed_at=datetime.now(timezone.utc), + duration_ms=0, + stderr="Device unreachable (ping pre-check failed)", + ) + db.add(tr) + await sse_manager.publish(SSEEvent( + type="task_result", + task_id=task_id, + data={ + "device_id": device.id, + "device_ip": device.ip, + "device_hostname": device.hostname, + "status": "skipped", + "duration_ms": 0, + "stdout": None, + "stderr": "Device unreachable (ping pre-check failed)", + "error": None, + }, + )) + return None + + async with AsyncSessionLocal() as db: + async with db.begin(): tr = TaskResult( task_id=task_id, device_id=device.id, @@ -284,6 +336,11 @@ async def _run_task(task_id: str) -> None: return await _emit(task_id, f"Running on {len(targets)} device(s)...", "info") + await sse_manager.publish(SSEEvent( + type="task_meta", + task_id=task_id, + data={"total": len(targets)}, + )) sem = asyncio.Semaphore(_TASK_DEVICE_CONCURRENCY) # Start heartbeat writer diff --git a/frontend/src/hooks/useSSE.ts b/frontend/src/hooks/useSSE.ts index c246ea5..c51ca2b 100644 --- a/frontend/src/hooks/useSSE.ts +++ b/frontend/src/hooks/useSSE.ts @@ -15,6 +15,7 @@ interface SSEHandlers { error: string | null }) => void onTaskStatus?: (data: { status: string; total: number; ok: number; failed: number }) => void + onTaskMeta?: (data: { total: number }) => void } export function useSSE(taskId: string | null | undefined, handlers: SSEHandlers) { @@ -35,6 +36,7 @@ export function useSSE(taskId: string | null | undefined, handlers: SSEHandlers) if (ev.event === 'task_event') handlersRef.current.onTaskEvent?.(data) else if (ev.event === 'task_result') handlersRef.current.onTaskResult?.(data) else if (ev.event === 'task_status') handlersRef.current.onTaskStatus?.(data) + else if (ev.event === 'task_meta') handlersRef.current.onTaskMeta?.(data) } catch { // ignore parse errors } diff --git a/frontend/src/pages/TaskDetail.tsx b/frontend/src/pages/TaskDetail.tsx index b13e540..8862a32 100644 --- a/frontend/src/pages/TaskDetail.tsx +++ b/frontend/src/pages/TaskDetail.tsx @@ -16,6 +16,7 @@ import { Descriptions, Modal, Popconfirm, + Progress, Row, Space, Spin, @@ -83,7 +84,7 @@ export function TaskDetailPage() { const [log, setLog] = useState([]) const [liveResults, setLiveResults] = useState>({}) - const logEndRef = useRef(null) + const [totalDevices, setTotalDevices] = useState(null) const [stdoutModal, setStdoutModal] = useState(null) const seenLogKeys = useRef(new Set()) @@ -118,12 +119,9 @@ export function TaskDetailPage() { device_hostname: data.device_hostname ?? null, }, })), + onTaskMeta: (data) => setTotalDevices(data.total), }) - useEffect(() => { - logEndRef.current?.scrollIntoView({ behavior: 'smooth' }) - }, [log]) - // Merge DB results with live SSE updates const mergedResults = useMemo<(TaskResult & Partial)[]>(() => { if (!task?.device_results) return [] @@ -210,6 +208,7 @@ export function TaskDetailPage() { const s = r.status if (s === 'success') return } color="success">Успешно if (s === 'failed') return } color="error">Ошибка + if (s === 'skipped') return } color="warning">Пропущено return } color="processing">Ожидание }, }, @@ -348,6 +347,30 @@ export function TaskDetailPage() { + {totalDevices != null && totalDevices > 0 && ( + + + {(() => { + const completed = mergedResults.filter((r) => r.status !== 'running').length + const failed = mergedResults.filter((r) => r.status === 'failed').length + const percent = Math.round((completed / totalDevices) * 100) + const isDone = !isActive + const progressStatus = isDone + ? (failed > 0 ? 'exception' : 'success') + : 'active' + return ( + `${completed} / ${totalDevices}`} + strokeColor={failed > 0 ? undefined : '#52c41a'} + /> + ) + })()} + + + )} + )) )} -
+