"""대시보드 위젯 데이터 API""" from typing import Annotated from fastapi import APIRouter, Depends from pydantic import BaseModel from sqlalchemy import func, select, text from sqlalchemy.ext.asyncio import AsyncSession from core.auth import get_current_user from core.database import get_session from models.document import Document from models.queue import ProcessingQueue from models.user import User router = APIRouter() class DomainCount(BaseModel): domain: str | None count: int class RecentDocument(BaseModel): id: int title: str | None file_format: str ai_domain: str | None created_at: str class PipelineStatus(BaseModel): stage: str status: str count: int class DashboardResponse(BaseModel): today_added: int today_by_domain: list[DomainCount] inbox_count: int law_alerts: int recent_documents: list[RecentDocument] pipeline_status: list[PipelineStatus] failed_count: int total_documents: int # 카운트 분리: 문서함(비-note/비-news) / 메모(memo+note) / 뉴스(news) documents_count: int = 0 memos_count: int = 0 news_count: int = 0 @router.get("/", response_model=DashboardResponse) async def get_dashboard( user: Annotated[User, Depends(get_current_user)], session: Annotated[AsyncSession, Depends(get_session)], ): """대시보드 위젯 데이터 집계""" # 오늘 추가된 문서 today_result = await session.execute( select(Document.ai_domain, func.count(Document.id)) .where(func.date(Document.created_at) == func.current_date()) .group_by(Document.ai_domain) ) today_rows = today_result.all() today_added = sum(row[1] for row in today_rows) # Inbox 미분류 수 (review_status = pending) inbox_result = await session.execute( select(func.count(Document.id)) .where( Document.review_status == "pending", Document.deleted_at == None, ) ) inbox_count = inbox_result.scalar() or 0 # 법령 알림 (오늘) law_result = await session.execute( select(func.count(Document.id)) .where( Document.source_channel == "law_monitor", func.date(Document.created_at) == func.current_date(), ) ) law_alerts = law_result.scalar() or 0 # 최근 문서 7건 recent_result = await session.execute( select(Document) .order_by(Document.created_at.desc()) .limit(7) ) recent_docs = recent_result.scalars().all() # 파이프라인 상태 (24h) pipeline_result = await session.execute( text(""" SELECT stage, status, COUNT(*) FROM processing_queue WHERE created_at > NOW() - INTERVAL '24 hours' GROUP BY stage, status """) ) # 실패 건수 failed_result = await session.execute( select(func.count()) .select_from(ProcessingQueue) .where(ProcessingQueue.status == "failed") ) failed_count = failed_result.scalar() or 0 # 전체 문서 수 + 카테고리별 분리 (단일 쿼리) # 문서함: 비-note, 비-news / 메모: memo+note / 뉴스: news 유입 경로 기준 count_result = await session.execute( text(""" SELECT COUNT(*) AS total, COUNT(*) FILTER (WHERE source_channel != 'news' AND file_type != 'note') AS documents, COUNT(*) FILTER (WHERE source_channel = 'memo' AND file_type = 'note') AS memos, COUNT(*) FILTER (WHERE source_channel = 'news') AS news FROM documents WHERE deleted_at IS NULL """) ) counts = count_result.one() total_documents = counts[0] documents_count = counts[1] memos_count = counts[2] news_count = counts[3] return DashboardResponse( today_added=today_added, today_by_domain=[ DomainCount(domain=row[0], count=row[1]) for row in today_rows ], inbox_count=inbox_count, law_alerts=law_alerts, recent_documents=[ RecentDocument( id=doc.id, title=doc.title, file_format=doc.file_format, ai_domain=doc.ai_domain, created_at=doc.created_at.isoformat() if doc.created_at else "", ) for doc in recent_docs ], pipeline_status=[ PipelineStatus(stage=row[0], status=row[1], count=row[2]) for row in pipeline_result ], failed_count=failed_count, total_documents=total_documents, documents_count=documents_count, memos_count=memos_count, news_count=news_count, )