Files
hyungi_document_server/app/main.py
T
Hyungi Ahn 9d4aa201a8 feat(study): study_questions 자동 임베딩 (PR-4)
문제 본문 + 보기 1~4 → bge-m3 1024차원. status 자체가 큐 역할 (별도 큐
테이블 없음 — ProcessingQueue 인프라 영향 0). APScheduler 1분 cron 이
status in {none, failed, stale} 행을 batch=10 처리. 새 문제는 default
'none' 으로 자동 backfill.

데이터 모델 (migrations 193~194):
- study_questions: embedding vector(1024), embedding_status VARCHAR(20)
  DEFAULT 'none' (none/pending/ready/failed/stale), embedding_updated_at,
  embedding_model
- HNSW partial index (vector_cosine_ops) WHERE deleted_at IS NULL AND
  embedding IS NOT NULL — bge-m3 cosine 기준, documents.embedding (ivfflat)
  과 ops 일관

재계산 트리거: question_text / choice_1~4 변경 시 ready→stale 자동.
correct_choice / explanation / subject / scope 변경은 재계산 안 함
(의미 검색에 영향 없음).

워커 (workers/study_question_embed_worker.py):
- race-safe pending 마킹 (조건부 UPDATE WHERE status IN none/failed/stale)
- AIClient.embed(text) bge-m3 호출, 15s timeout
- 실패 시 status='failed', 직전 embedding 보존, 다음 cron 틱에 재시도
- 본문 = "문제: ...\n보기:\n1. ...\n2. ...\n3. ...\n4. ..." (subject/scope
  의도 제외 — 분류명이 의미 검색 노이즈)

후속 PR 예정: 비슷한 문제 검색 UI / 중복 입력 감지 / RAG 정확도 향상 /
오답 클러스터링. 본 PR 은 임베딩 저장·재계산·backfill 까지만.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 08:54:02 +09:00

186 lines
8.2 KiB
Python

