diff --git a/app/api/queue_overview.py b/app/api/queue_overview.py new file mode 100644 index 0000000..a1ea3be --- /dev/null +++ b/app/api/queue_overview.py @@ -0,0 +1,79 @@ +"""처리 머신 보드 API — GET /api/queue/overview (plan ds-processing-ui-6an). + +홈 stage 평면 테이블을 "머신 관점 보드(누가 일하나)"로 — 집계 로직은 +services/queue_overview.py (순수 판정부 분리). 응답 스키마는 FE 와 계약 고정. +응답에 raw 모델명 노출 금지 — 머신 label 만. +""" + +from typing import Annotated, Literal + +from fastapi import APIRouter, Depends +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from core.auth import get_current_user +from core.database import get_session +from models.user import User +from services.queue_overview import build_overview + +router = APIRouter() + + +class CurrentItem(BaseModel): + """머신이 지금 처리 중인 문서 (최대 2건).""" + document_id: int + title: str + stage: str + + +class MachineCard(BaseModel): + """머신 카드 — stage 귀속 합산 + 완료 실적(summarize 는 풀 분리) + state.""" + key: Literal["gpu", "macmini", "macbook"] + label: str + state: Literal["active", "deferred", "idle"] + stages: list[str] + pending: int + processing: int + failed: int + done_1h: int + done_today: int + deferred_pending: int + current: list[CurrentItem] + + +class SummarizeEta(BaseModel): + """summarize 풀 ETA — done > inflow 일 때만 eta_minutes 산출.""" + pending: int + done_rate_1h: int + inflow_rate_1h: int + eta_minutes: int | None + + +class TrendBucket(BaseModel): + """summarize 24h 추이 버킷 — hour 는 KST "HH:00" 라벨.""" + hour: str + inflow: int + done: int + + +class Totals(BaseModel): + """전 stage 합계.""" + pending: int + processing: int + failed: int + + +class QueueOverviewResponse(BaseModel): + machines: list[MachineCard] + summarize_eta: SummarizeEta + trend_24h: list[TrendBucket] + totals: Totals + + +@router.get("/overview", response_model=QueueOverviewResponse) +async def get_queue_overview( + user: Annotated[User, Depends(get_current_user)], + session: Annotated[AsyncSession, Depends(get_session)], +): + """머신 관점 처리 보드 + summarize ETA 집계 (라이브 계산, 신규 테이블 0)""" + return QueueOverviewResponse.model_validate(await build_overview(session)) diff --git a/app/main.py b/app/main.py index da3516e..66ef288 100644 --- a/app/main.py +++ b/app/main.py @@ -22,6 +22,7 @@ from api.events import router as events_router from api.library import router as library_router from api.memos import router as memos_router from api.news import router as news_router +from api.queue_overview import router as queue_overview_router from api.search import router as search_router from api.setup import router as setup_router from api.study_question_progress import router as study_question_progress_router @@ -183,6 +184,8 @@ app.include_router(events_router, prefix="/api/events", tags=["events"]) app.include_router(dashboard_router, prefix="/api/dashboard", tags=["dashboard"]) app.include_router(library_router, prefix="/api/library", tags=["library"]) app.include_router(news_router, prefix="/api/news", tags=["news"]) +# 처리 머신 보드 (plan ds-processing-ui-6an) — GET /api/queue/overview +app.include_router(queue_overview_router, prefix="/api/queue", tags=["queue"]) app.include_router(digest_router, prefix="/api/digest", tags=["digest"]) app.include_router(briefing_router, prefix="/api/briefing", tags=["briefing"]) app.include_router(audio_router, prefix="/api/audio", tags=["audio"]) diff --git a/app/services/queue_overview.py b/app/services/queue_overview.py new file mode 100644 index 0000000..63779b6 --- /dev/null +++ b/app/services/queue_overview.py @@ -0,0 +1,375 @@ +"""처리 머신 보드 + ETA 집계 (plan ds-processing-ui-6an, 안2+안5/6). + +GET /api/queue/overview 의 집계 로직. 모든 수치는 기존 processing_queue / +documents 컬럼에서 라이브 계산 — 신규 테이블/마이그레이션 0 (HARD 제약). + +구조: SQL 수집부(build_overview 내부 5쿼리)와 판정부(순수 함수)를 분리. +판정부(rows_to_* / build_machines / build_summarize_eta / build_trend / +build_totals / compute_eta_minutes)는 DB 없이 단위테스트 가능. + +귀속 규칙 (단일 진실): +- stage→machine 정적 맵: gpu = extract/embed/chunk/markdown/preview/thumbnail/ + fulltext/stt · macmini = classify/summarize · macbook = deep_summary + (단, settings.ai.deep 부재 시 deep_summary 도 macmini 귀속). +- summarize 는 풀(pool): pending/processing/failed 는 macmini 귀속이되, 완료 + 실적(done_*)은 documents.ai_model_version 조인으로 분리 — 'qwen-macbook' + 이면 macbook 실적, 아니면 macmini 실적. +- deferred_pending(payload.deferred_until 미래)은 macbook 카드 귀속 + (보류 = 맥북 불가 신호). +""" + +from datetime import datetime, timedelta +from posixpath import basename +from zoneinfo import ZoneInfo + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from core.config import settings + +KST = ZoneInfo("Asia/Seoul") + +# 내부 판별용 alias — 응답에 raw 모델명 노출 금지, 머신 label 만 노출. +_MACBOOK_MODEL_ALIAS = "qwen-macbook" + +# stage→machine 정적 맵 재료 (선언 순서 = 카드 stages 표시 순서) +_GPU_STAGES = ( + "extract", "embed", "chunk", "markdown", + "preview", "thumbnail", "fulltext", "stt", +) +_MACMINI_STAGES = ("classify", "summarize") +_MACBOOK_STAGES = ("deep_summary",) +_STAGE_ORDER = _GPU_STAGES + _MACMINI_STAGES + _MACBOOK_STAGES + +_MACHINE_KEYS = ("gpu", "macmini", "macbook") +_MACHINE_LABELS = { + "gpu": "GPU 서버", + "macmini": "맥미니", + "macbook": "맥북 M5 Max", +} + +# 머신 카드당 current 표시 상한 +_CURRENT_LIMIT = 2 + + +def stage_machine_map(deep_enabled: bool) -> dict[str, str]: + """stage → machine key 맵. deep 슬롯 부재 시 deep_summary 는 macmini 귀속.""" + mapping: dict[str, str] = {} + for s in _GPU_STAGES: + mapping[s] = "gpu" + for s in _MACMINI_STAGES: + mapping[s] = "macmini" + for s in _MACBOOK_STAGES: + mapping[s] = "macbook" if deep_enabled else "macmini" + return mapping + + +def _zero_stage() -> dict: + return { + "pending": 0, "processing": 0, "failed": 0, + "done_1h": 0, "done_today": 0, "done_15m": 0, + "deferred_pending": 0, "created_1h": 0, + } + + +def rows_to_stage_stats(rows) -> dict[str, dict]: + """stage×status 집계 쿼리 행 → {stage: {pending, ..., created_1h}} 변환.""" + stats: dict[str, dict] = {} + for row in rows: + stats[row[0]] = { + "pending": int(row[1] or 0), + "processing": int(row[2] or 0), + "failed": int(row[3] or 0), + "done_1h": int(row[4] or 0), + "done_today": int(row[5] or 0), + "done_15m": int(row[6] or 0), + "deferred_pending": int(row[7] or 0), + "created_1h": int(row[8] or 0), + } + return stats + + +def rows_to_summarize_split(rows) -> dict[str, dict]: + """summarize 완료 실적 분리 쿼리 행 → {"macbook"|"macmini": {done_*}}. + + is_macbook = documents.ai_model_version 이 'qwen-macbook' 인지 (내부 판별 전용). + """ + split = { + "macbook": {"done_1h": 0, "done_today": 0, "done_15m": 0}, + "macmini": {"done_1h": 0, "done_today": 0, "done_15m": 0}, + } + for row in rows: + key = "macbook" if row[0] else "macmini" + split[key]["done_1h"] += int(row[1] or 0) + split[key]["done_today"] += int(row[2] or 0) + split[key]["done_15m"] += int(row[3] or 0) + return split + + +def display_title(row: dict) -> str: + """표시용 제목 — title > original_filename > file_path basename > 문서 id.""" + if row.get("title"): + return row["title"] + if row.get("original_filename"): + return row["original_filename"] + if row.get("file_path"): + return basename(row["file_path"].rstrip("/")) + return f"문서 #{row['document_id']}" + + +def build_machines( + stage_stats: dict[str, dict], + summarize_split: dict[str, dict], + current_rows: list[dict], + *, + deep_enabled: bool, +) -> list[dict]: + """머신 카드 3장 (gpu / macmini / macbook) 구성 — 귀속 규칙의 판정부.""" + smap = stage_machine_map(deep_enabled) + + def g(stage: str, field: str) -> int: + return stage_stats.get(stage, {}).get(field, 0) + + # current 귀속: processing 행을 머신별 최대 2건 (summarize processing → macmini) + current_by_machine: dict[str, list[dict]] = {k: [] for k in _MACHINE_KEYS} + for row in current_rows: + machine = smap.get(row["stage"]) + if machine and len(current_by_machine[machine]) < _CURRENT_LIMIT: + current_by_machine[machine].append({ + "document_id": row["document_id"], + "title": display_title(row), + "stage": row["stage"], + }) + + machines = [] + for key in _MACHINE_KEYS: + stages = [s for s in _STAGE_ORDER if smap[s] == key] + + pending = sum(g(s, "pending") for s in stages) + processing = sum(g(s, "processing") for s in stages) + failed = sum(g(s, "failed") for s in stages) + + # 완료 실적: summarize 는 풀이라 stage 합산에서 제외하고 split 로 귀속 + done_1h = sum(g(s, "done_1h") for s in stages if s != "summarize") + done_today = sum(g(s, "done_today") for s in stages if s != "summarize") + done_15m = sum(g(s, "done_15m") for s in stages if s != "summarize") + if key in summarize_split: + done_1h += summarize_split[key]["done_1h"] + done_today += summarize_split[key]["done_today"] + done_15m += summarize_split[key]["done_15m"] + + # 보류 백오프 = 맥북 불가 신호 → macbook 카드 귀속 (deep 슬롯 유무 무관) + deferred_pending = ( + g("summarize", "deferred_pending") + g("deep_summary", "deferred_pending") + if key == "macbook" else 0 + ) + + # state 판정 — macbook 의 done_15m 은 위에서 deep_summary stage 분 + summarize 풀 + # split 분이 이미 합산돼 있고, processing 은 deep_summary 진행 중(=맥북 생성 중)을 + # 포함하므로 동일 분기로 충분 (검수 보정 2026-06-11: deep 처리 중에도 active). + if key == "macbook" and deferred_pending > 0: + state = "deferred" + else: + state = "active" if (processing > 0 or done_15m > 0) else "idle" + + machines.append({ + "key": key, + "label": _MACHINE_LABELS[key], + "state": state, + "stages": stages, + "pending": pending, + "processing": processing, + "failed": failed, + "done_1h": done_1h, + "done_today": done_today, + "deferred_pending": deferred_pending, + "current": current_by_machine[key], + }) + return machines + + +def compute_eta_minutes(pending: int, done_1h: int, inflow_1h: int) -> int | None: + """ETA(분) = 순소화율 기반. done > inflow 일 때만 산출, 아니면 None (소화 불가).""" + if done_1h > inflow_1h: + return round(pending / (done_1h - inflow_1h) * 60) + return None + + +def build_summarize_eta(stage_stats: dict[str, dict]) -> dict: + """summarize 풀 ETA — pending 은 보류(deferred) 포함 총수.""" + s = stage_stats.get("summarize", _zero_stage()) + pending = s["pending"] + done_rate = s["done_1h"] + inflow_rate = s["created_1h"] + return { + "pending": pending, + "done_rate_1h": done_rate, + "inflow_rate_1h": inflow_rate, + "eta_minutes": compute_eta_minutes(pending, done_rate, inflow_rate), + } + + +def build_trend( + inflow_buckets: dict[str, int], + done_buckets: dict[str, int], + now_kst: datetime, +) -> list[dict]: + """summarize 24h 추이 — KST 시간 버킷 24개 (오래된 것부터, 빈 버킷 0). + + 버킷 key = "YYYY-MM-DD HH:00" (KST). SQL to_char 출력과 동일 포맷. + """ + base = now_kst.replace(minute=0, second=0, microsecond=0) + trend = [] + for i in range(23, -1, -1): + bucket = base - timedelta(hours=i) + key = bucket.strftime("%Y-%m-%d %H:00") + trend.append({ + "hour": bucket.strftime("%H:00"), + "inflow": inflow_buckets.get(key, 0), + "done": done_buckets.get(key, 0), + }) + return trend + + +def build_totals(stage_stats: dict[str, dict]) -> dict: + """전 stage 합계.""" + return { + "pending": sum(s["pending"] for s in stage_stats.values()), + "processing": sum(s["processing"] for s in stage_stats.values()), + "failed": sum(s["failed"] for s in stage_stats.values()), + } + + +def compose_overview( + stage_stats: dict[str, dict], + summarize_split: dict[str, dict], + inflow_buckets: dict[str, int], + done_buckets: dict[str, int], + current_rows: list[dict], + *, + deep_enabled: bool, + now_kst: datetime, +) -> dict: + """수집된 통계 → 응답 dict (계약 shape). 순수 함수 — DB 불요.""" + return { + "machines": build_machines( + stage_stats, summarize_split, current_rows, deep_enabled=deep_enabled + ), + "summarize_eta": build_summarize_eta(stage_stats), + "trend_24h": build_trend(inflow_buckets, done_buckets, now_kst), + "totals": build_totals(stage_stats), + } + + +# ─── SQL 수집부 (총 5쿼리) ──────────────────────────────────────────────────── + +# 1) stage×status 집계 + 시간창 완료/유입 + 보류 (1방) +_STAGE_STATS_SQL = """ + SELECT + stage, + COUNT(*) FILTER (WHERE status = 'pending') AS pending, + COUNT(*) FILTER (WHERE status = 'processing') AS processing, + COUNT(*) FILTER (WHERE status = 'failed') AS failed, + COUNT(*) FILTER (WHERE status = 'completed' + AND completed_at > NOW() - INTERVAL '1 hour') AS done_1h, + COUNT(*) FILTER (WHERE status = 'completed' + AND completed_at > :kst_midnight) AS done_today, + COUNT(*) FILTER (WHERE status = 'completed' + AND completed_at > NOW() - INTERVAL '15 minutes') AS done_15m, + COUNT(*) FILTER (WHERE status = 'pending' + AND payload ->> 'deferred_until' IS NOT NULL + AND (payload ->> 'deferred_until')::timestamptz > NOW()) + AS deferred_pending, + COUNT(*) FILTER (WHERE created_at > NOW() - INTERVAL '1 hour') AS created_1h + FROM processing_queue + GROUP BY stage +""" + +# 2) summarize 풀 완료 실적 분리 (documents.ai_model_version 조인, 1방) +# 스캔 하한 = 오늘 0시(KST)와 1h 전 중 더 이른 시각 (자정 직후 1h 창 보전). +_SUMMARIZE_SPLIT_SQL = """ + SELECT + COALESCE(d.ai_model_version = :macbook_alias, false) AS is_macbook, + COUNT(*) FILTER (WHERE q.completed_at > NOW() - INTERVAL '1 hour') AS done_1h, + COUNT(*) FILTER (WHERE q.completed_at > :kst_midnight) AS done_today, + COUNT(*) FILTER (WHERE q.completed_at > NOW() - INTERVAL '15 minutes') AS done_15m + FROM processing_queue q + JOIN documents d ON d.id = q.document_id + WHERE q.stage = 'summarize' + AND q.status = 'completed' + AND q.completed_at > LEAST(:kst_midnight, NOW() - INTERVAL '1 hour') + GROUP BY 1 +""" + +# 3/4) summarize 24h 추이 — KST 시간 버킷 (inflow/done 각 1방) +_TREND_INFLOW_SQL = """ + SELECT to_char(date_trunc('hour', created_at AT TIME ZONE 'Asia/Seoul'), + 'YYYY-MM-DD HH24:00') AS bucket, + COUNT(*) AS n + FROM processing_queue + WHERE stage = 'summarize' + AND created_at > NOW() - INTERVAL '24 hours' + GROUP BY 1 +""" + +_TREND_DONE_SQL = """ + SELECT to_char(date_trunc('hour', completed_at AT TIME ZONE 'Asia/Seoul'), + 'YYYY-MM-DD HH24:00') AS bucket, + COUNT(*) AS n + FROM processing_queue + WHERE stage = 'summarize' + AND status = 'completed' + AND completed_at > NOW() - INTERVAL '24 hours' + GROUP BY 1 +""" + +# 5) processing 행 + 표시용 제목 재료 (1방 — 머신별 2건 슬라이스는 판정부에서) +_CURRENT_SQL = """ + SELECT q.stage, q.document_id, d.title, d.original_filename, d.file_path + FROM processing_queue q + JOIN documents d ON d.id = q.document_id + WHERE q.status = 'processing' + ORDER BY q.started_at DESC NULLS LAST + LIMIT 50 +""" + + +async def build_overview(session: AsyncSession) -> dict: + """5쿼리 수집 → compose_overview 판정 → 응답 dict.""" + now_kst = datetime.now(KST) + kst_midnight = now_kst.replace(hour=0, minute=0, second=0, microsecond=0) + deep_enabled = settings.ai is not None and settings.ai.deep is not None + + stage_rows = ( + await session.execute(text(_STAGE_STATS_SQL), {"kst_midnight": kst_midnight}) + ).all() + split_rows = ( + await session.execute( + text(_SUMMARIZE_SPLIT_SQL), + {"kst_midnight": kst_midnight, "macbook_alias": _MACBOOK_MODEL_ALIAS}, + ) + ).all() + inflow_rows = (await session.execute(text(_TREND_INFLOW_SQL))).all() + done_rows = (await session.execute(text(_TREND_DONE_SQL))).all() + current_result = (await session.execute(text(_CURRENT_SQL))).all() + + current_rows = [ + { + "stage": row[0], + "document_id": row[1], + "title": row[2], + "original_filename": row[3], + "file_path": row[4], + } + for row in current_result + ] + + return compose_overview( + rows_to_stage_stats(stage_rows), + rows_to_summarize_split(split_rows), + {row[0]: int(row[1]) for row in inflow_rows}, + {row[0]: int(row[1]) for row in done_rows}, + current_rows, + deep_enabled=deep_enabled, + now_kst=now_kst, + ) diff --git a/frontend/src/lib/components/QueueDrawer.svelte b/frontend/src/lib/components/QueueDrawer.svelte new file mode 100644 index 0000000..e738088 --- /dev/null +++ b/frontend/src/lib/components/QueueDrawer.svelte @@ -0,0 +1,106 @@ + + + + +{#if open} +
+ + + + + +
+{/if} diff --git a/frontend/src/lib/stores/queueOverview.ts b/frontend/src/lib/stores/queueOverview.ts new file mode 100644 index 0000000..d46d61f --- /dev/null +++ b/frontend/src/lib/stores/queueOverview.ts @@ -0,0 +1,63 @@ +// 처리 큐 overview store — GET /api/queue/overview 를 60초 주기로 폴링. +// system.ts 의 dashboardSummary 와 같은 구독 기반 패턴 (첫 subscribe 시 시작). +// +// 의도적으로 api() 헬퍼를 쓰지 않는다 — 폴링 경로의 401 이 refresh 실패 → +// window.location='/login' 강제 logout 부수효과를 일으키면 안 됨 (eid 리뷰 +// finding 재발 방지). 백엔드 미배포(404)/401/네트워크 실패 전부 silent 하게 +// null 로 수렴하고, 소비자(스트립/보드/드로어)는 null 이면 스스로 숨는다. + +import { writable } from 'svelte/store'; +import { browser } from '$app/environment'; +import { getAccessToken } from '$lib/api'; +import type { QueueOverview } from '$lib/types/queue'; + +const POLL_INTERVAL_MS = 60_000; + +let pollHandle: ReturnType | null = null; +let subscriberCount = 0; +let inFlight: Promise | null = null; + +const internal = writable(null, (_set) => { + subscriberCount += 1; + if (subscriberCount === 1 && browser) { + void refreshQueueOverview(); + pollHandle = setInterval(() => void refreshQueueOverview(), POLL_INTERVAL_MS); + } + return () => { + subscriberCount -= 1; + if (subscriberCount === 0 && pollHandle) { + clearInterval(pollHandle); + pollHandle = null; + } + }; +}); + +export const queueOverview = { subscribe: internal.subscribe }; + +/** 경량 fetch — 실패는 전부 null (silent 비차단, 강제 logout 경로 없음) */ +async function fetchOverview(): Promise { + try { + const headers: Record = {}; + const token = getAccessToken(); + if (token) headers['Authorization'] = `Bearer ${token}`; + const res = await fetch('/api/queue/overview', { headers, credentials: 'include' }); + if (!res.ok) return null; + return (await res.json()) as QueueOverview; + } catch { + return null; + } +} + +/** 수동/추가 폴링용 — 홈은 자체 30s interval 로 이 함수를 호출 (동시 fetch 합치기) */ +export async function refreshQueueOverview(): Promise { + if (!browser) return; + if (inFlight) return inFlight; + inFlight = (async () => { + try { + internal.set(await fetchOverview()); + } finally { + inFlight = null; + } + })(); + return inFlight; +} diff --git a/frontend/src/lib/stores/uiState.svelte.ts b/frontend/src/lib/stores/uiState.svelte.ts index 2cdc75d..c5ce671 100644 --- a/frontend/src/lib/stores/uiState.svelte.ts +++ b/frontend/src/lib/stores/uiState.svelte.ts @@ -3,7 +3,9 @@ // (toast는 별도 store. drawer가 persistent inline panel(예: xl+ meta rail)일 때는 // 여기 시스템 밖이다 — 그저 레이아웃의 일부.) -type Drawer = { id: 'sidebar' | 'meta' } | null; +// 'queue' = 처리 현황 드로어 (상태 스트립 클릭 시 우측) — 단일 slot 규칙 동일 +export type DrawerId = 'sidebar' | 'meta' | 'queue'; +type Drawer = { id: DrawerId } | null; type Modal = { id: string }; class UIState { @@ -11,14 +13,14 @@ class UIState { modalStack = $state([]); // ── Drawer (단일 slot) ────────────────────────────── - openDrawer(id: 'sidebar' | 'meta') { + openDrawer(id: DrawerId) { // 새 drawer 열면 이전 drawer는 자동으로 사라진다 (단일 slot) this.drawer = { id }; } closeDrawer() { this.drawer = null; } - isDrawerOpen(id: 'sidebar' | 'meta') { + isDrawerOpen(id: DrawerId) { return this.drawer?.id === id; } diff --git a/frontend/src/lib/types/queue.ts b/frontend/src/lib/types/queue.ts new file mode 100644 index 0000000..f0b15e7 --- /dev/null +++ b/frontend/src/lib/types/queue.ts @@ -0,0 +1,64 @@ +/** + * GET /api/queue/overview 응답 타입. + * + * Backend 는 병렬 트랙에서 구현 중 — 계약 고정 (feat/ds-processing-board). + * 필드 변경 시 양쪽 동시 수정 필수. + */ + +export type MachineKey = 'gpu' | 'macmini' | 'macbook'; + +/** 머신 상태 — active(가동) / deferred(보류) / idle(대기) */ +export type MachineState = 'active' | 'deferred' | 'idle'; + +/** 머신이 지금 처리 중인 문서 1건 */ +export interface MachineCurrentItem { + document_id: number; + title: string; + stage: string; +} + +export interface MachineOverview { + key: MachineKey; + label: string; + state: MachineState; + /** 담당 단계 키 목록 (extract/classify/... — 홈 STAGE_LABEL 로 한글화) */ + stages: string[]; + pending: number; + processing: number; + failed: number; + /** 최근 1시간 완료 건수 (처리율 N/h 표기) */ + done_1h: number; + done_today: number; + /** 보류 건수 — 맥북 sleep 등으로 자동 재개 대기 중 */ + deferred_pending: number; + current: MachineCurrentItem[]; +} + +/** 요약 백로그 ETA (안5 라이트) — 추정치, 유입 변동 시 오차 */ +export interface SummarizeEta { + pending: number; + done_rate_1h: number; + inflow_rate_1h: number; + /** null = 유입이 소화를 앞섬 (백필 중) — 소진 예상 불가 */ + eta_minutes: number | null; +} + +/** 시간당 유입 vs 소화 (이번 트랙 미렌더 — 후속 추세 위젯 슬롯) */ +export interface TrendPoint { + hour: string; + inflow: number; + done: number; +} + +export interface QueueTotals { + pending: number; + processing: number; + failed: number; +} + +export interface QueueOverview { + machines: MachineOverview[]; + summarize_eta: SummarizeEta; + trend_24h: TrendPoint[]; + totals: QueueTotals; +} diff --git a/frontend/src/lib/utils/queueDisplay.ts b/frontend/src/lib/utils/queueDisplay.ts new file mode 100644 index 0000000..9310164 --- /dev/null +++ b/frontend/src/lib/utils/queueDisplay.ts @@ -0,0 +1,38 @@ +// 처리 머신 보드 / 상태 스트립 / 드로어 공용 표시 헬퍼. +// 상태 표현은 dot + 칩 (이모지 금지 원칙) — 토큰 클래스만 사용. + +import type { MachineState } from '$lib/types/queue'; + +/** 머신 상태 한글 라벨 */ +export const MACHINE_STATE_LABEL: Record = { + active: '가동', + deferred: '보류', + idle: '대기', +}; + +/** 상태 dot 색 — 가동=success / 보류=warning / 대기=faint */ +export function machineDotClass(state: MachineState): string { + if (state === 'active') return 'bg-success'; + if (state === 'deferred') return 'bg-warning'; + return 'bg-faint'; +} + +/** 상태 칩 톤 — 가동=accent / 보류=warn / 대기=dim */ +export function machineChipClass(state: MachineState): string { + if (state === 'active') return 'bg-accent/10 text-accent'; + if (state === 'deferred') return 'bg-warning/10 text-warning'; + return 'bg-surface-hover text-faint'; +} + +/** 처리율 표기 — 정수는 그대로, 소수는 한 자리 */ +export function formatRate(n: number): string { + return Number.isInteger(n) ? n.toLocaleString() : n.toFixed(1); +} + +/** ETA 분 → "약 N분/N시간 후 소진 예상" (추정치 — title 로 명시는 호출부 책임) */ +export function etaPhrase(minutes: number): string { + if (minutes < 60) return `약 ${Math.max(1, Math.round(minutes))}분 후 소진 예상`; + const hours = minutes / 60; + const text = hours >= 10 ? String(Math.round(hours)) : String(Math.round(hours * 10) / 10); + return `약 ${text}시간 후 소진 예상`; +} diff --git a/frontend/src/routes/+layout.svelte b/frontend/src/routes/+layout.svelte index d450efe..aa0ee81 100644 --- a/frontend/src/routes/+layout.svelte +++ b/frontend/src/routes/+layout.svelte @@ -8,8 +8,11 @@ import { toasts, removeToast } from '$lib/stores/toast'; import { refresh as refreshPublicConfig } from '$lib/stores/config'; import { ui } from '$lib/stores/uiState.svelte'; + import { queueOverview } from '$lib/stores/queueOverview'; + import { MACHINE_STATE_LABEL, machineChipClass } from '$lib/utils/queueDisplay'; import Sidebar from '$lib/components/Sidebar.svelte'; import SystemStatusDot from '$lib/components/SystemStatusDot.svelte'; + import QueueDrawer from '$lib/components/QueueDrawer.svelte'; import QuickMemoButton from '$lib/components/QuickMemoButton.svelte'; import IconButton from '$lib/components/ui/IconButton.svelte'; import Drawer from '$lib/components/ui/Drawer.svelte'; @@ -65,6 +68,15 @@ let showChrome = $derived($isAuthenticated && !NO_CHROME_PATHS.some(p => $page.url.pathname.startsWith(p))); let showSidebar = $derived(showChrome && !NO_SIDEBAR_PATHS.some(p => $page.url.pathname.startsWith(p))); + // 처리 현황 스트립 (안6 라이트) — 60s 폴링 store 공유. fetch 실패/401 시 + // store 가 null → 스트립 자체를 숨김 (silent 비차단, 로그인 페이지 동일). + let queue = $derived($queueOverview); + let queueMacbook = $derived(queue?.machines?.find((m) => m.key === 'macbook') ?? null); + function toggleQueueDrawer() { + if (ui.isDrawerOpen('queue')) ui.closeDrawer(); + else ui.openDrawer('queue'); + } + function handleKeydown(e) { if (e.key === '/' && !['INPUT', 'TEXTAREA'].includes(document.activeElement?.tagName)) { e.preventDefault(); @@ -162,6 +174,28 @@ + + {#if queue} + + {/if} +
{#if showSidebar} @@ -191,6 +225,9 @@
+ + + diff --git a/frontend/src/routes/+page.svelte b/frontend/src/routes/+page.svelte index 35a590f..ce57d71 100644 --- a/frontend/src/routes/+page.svelte +++ b/frontend/src/routes/+page.svelte @@ -13,6 +13,11 @@ import { domainBgClass, domainLabel } from '$lib/utils/domainSlug'; import { user } from '$lib/stores/auth'; import { api } from '$lib/api'; + import { queueOverview, refreshQueueOverview } from '$lib/stores/queueOverview'; + import { + MACHINE_STATE_LABEL, machineChipClass, machineDotClass, formatRate, etaPhrase, + } from '$lib/utils/queueDisplay'; + import type { QueueOverview } from '$lib/types/queue'; import EmptyState from '$lib/components/ui/EmptyState.svelte'; import Skeleton from '$lib/components/ui/Skeleton.svelte'; import { @@ -125,6 +130,28 @@ preview: '미리보기', thumbnail: '썸네일', }; + // ─── 처리 머신 보드 (안2) + ETA (안5 라이트) — GET /api/queue/overview ─── + // 홈은 30s 폴링 (store 기본 60s 위에 추가 — inFlight 합치기로 중복 호출 0). + // 백엔드 미배포/실패 시 store=null → 보드 자체가 조용히 생략 (silent 비차단). + let queue = $derived($queueOverview); + + // 머신 담당 단계 라벨 — STAGE_LABEL 재사용 + overview 전용 단계 보강 + // (backend services/queue_overview.py _STAGE_ORDER 와 동기), 미지 키는 raw + const QUEUE_STAGE_LABEL: Record = { + ...STAGE_LABEL, + summarize: '요약', chunk: '청크', markdown: '마크다운', + fulltext: '전문', deep_summary: '심층분석', + }; + function queueStageLabel(stage: string): string { + return QUEUE_STAGE_LABEL[stage] ?? stage; + } + + onMount(() => { + void refreshQueueOverview(); + const handle = setInterval(() => void refreshQueueOverview(), 30_000); + return () => clearInterval(handle); + }); + interface PipelineRow { stage: string; label: string; pending: number; processing: number; failed: number; total: number; @@ -420,7 +447,68 @@ - + + {#if queue} +
+
처리 머신
+
+ {#each queue.machines as m (m.key)} +
+ +
+ + + {m.label} + + + {MACHINE_STATE_LABEL[m.state]} + +
+ + {#if m.stages.length > 0} +
+ {#each m.stages as s (s)} + {queueStageLabel(s)} + {/each} +
+ {/if} + +
+ 대기 {m.pending.toLocaleString()} + · 처리율 {formatRate(m.done_1h)}/h + · 오늘 {m.done_today.toLocaleString()}건 +
+ + {#if m.key === 'macbook' && m.deferred_pending > 0} +
보류 {m.deferred_pending.toLocaleString()}건 — 자동 재개 대기
+ {/if} + + {#if m.current.length > 0} +
`${c.title} (${queueStageLabel(c.stage)})`).join(' · ')}> + 지금: {m.current[0].title} ({queueStageLabel(m.current[0].stage)}){m.current.length > 1 ? ` 외 ${m.current.length - 1}건` : ''} +
+ {/if} +
+ {/each} +
+ + +
+ 요약 대기 {queue.summarize_eta.pending.toLocaleString()}건 + — 소화 {formatRate(queue.summarize_eta.done_rate_1h)}/h + · 유입 {formatRate(queue.summarize_eta.inflow_rate_1h)}/h + {#if queue.summarize_eta.eta_minutes != null} + · {etaPhrase(queue.summarize_eta.eta_minutes)} + {:else} + · 유입 우세(백필 중) + {/if} +
+
+ {/if} + +
- 파이프라인 상세 + 단계 상세 {#if totalFailed > 0}실패 {totalFailed}{/if} diff --git a/tests/test_queue_overview.py b/tests/test_queue_overview.py new file mode 100644 index 0000000..9db1e42 --- /dev/null +++ b/tests/test_queue_overview.py @@ -0,0 +1,345 @@ +"""GET /api/queue/overview 판정부 단위테스트 — DB 불요 (plan ds-processing-ui-6an). + +services/queue_overview 의 SQL 수집부와 분리된 순수 판정 함수 +(stage_machine_map / build_machines / build_summarize_eta / build_trend / +build_totals / compute_eta_minutes / rows_to_* / display_title) 를 +mock 행으로 검증한다. 통합(실 SQL)은 배포 후 라이브 smoke 로 확인. +""" + +from datetime import datetime +from zoneinfo import ZoneInfo + +from services.queue_overview import ( + build_machines, + build_summarize_eta, + build_totals, + build_trend, + compose_overview, + compute_eta_minutes, + display_title, + rows_to_stage_stats, + rows_to_summarize_split, + stage_machine_map, +) + +KST = ZoneInfo("Asia/Seoul") + + +def _stage(**kw) -> dict: + """stage 통계 1건 — 미지정 필드 0.""" + base = { + "pending": 0, "processing": 0, "failed": 0, + "done_1h": 0, "done_today": 0, "done_15m": 0, + "deferred_pending": 0, "created_1h": 0, + } + base.update(kw) + return base + + +def _split(macbook: dict | None = None, macmini: dict | None = None) -> dict: + """summarize 풀 완료 실적 split — 미지정 0.""" + zero = {"done_1h": 0, "done_today": 0, "done_15m": 0} + return { + "macbook": {**zero, **(macbook or {})}, + "macmini": {**zero, **(macmini or {})}, + } + + +def _machine(machines: list[dict], key: str) -> dict: + return next(m for m in machines if m["key"] == key) + + +# ─── stage→machine 귀속 맵 ──────────────────────────────────────────────────── + +def test_stage_machine_map_deep_enabled(): + smap = stage_machine_map(deep_enabled=True) + for s in ("extract", "embed", "chunk", "markdown", "preview", "thumbnail", "fulltext", "stt"): + assert smap[s] == "gpu" + assert smap["classify"] == "macmini" + assert smap["summarize"] == "macmini" + assert smap["deep_summary"] == "macbook" + + +def test_stage_machine_map_deep_disabled(): + """deep 슬롯 부재 시 deep_summary 도 macmini 귀속.""" + smap = stage_machine_map(deep_enabled=False) + assert smap["deep_summary"] == "macmini" + + +# ─── 머신 카드 귀속 합산 ────────────────────────────────────────────────────── + +def test_gpu_stage_counts_attribution(): + stats = { + "extract": _stage(pending=3, processing=1, done_1h=5, done_today=9, done_15m=1), + "stt": _stage(failed=2, done_1h=1, done_today=2), + } + machines = build_machines(stats, _split(), [], deep_enabled=True) + gpu = _machine(machines, "gpu") + assert (gpu["pending"], gpu["processing"], gpu["failed"]) == (3, 1, 2) + assert (gpu["done_1h"], gpu["done_today"]) == (6, 11) + # gpu 의 stages 는 정적 8종 전부 (집계 0 이어도 표시) + assert gpu["stages"] == [ + "extract", "embed", "chunk", "markdown", + "preview", "thumbnail", "fulltext", "stt", + ] + + +def test_summarize_pool_split_attribution(): + """summarize pending/failed = macmini 귀속, 완료 실적은 split 로 분리 — + stage-level summarize done 수치는 카드에 이중 합산되지 않는다.""" + stats = { + "classify": _stage(done_1h=2, done_today=3), + "summarize": _stage(pending=7, failed=1, done_1h=10, done_today=20), + } + split = _split(macbook={"done_1h": 4, "done_today": 8}, macmini={"done_1h": 6, "done_today": 12}) + machines = build_machines(stats, split, [], deep_enabled=True) + macmini = _machine(machines, "macmini") + macbook = _machine(machines, "macbook") + + assert macmini["pending"] == 7 and macmini["failed"] == 1 + assert macmini["done_1h"] == 2 + 6 # classify + macmini 몫 (10 아님) + assert macmini["done_today"] == 3 + 12 + assert macbook["done_1h"] == 4 and macbook["done_today"] == 8 + assert macbook["pending"] == 0 # 풀 pending 은 macmini 만 + + +def test_deep_disabled_deep_summary_counts_to_macmini(): + stats = {"deep_summary": _stage(pending=2, processing=1, done_1h=3, done_today=4)} + machines = build_machines(stats, _split(), [], deep_enabled=False) + macmini = _machine(machines, "macmini") + macbook = _machine(machines, "macbook") + assert macmini["pending"] == 2 and macmini["processing"] == 1 + assert macmini["done_1h"] == 3 and macmini["done_today"] == 4 + assert macbook["stages"] == [] and macbook["pending"] == 0 + assert _machine(machines, "macmini")["stages"] == ["classify", "summarize", "deep_summary"] + + +def test_deferred_pending_always_on_macbook_card(): + """보류(deferred_until 미래)는 summarize+deep_summary 합산으로 macbook 카드 귀속. + deep 슬롯 유무와 무관 (보류 = 맥북 불가 신호).""" + stats = { + "summarize": _stage(pending=5, deferred_pending=2), + "deep_summary": _stage(pending=1, deferred_pending=1), + } + for deep_enabled in (True, False): + machines = build_machines(stats, _split(), [], deep_enabled=deep_enabled) + assert _machine(machines, "macbook")["deferred_pending"] == 3 + assert _machine(machines, "gpu")["deferred_pending"] == 0 + assert _machine(machines, "macmini")["deferred_pending"] == 0 + + +# ─── state 판정 ─────────────────────────────────────────────────────────────── + +def test_macbook_state_deferred_wins_over_active(): + stats = {"summarize": _stage(pending=1, deferred_pending=1)} + split = _split(macbook={"done_15m": 3}) # 최근 완료가 있어도 deferred 우선 + machines = build_machines(stats, split, [], deep_enabled=True) + assert _machine(machines, "macbook")["state"] == "deferred" + + +def test_macbook_state_active_on_recent_qwen_done(): + split = _split(macbook={"done_15m": 1}) + machines = build_machines({}, split, [], deep_enabled=True) + assert _machine(machines, "macbook")["state"] == "active" + + +def test_macbook_state_idle(): + machines = build_machines({}, _split(), [], deep_enabled=True) + assert _machine(machines, "macbook")["state"] == "idle" + + +def test_gpu_state_active_on_processing(): + stats = {"extract": _stage(processing=1)} + machines = build_machines(stats, _split(), [], deep_enabled=True) + assert _machine(machines, "gpu")["state"] == "active" + + +def test_gpu_state_active_on_recent_done(): + stats = {"embed": _stage(done_15m=2)} + machines = build_machines(stats, _split(), [], deep_enabled=True) + assert _machine(machines, "gpu")["state"] == "active" + + +def test_gpu_state_idle_when_old_done_only(): + stats = {"embed": _stage(done_1h=5, done_today=9)} # 15분 내 완료 없음 + machines = build_machines(stats, _split(), [], deep_enabled=True) + assert _machine(machines, "gpu")["state"] == "idle" + + +def test_macmini_state_not_active_on_macbook_pool_done(): + """summarize 풀 완료가 전부 macbook 몫이면 macmini 는 active 아님 (귀속 기준).""" + stats = {"summarize": _stage(done_15m=1)} + split = _split(macbook={"done_15m": 1}) + machines = build_machines(stats, split, [], deep_enabled=True) + assert _machine(machines, "macmini")["state"] == "idle" + + +def test_macmini_state_active_on_summarize_processing(): + stats = {"summarize": _stage(processing=1)} + machines = build_machines(stats, _split(), [], deep_enabled=True) + assert _machine(machines, "macmini")["state"] == "active" + + +# ─── current 귀속 ───────────────────────────────────────────────────────────── + +def test_current_summarize_to_macmini_max_two(): + rows = [ + {"stage": "summarize", "document_id": 1, "title": "문서A", "original_filename": None, "file_path": None}, + {"stage": "summarize", "document_id": 2, "title": "문서B", "original_filename": None, "file_path": None}, + {"stage": "summarize", "document_id": 3, "title": "문서C", "original_filename": None, "file_path": None}, + {"stage": "extract", "document_id": 4, "title": "문서D", "original_filename": None, "file_path": None}, + ] + machines = build_machines({}, _split(), rows, deep_enabled=True) + macmini = _machine(machines, "macmini") + gpu = _machine(machines, "gpu") + assert [c["document_id"] for c in macmini["current"]] == [1, 2] # 최대 2건 + assert macmini["current"][0] == {"document_id": 1, "title": "문서A", "stage": "summarize"} + assert [c["document_id"] for c in gpu["current"]] == [4] + assert _machine(machines, "macbook")["current"] == [] + + +def test_current_deep_summary_follows_deep_slot(): + rows = [{"stage": "deep_summary", "document_id": 9, "title": "심층", "original_filename": None, "file_path": None}] + enabled = build_machines({}, _split(), rows, deep_enabled=True) + disabled = build_machines({}, _split(), rows, deep_enabled=False) + assert _machine(enabled, "macbook")["current"][0]["document_id"] == 9 + assert _machine(disabled, "macmini")["current"][0]["document_id"] == 9 + + +def test_display_title_fallback_chain(): + assert display_title({"document_id": 1, "title": "제목"}) == "제목" + assert display_title({"document_id": 1, "title": None, "original_filename": "a.pdf"}) == "a.pdf" + assert display_title( + {"document_id": 1, "title": None, "original_filename": None, "file_path": "/documents/PKM/Inbox/b.hwp"} + ) == "b.hwp" + assert display_title( + {"document_id": 7, "title": None, "original_filename": None, "file_path": None} + ) == "문서 #7" + + +# ─── summarize ETA ──────────────────────────────────────────────────────────── + +def test_eta_minutes_positive_drain(): + # 순소화 6건/h, 잔량 30건 → 300분 + assert compute_eta_minutes(30, 10, 4) == 300 + + +def test_eta_minutes_null_when_not_draining(): + assert compute_eta_minutes(30, 4, 10) is None # 유입 > 소화 + assert compute_eta_minutes(30, 5, 5) is None # 동률도 null + assert compute_eta_minutes(30, 0, 0) is None + + +def test_eta_minutes_zero_pending(): + assert compute_eta_minutes(0, 10, 4) == 0 + + +def test_build_summarize_eta_pending_includes_deferred(): + stats = {"summarize": _stage(pending=12, deferred_pending=5, done_1h=8, created_1h=2)} + eta = build_summarize_eta(stats) + assert eta == { + "pending": 12, # 보류 포함 총수 (pending 자체에 deferred 포함) + "done_rate_1h": 8, + "inflow_rate_1h": 2, + "eta_minutes": round(12 / 6 * 60), + } + + +def test_build_summarize_eta_empty_stats(): + eta = build_summarize_eta({}) + assert eta == {"pending": 0, "done_rate_1h": 0, "inflow_rate_1h": 0, "eta_minutes": None} + + +# ─── trend 24h ──────────────────────────────────────────────────────────────── + +def test_trend_24_buckets_oldest_first_with_gaps(): + now_kst = datetime(2026, 6, 11, 14, 30, tzinfo=KST) + inflow = {"2026-06-11 13:00": 3, "2026-06-10 15:00": 1} # 15:00 어제 = 최고령 버킷 + done = {"2026-06-11 14:00": 2} + trend = build_trend(inflow, done, now_kst) + + assert len(trend) == 24 + assert trend[0] == {"hour": "15:00", "inflow": 1, "done": 0} # 오래된 것부터 + assert trend[-1] == {"hour": "14:00", "inflow": 0, "done": 2} # 현재 시각 버킷 + assert trend[-2] == {"hour": "13:00", "inflow": 3, "done": 0} + # 빈 버킷은 0 + assert sum(b["inflow"] for b in trend) == 4 + assert sum(b["done"] for b in trend) == 2 + + +def test_trend_ignores_out_of_window_bucket(): + """창 밖(24버킷 미포함) key 는 무시 — cutoff 경계 행이 섞여도 안전.""" + now_kst = datetime(2026, 6, 11, 14, 30, tzinfo=KST) + inflow = {"2026-06-10 14:00": 99} # 14:00 어제 — 창의 최고령(15:00 어제) 이전 + trend = build_trend(inflow, {}, now_kst) + assert sum(b["inflow"] for b in trend) == 0 + + +def test_trend_kst_midnight_crossing_labels(): + now_kst = datetime(2026, 6, 11, 2, 5, tzinfo=KST) + trend = build_trend({}, {}, now_kst) + assert trend[-1]["hour"] == "02:00" + assert trend[0]["hour"] == "03:00" # 전날 03:00 (라벨은 HH:00 만) + assert [b["hour"] for b in trend[-3:]] == ["00:00", "01:00", "02:00"] + + +# ─── totals / row 변환 / 전체 조립 ─────────────────────────────────────────── + +def test_totals_sum_all_stages(): + stats = { + "extract": _stage(pending=1, processing=2, failed=3), + "summarize": _stage(pending=4, failed=1), + "deep_summary": _stage(pending=2), + } + assert build_totals(stats) == {"pending": 7, "processing": 2, "failed": 4} + + +def test_rows_to_stage_stats_conversion(): + rows = [ + ("extract", 3, 1, 0, 5, 9, 1, 0, 2), + ("summarize", 7, None, 1, 10, 20, 0, 2, 4), # None 방어 + ] + stats = rows_to_stage_stats(rows) + assert stats["extract"]["pending"] == 3 and stats["extract"]["created_1h"] == 2 + assert stats["summarize"]["processing"] == 0 + assert stats["summarize"]["deferred_pending"] == 2 + + +def test_rows_to_summarize_split_conversion(): + rows = [ + (True, 4, 8, 1), # is_macbook + (False, 6, 12, 0), + ] + split = rows_to_summarize_split(rows) + assert split["macbook"] == {"done_1h": 4, "done_today": 8, "done_15m": 1} + assert split["macmini"] == {"done_1h": 6, "done_today": 12, "done_15m": 0} + + +def test_rows_to_summarize_split_empty(): + split = rows_to_summarize_split([]) + assert split["macbook"]["done_1h"] == 0 and split["macmini"]["done_today"] == 0 + + +def test_compose_overview_contract_shape(): + """응답 dict 의 키가 FE 계약 shape 과 정확히 일치하는지 고정.""" + out = compose_overview( + {"summarize": _stage(pending=1)}, + _split(), + {}, {}, [], + deep_enabled=True, + now_kst=datetime(2026, 6, 11, 14, 30, tzinfo=KST), + ) + assert set(out.keys()) == {"machines", "summarize_eta", "trend_24h", "totals"} + assert [m["key"] for m in out["machines"]] == ["gpu", "macmini", "macbook"] + for m in out["machines"]: + assert set(m.keys()) == { + "key", "label", "state", "stages", "pending", "processing", "failed", + "done_1h", "done_today", "deferred_pending", "current", + } + assert m["state"] in ("active", "deferred", "idle") + assert set(out["summarize_eta"].keys()) == {"pending", "done_rate_1h", "inflow_rate_1h", "eta_minutes"} + assert len(out["trend_24h"]) == 24 + assert set(out["trend_24h"][0].keys()) == {"hour", "inflow", "done"} + assert set(out["totals"].keys()) == {"pending", "processing", "failed"} + # 머신 label 고정 (raw 모델명 노출 금지 — label 만) + assert [m["label"] for m in out["machines"]] == ["GPU 서버", "맥미니", "맥북 M5 Max"]