"""hyungi_Document_Server — FastAPI 엔트리포인트"""
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
from sqlalchemy import func, select, text
from api.audio import router as audio_router
from api.auth import router as auth_router
from api.config import router as config_router
from api.dashboard import router as dashboard_router
from api.digest import router as digest_router
from api.document_notes import router as document_notes_router
from api.document_reads import router as document_reads_router
from api.documents import router as documents_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.search import router as search_router
from api.setup import router as setup_router
from api.study_questions import router as study_questions_router
from api.study_sessions import router as study_sessions_router
from api.study_topics import router as study_topics_router
from api.video import router as video_router
from core.config import settings
from core.database import async_session, engine, init_db
from models.user import User
@asynccontextmanager
async def lifespan(app: FastAPI):
"""앱 시작/종료 시 실행되는 lifespan 핸들러"""
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from services.search.query_analyzer import prewarm_analyzer
from workers.daily_digest import run as daily_digest_run
from workers.digest_worker import run as global_digest_run
from workers.file_watcher import watch_inbox
from workers.law_monitor import run as law_monitor_run
from workers.mailplus_archive import run as mailplus_run
from workers.news_collector import run as news_collector_run
from workers.queue_consumer import consume_queue
from workers.study_question_embed_worker import run as study_q_embed_run
from workers.tier_backfill import run as tier_backfill_run
from workers.upload_cleanup import cleanup_orphan_uploads
# 시작: DB 연결 확인
await init_db()
# NAS 마운트 확인 (NFS 미마운트 시 로컬 빈 디렉토리에 쓰는 것 방지)
from pathlib import Path
nas_check = Path(settings.nas_mount_path) / "PKM"
if not nas_check.is_dir():
raise RuntimeError(
f"NAS 마운트 확인 실패: {nas_check} 디렉토리 없음. "
f"NFS 마운트 상태를 확인하세요."
)
# APScheduler: 백그라운드 작업
scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
# 상시 실행
scheduler.add_job(consume_queue, "interval", minutes=1, id="queue_consumer")
scheduler.add_job(watch_inbox, "interval", minutes=5, id="file_watcher")
scheduler.add_job(cleanup_orphan_uploads, "interval", minutes=10, id="upload_cleanup")
# PR-4: study_questions 자동 임베딩 (status='none/failed/stale' 행을 batch=10 처리).
# 별도 큐 테이블 없이 status 자체가 큐. backfill 도 cron 이 'none' 행을 자연스럽게 처리.
scheduler.add_job(study_q_embed_run, "interval", minutes=1, id="study_q_embed")
# PR-B 레거시 tier 백필 — 30분 주기로 호출되지만 KST 00:00~06:00 시간대만 실제 enqueue.
# safety > law > manual 우선순위로 25건씩. 6720 레거시 → 야간당 ~150건 → 약 45일 소화.
scheduler.add_job(tier_backfill_run, "interval", minutes=30, id="tier_backfill")
# 일일 스케줄 (KST)
scheduler.add_job(law_monitor_run, CronTrigger(hour=7), id="law_monitor")
scheduler.add_job(mailplus_run, CronTrigger(hour=7), id="mailplus_morning")
scheduler.add_job(mailplus_run, CronTrigger(hour=18), id="mailplus_evening")
scheduler.add_job(daily_digest_run, CronTrigger(hour=20), id="daily_digest")
scheduler.add_job(global_digest_run, CronTrigger(hour=4, minute=0), id="global_digest")
scheduler.add_job(news_collector_run, "interval", hours=6, id="news_collector")
scheduler.start()
# Phase 2.1 (async 구조): QueryAnalyzer prewarm.
# 대표 쿼리 15~20개를 background task로 분석해 cache 적재.
# 첫 사용자 요청부터 cache hit rate 70~80% 목표.
# 논블로킹 — startup을 막지 않음. MLX 부하 완화 위해 delay_between=0.5.
prewarm_task = asyncio.create_task(prewarm_analyzer())
prewarm_task.add_done_callback(
lambda t: t.exception() and None # 예외는 query_analyzer 내부에서 로깅
)
yield
# 종료: 스케줄러 → DB 순서로 정리
scheduler.shutdown(wait=False)
await engine.dispose()
app = FastAPI(
title="hyungi_Document_Server",
description="Self-hosted PKM 웹 애플리케이션 API",
version="2.0.0",
lifespan=lifespan,
)
# ─── 라우터 등록 ───
app.include_router(setup_router, prefix="/api/setup", tags=["setup"])
app.include_router(config_router, prefix="/api/config", tags=["config"])
app.include_router(auth_router, prefix="/api/auth", tags=["auth"])
app.include_router(documents_router, prefix="/api/documents", tags=["documents"])
# 회독 카운트 — /api/documents/{id}/read* 경로. documents_router 와 prefix 같아 충돌 없음.
app.include_router(document_reads_router, prefix="/api/documents", tags=["document-reads"])
app.include_router(document_notes_router, prefix="/api/documents", tags=["document-notes"])
app.include_router(search_router, prefix="/api/search", tags=["search"])
app.include_router(memos_router, prefix="/api/memos", tags=["memos"])
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"])
app.include_router(digest_router, prefix="/api/digest", tags=["digest"])
app.include_router(audio_router, prefix="/api/audio", tags=["audio"])
app.include_router(video_router, prefix="/api/video", tags=["video"])
app.include_router(study_sessions_router, prefix="/api/study-sessions", tags=["study-sessions"])
app.include_router(study_topics_router, prefix="/api/study-topics", tags=["study-topics"])
# study_questions: 라우터 안에서 /study-topics/{id}/questions 와 /study-questions/{id} 두 줄기를 모두 정의하므로 prefix=/api 로 등록
app.include_router(study_questions_router, prefix="/api", tags=["study-questions"])
# TODO: Phase 5에서 추가
# app.include_router(tasks.router, prefix="/api/tasks", tags=["tasks"])
# app.include_router(export.router, prefix="/api/export", tags=["export"])
# ─── 셋업 미들웨어: 유저 0명이면 /setup으로 리다이렉트 ───
SETUP_BYPASS_PREFIXES = (
"/api/setup", "/api/config", "/setup", "/health", "/docs", "/openapi.json", "/redoc",
)
@app.middleware("http")
async def setup_redirect_middleware(request: Request, call_next):
path = request.url.path
# 바이패스 경로는 항상 통과
if any(path.startswith(p) for p in SETUP_BYPASS_PREFIXES):
return await call_next(request)
# 유저 존재 여부 확인
try:
async with async_session() as session:
result = await session.execute(select(func.count(User.id)))
user_count = result.scalar()
if user_count == 0:
return RedirectResponse(url="/setup")
except Exception:
pass # DB 연결 실패 시 통과 (health에서 확인 가능)
return await call_next(request)
# ─── 셋업 페이지 라우트 (API가 아닌 HTML 페이지) ───
@app.get("/setup")
async def setup_page_redirect(request: Request):
"""셋업 위자드 페이지로 포워딩"""
from api.setup import setup_page
from core.database import get_session
async for session in get_session():
return await setup_page(request, session)
@app.get("/health")
async def health_check():
"""헬스체크 — DB 연결 상태 포함"""
db_ok = False
try:
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
db_ok = True
except Exception:
pass
return {
"status": "ok" if db_ok else "degraded",
"version": "2.0.0",
"database": "connected" if db_ok else "disconnected",
}