Spaces:

juhaha1230
/

AI_Menu_Search

Sleeping

Juhaha

feat: QC routes + query policy + entity/prefix promotion + curated aliases (dio_qc_20260907)

c65fc74 about 2 months ago

14.3 kB

"""
영웅문 S# AI 메뉴 검색 — REST API 서버
FastAPI + Swagger UI
실행:
    .venv/Scripts/uvicorn.exe api.main:app --host 0.0.0.0 --port 8000
    또는
    .venv/Scripts/python.exe -m uvicorn api.main:app --host 0.0.0.0 --port 8000
Swagger UI: http://localhost:8000/docs
ReDoc:      http://localhost:8000/redoc
"""
import sys
import json
import hashlib
import logging
import os
from datetime import datetime
from pathlib import Path
from typing import Optional

sys.path.insert(0, str(Path(__file__).parent.parent))

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from config import SIMILARITY_THRESHOLD
from core.dataset_manifest import resolve_app_version

# ── 로거 설정 ─────────────────────────────────────────────────────────────────
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

# ── 로그 파일 경로 ─────────────────────────────────────────────────────────────
LOG_DIR = Path(__file__).parent.parent / "logs"
LOG_DIR.mkdir(exist_ok=True)
QUERY_LOG_PATH = LOG_DIR / "api_queries.jsonl"

# ── FastAPI 앱 ────────────────────────────────────────────────────────────────
app = FastAPI(
    title="영웅문 S# AI 메뉴 검색 API",
    description="""
자연어로 영웅문 S# 앱 메뉴를 검색합니다.
## 주요 기능
- 구어체 자연어 질문 → 관련 메뉴 Top-N 추천
- HyDE 기반 어휘 갭 해소
- 선택적으로 품질 미달 쿼리 재작성 (기본 비활성)
- 카테고리 자동 감지 및 필터
## 사용 예시
```
POST /api/search
{ "query": "내 주식 얼마야" }
```
    """,
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# CORS. 내부망 배포에서는 API_CORS_ALLOW_ORIGINS에 허용 origin을 쉼표로 지정한다.
_cors_origins = [
    value.strip()
    for value in os.getenv("API_CORS_ALLOW_ORIGINS", "*").split(",")
    if value.strip()
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=_cors_origins,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ── 카테고리 목록 ─────────────────────────────────────────────────────────────
CATEGORIES = [
    "국내주식", "해외주식", "파생상품",
    "자산/뱅킹", "금융상품", "주식더모으기",
    "랭킹/영웅전", "대출/카드/보험", "환전", "생활/혜택",
]


# ── Pydantic 모델 ─────────────────────────────────────────────────────────────

class SearchRequest(BaseModel):
    query: str = Field(
        ...,
        description="자연어 검색어",
        examples=["내 주식 얼마야", "주문 취소하고 싶어", "달러 환전하고 싶어"],
    )
    top_n: int = Field(5, ge=1, le=20, description="반환할 결과 수 (기본 5, 최대 20)")
    category_filter: Optional[str] = Field(
        None,
        description=f"카테고리 필터. 전체 검색 시 null. 가능한 값: {CATEGORIES}",
        examples=["국내주식", "해외주식"],
    )
    use_reranker: bool = Field(False, description="Cross-Encoder 리랭킹 활성화 (기본 false)")
    use_hyde: bool = Field(True, description="HyDE 활성화 — 어휘 갭 해소 (권장, 기본 true)")
    allow_query_rewrite: bool = Field(
        False,
        description="품질 미달 시 LLM 쿼리 재작성 허용 (실험용, 기본 false)",
    )
    app_version: Optional[str] = Field(
        None,
        description="사용자 앱 버전. 생략 시 현재 운영 데이터 버전",
        examples=["1.5.5"],
    )


class MenuResult(BaseModel):
    rank: int = Field(..., description="검색 순위")
    menu_id: str = Field(..., description="메뉴 ID (예: SCR_3032)")
    menu_name: str = Field(..., description="메뉴명 (예: 국내잔고)")
    menu_path: str = Field(..., description="메뉴 경로 (예: 국내주식 > 계좌 > 국내잔고)")
    category: Optional[str] = Field(None, description="카테고리")
    similarity: float = Field(..., description="정규화된 검색 순위 점수 (0.0 ~ 1.0)")
    similarity_pct: str = Field(..., description="검색 순위 점수 퍼센트")
    confidence: float = Field(..., description="Dense/BM25/Q↔Q 검색 근거 신뢰도")


class SearchResponse(BaseModel):
    query: str = Field(..., description="원본 검색어")
    refined_query: str = Field(..., description="최종 검색 쿼리 (재작성된 경우 변경됨)")
    detected_category: Optional[str] = Field(None, description="자동 감지된 카테고리")
    intent: str = Field(..., description="감지된 의도 (주문/잔고조회/시세/차트/기타)")
    results: list[MenuResult] = Field(..., description="검색 결과 목록")
    result_count: int = Field(..., description="반환된 결과 수")
    quality_score: float = Field(..., description="검색 품질 점수 (0.0 ~ 1.0)")
    is_satisfactory: bool = Field(..., description="품질 만족 여부 (0.55 이상 시 true)")
    retry_count: int = Field(..., description="쿼리 재작성 횟수 (0 ~ 2)")
    rewrite_history: list[str] = Field(default_factory=list, description="쿼리 재작성 이력")
    elapsed_ms: int = Field(..., description="처리 시간 (밀리초)")
    app_version: str = Field(..., description="검색에 사용된 앱 버전")
    dataset_version: str = Field(..., description="메뉴 데이터셋 버전")
    index_version: str = Field(..., description="검색 인덱스 버전")
    applicable: bool = Field(..., description="임계값 이상 메뉴 결과 존재 여부")


class HealthResponse(BaseModel):
    status: str = Field(..., description="서버 상태 (ok / error)")
    menu_count: int = Field(..., description="인덱싱된 메뉴 수")
    version: str = Field(..., description="API 버전")
    timestamp: str = Field(..., description="현재 서버 시각")


class CategoriesResponse(BaseModel):
    categories: list[str] = Field(..., description="사용 가능한 카테고리 목록")


# ── 헬퍼: 로깅 ───────────────────────────────────────────────────────────────

def _log_query(req: SearchRequest, resp: SearchResponse) -> None:
    """검색 품질 로그. 사용자 원문은 명시적으로 허용한 경우에만 저장한다."""
    if os.getenv("API_QUERY_LOG_ENABLED", "1") not in ("1", "true", "True"):
        return
    entry = {
        "timestamp": datetime.now().isoformat(),
        "query_hash": hashlib.sha256(req.query.encode("utf-8")).hexdigest(),
        "category_filter": req.category_filter,
        "detected_category": resp.detected_category,
        "app_version": resp.app_version,
        "dataset_version": resp.dataset_version,
        "results": [
            {
                "rank": r.rank,
                "menu_id": r.menu_id,
                "menu_name": r.menu_name,
                "similarity": r.similarity,
            }
            for r in resp.results
        ],
        "quality_score": resp.quality_score,
        "retry_count": resp.retry_count,
        "elapsed_ms": resp.elapsed_ms,
    }
    if os.getenv("API_LOG_QUERY_TEXT", "0") in ("1", "true", "True"):
        entry["query"] = req.query
        entry["refined_query"] = resp.refined_query
    with open(QUERY_LOG_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")


# ── 엔진 초기화 ───────────────────────────────────────────────────────────────

_engines: dict[str, object] = {}

def _get_engine(app_version: str):
    if app_version not in _engines:
        from core.search_engine import MenuSearchEngine
        _engines[app_version] = MenuSearchEngine.get_instance(app_version)
    return _engines[app_version]


@app.on_event("startup")
async def startup_event():
    """서버 시작 시 검색 엔진 미리 로드"""
    logger.info("검색 엔진 초기화 중...")
    app_version, _ = resolve_app_version(None)
    _get_engine(app_version)
    logger.info("검색 엔진 초기화 완료")


# ── 엔드포인트 ────────────────────────────────────────────────────────────────

@app.post(
    "/api/search",
    response_model=SearchResponse,
    summary="메뉴 검색",
    description="자연어 쿼리로 영웅문 S# 메뉴를 검색합니다.",
    tags=["검색"],
)
async def search_menu(req: SearchRequest):
    from core.agent import run_agent

    start = datetime.now()

    # 카테고리 필터 검증
    if req.category_filter and req.category_filter not in CATEGORIES:
        raise HTTPException(
            status_code=400,
            detail=f"유효하지 않은 카테고리: '{req.category_filter}'. 가능한 값: {CATEGORIES}",
        )

    try:
        selected_app_version, manifest = resolve_app_version(req.app_version)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

    try:
        result = run_agent(
            query=req.query,
            top_n=req.top_n,
            threshold=SIMILARITY_THRESHOLD,
            category_filter=req.category_filter,
            use_reranker=req.use_reranker,
            use_hyde=req.use_hyde,
            allow_rewrite=req.allow_query_rewrite,
            app_version=selected_app_version,
        )
    except Exception as e:
        logger.error(f"검색 오류: {e}")
        raise HTTPException(status_code=500, detail=f"검색 중 오류 발생: {str(e)}")

    elapsed_ms = int((datetime.now() - start).total_seconds() * 1000)

    # 결과 변환
    menu_results = []
    for i, r in enumerate(result.get("search_results", []), 1):
        menu_results.append(MenuResult(
            rank=i,
            menu_id=r.get("menu_id", ""),
            menu_name=r.get("menu_name", r.get("menu_id", "")),
            menu_path=r.get("menu_path", ""),
            category=r.get("category"),
            similarity=round(r.get("similarity", 0.0), 4),
            similarity_pct=r.get("similarity_pct", f"{r.get('similarity', 0):.1%}"),
            confidence=round(r.get("confidence", 0.0), 4),
        ))

    response = SearchResponse(
        query=req.query,
        refined_query=result.get("refined_query", req.query),
        detected_category=result.get("detected_category"),
        intent=result.get("intent", "기타"),
        results=menu_results,
        result_count=len(menu_results),
        quality_score=round(result.get("quality_score", 0.0), 4),
        is_satisfactory=result.get("is_satisfactory", False),
        retry_count=result.get("retry_count", 0),
        rewrite_history=result.get("rewrite_history", []),
        elapsed_ms=elapsed_ms,
        app_version=selected_app_version,
        dataset_version=str(manifest.get("dataset_version", "")),
        index_version=str(manifest.get("index_version", "")),
        applicable=bool(menu_results),
    )

    # 로그 기록
    try:
        _log_query(req, response)
    except Exception as e:
        logger.warning(f"로그 기록 실패: {e}")

    logger.info(
        f"검색 완료 | query_hash={hashlib.sha256(req.query.encode('utf-8')).hexdigest()[:12]} | "
        f"결과={len(menu_results)}개 | 품질={response.quality_score:.2f} | "
        f"재시도={response.retry_count} | {elapsed_ms}ms"
    )

    return response


@app.get(
    "/api/health",
    response_model=HealthResponse,
    summary="헬스체크",
    description="서버 상태 및 기본 정보를 확인합니다.",
    tags=["시스템"],
)
async def health(app_version: Optional[str] = None):
    try:
        selected_app_version, manifest = resolve_app_version(app_version)
        engine = _get_engine(selected_app_version)
        stats = engine.vectorstore.get_stats()
        return HealthResponse(
            status="ok",
            menu_count=stats.get("total_menus", 0),
            version=str(manifest.get("index_version", "1.0.0")),
            timestamp=datetime.now().isoformat(),
        )
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"서버 오류: {str(e)}")


@app.get(
    "/api/categories",
    response_model=CategoriesResponse,
    summary="카테고리 목록",
    description="검색 필터로 사용 가능한 카테고리 목록을 반환합니다.",
    tags=["메타"],
)
async def get_categories():
    return CategoriesResponse(categories=CATEGORIES)


@app.get(
    "/api/stats",
    summary="검색 통계",
    description="누적 검색 쿼리 수 등 운영 통계를 반환합니다.",
    tags=["시스템"],
)
async def get_stats():
    total_queries = 0
    if QUERY_LOG_PATH.exists():
        with open(QUERY_LOG_PATH, "r", encoding="utf-8") as f:
            total_queries = sum(1 for _ in f)

    return {
        "total_queries": total_queries,
        "log_path": str(QUERY_LOG_PATH),
        "timestamp": datetime.now().isoformat(),
    }


# ── 직접 실행 ─────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "api.main:app",
        host="0.0.0.0",
        port=8000,
        reload=False,
        log_level="info",
    )

	"""
	영웅문 S# AI 메뉴 검색 — REST API 서버
	FastAPI + Swagger UI

	실행:
	.venv/Scripts/uvicorn.exe api.main:app --host 0.0.0.0 --port 8000
	또는
	.venv/Scripts/python.exe -m uvicorn api.main:app --host 0.0.0.0 --port 8000

	Swagger UI: http://localhost:8000/docs
	ReDoc: http://localhost:8000/redoc
	"""
	import sys
	import json
	import hashlib
	import logging
	import os
	from datetime import datetime
	from pathlib import Path
	from typing import Optional

	sys.path.insert(0, str(Path(__file__).parent.parent))

	from fastapi import FastAPI, HTTPException, Request
	from fastapi.middleware.cors import CORSMiddleware
	from pydantic import BaseModel, Field
	from config import SIMILARITY_THRESHOLD
	from core.dataset_manifest import resolve_app_version

	# ── 로거 설정 ─────────────────────────────────────────────────────────────────
	logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
	logger = logging.getLogger(__name__)

	# ── 로그 파일 경로 ─────────────────────────────────────────────────────────────
	LOG_DIR = Path(__file__).parent.parent / "logs"
	LOG_DIR.mkdir(exist_ok=True)
	QUERY_LOG_PATH = LOG_DIR / "api_queries.jsonl"

	# ── FastAPI 앱 ────────────────────────────────────────────────────────────────
	app = FastAPI(
	title="영웅문 S# AI 메뉴 검색 API",
	description="""
	자연어로 영웅문 S# 앱 메뉴를 검색합니다.

	## 주요 기능
	- 구어체 자연어 질문 → 관련 메뉴 Top-N 추천
	- HyDE 기반 어휘 갭 해소
	- 선택적으로 품질 미달 쿼리 재작성 (기본 비활성)
	- 카테고리 자동 감지 및 필터

	## 사용 예시
	```
	POST /api/search
	{ "query": "내 주식 얼마야" }
	```
	""",
	version="1.0.0",
	docs_url="/docs",
	redoc_url="/redoc",
	)

	# CORS. 내부망 배포에서는 API_CORS_ALLOW_ORIGINS에 허용 origin을 쉼표로 지정한다.
	_cors_origins = [
	value.strip()
	for value in os.getenv("API_CORS_ALLOW_ORIGINS", "*").split(",")
	if value.strip()
	]
	app.add_middleware(
	CORSMiddleware,
	allow_origins=_cors_origins,
	allow_methods=["*"],
	allow_headers=["*"],
	)

	# ── 카테고리 목록 ─────────────────────────────────────────────────────────────
	CATEGORIES = [
	"국내주식", "해외주식", "파생상품",
	"자산/뱅킹", "금융상품", "주식더모으기",
	"랭킹/영웅전", "대출/카드/보험", "환전", "생활/혜택",
	]


	# ── Pydantic 모델 ─────────────────────────────────────────────────────────────

	class SearchRequest(BaseModel):
	query: str = Field(
	...,
	description="자연어 검색어",
	examples=["내 주식 얼마야", "주문 취소하고 싶어", "달러 환전하고 싶어"],
	)
	top_n: int = Field(5, ge=1, le=20, description="반환할 결과 수 (기본 5, 최대 20)")
	category_filter: Optional[str] = Field(
	None,
	description=f"카테고리 필터. 전체 검색 시 null. 가능한 값: {CATEGORIES}",
	examples=["국내주식", "해외주식"],
	)
	use_reranker: bool = Field(False, description="Cross-Encoder 리랭킹 활성화 (기본 false)")
	use_hyde: bool = Field(True, description="HyDE 활성화 — 어휘 갭 해소 (권장, 기본 true)")
	allow_query_rewrite: bool = Field(
	False,
	description="품질 미달 시 LLM 쿼리 재작성 허용 (실험용, 기본 false)",
	)
	app_version: Optional[str] = Field(
	None,
	description="사용자 앱 버전. 생략 시 현재 운영 데이터 버전",
	examples=["1.5.5"],
	)


	class MenuResult(BaseModel):
	rank: int = Field(..., description="검색 순위")
	menu_id: str = Field(..., description="메뉴 ID (예: SCR_3032)")
	menu_name: str = Field(..., description="메뉴명 (예: 국내잔고)")
	menu_path: str = Field(..., description="메뉴 경로 (예: 국내주식 > 계좌 > 국내잔고)")
	category: Optional[str] = Field(None, description="카테고리")
	similarity: float = Field(..., description="정규화된 검색 순위 점수 (0.0 ~ 1.0)")
	similarity_pct: str = Field(..., description="검색 순위 점수 퍼센트")
	confidence: float = Field(..., description="Dense/BM25/Q↔Q 검색 근거 신뢰도")


	class SearchResponse(BaseModel):
	query: str = Field(..., description="원본 검색어")
	refined_query: str = Field(..., description="최종 검색 쿼리 (재작성된 경우 변경됨)")
	detected_category: Optional[str] = Field(None, description="자동 감지된 카테고리")
	intent: str = Field(..., description="감지된 의도 (주문/잔고조회/시세/차트/기타)")
	results: list[MenuResult] = Field(..., description="검색 결과 목록")
	result_count: int = Field(..., description="반환된 결과 수")
	quality_score: float = Field(..., description="검색 품질 점수 (0.0 ~ 1.0)")
	is_satisfactory: bool = Field(..., description="품질 만족 여부 (0.55 이상 시 true)")
	retry_count: int = Field(..., description="쿼리 재작성 횟수 (0 ~ 2)")
	rewrite_history: list[str] = Field(default_factory=list, description="쿼리 재작성 이력")
	elapsed_ms: int = Field(..., description="처리 시간 (밀리초)")
	app_version: str = Field(..., description="검색에 사용된 앱 버전")
	dataset_version: str = Field(..., description="메뉴 데이터셋 버전")
	index_version: str = Field(..., description="검색 인덱스 버전")
	applicable: bool = Field(..., description="임계값 이상 메뉴 결과 존재 여부")


	class HealthResponse(BaseModel):
	status: str = Field(..., description="서버 상태 (ok / error)")
	menu_count: int = Field(..., description="인덱싱된 메뉴 수")
	version: str = Field(..., description="API 버전")
	timestamp: str = Field(..., description="현재 서버 시각")


	class CategoriesResponse(BaseModel):
	categories: list[str] = Field(..., description="사용 가능한 카테고리 목록")


	# ── 헬퍼: 로깅 ───────────────────────────────────────────────────────────────

	def _log_query(req: SearchRequest, resp: SearchResponse) -> None:
	"""검색 품질 로그. 사용자 원문은 명시적으로 허용한 경우에만 저장한다."""
	if os.getenv("API_QUERY_LOG_ENABLED", "1") not in ("1", "true", "True"):
	return
	entry = {
	"timestamp": datetime.now().isoformat(),
	"query_hash": hashlib.sha256(req.query.encode("utf-8")).hexdigest(),
	"category_filter": req.category_filter,
	"detected_category": resp.detected_category,
	"app_version": resp.app_version,
	"dataset_version": resp.dataset_version,
	"results": [
	{
	"rank": r.rank,
	"menu_id": r.menu_id,
	"menu_name": r.menu_name,
	"similarity": r.similarity,
	}
	for r in resp.results
	],
	"quality_score": resp.quality_score,
	"retry_count": resp.retry_count,
	"elapsed_ms": resp.elapsed_ms,
	}
	if os.getenv("API_LOG_QUERY_TEXT", "0") in ("1", "true", "True"):
	entry["query"] = req.query
	entry["refined_query"] = resp.refined_query
	with open(QUERY_LOG_PATH, "a", encoding="utf-8") as f:
	f.write(json.dumps(entry, ensure_ascii=False) + "\n")


	# ── 엔진 초기화 ───────────────────────────────────────────────────────────────

	_engines: dict[str, object] = {}

	def _get_engine(app_version: str):
	if app_version not in _engines:
	from core.search_engine import MenuSearchEngine
	_engines[app_version] = MenuSearchEngine.get_instance(app_version)
	return _engines[app_version]


	@app.on_event("startup")
	async def startup_event():
	"""서버 시작 시 검색 엔진 미리 로드"""
	logger.info("검색 엔진 초기화 중...")
	app_version, _ = resolve_app_version(None)
	_get_engine(app_version)
	logger.info("검색 엔진 초기화 완료")


	# ── 엔드포인트 ────────────────────────────────────────────────────────────────

	@app.post(
	"/api/search",
	response_model=SearchResponse,
	summary="메뉴 검색",
	description="자연어 쿼리로 영웅문 S# 메뉴를 검색합니다.",
	tags=["검색"],
	)
	async def search_menu(req: SearchRequest):
	from core.agent import run_agent

	start = datetime.now()

	# 카테고리 필터 검증
	if req.category_filter and req.category_filter not in CATEGORIES:
	raise HTTPException(
	status_code=400,
	detail=f"유효하지 않은 카테고리: '{req.category_filter}'. 가능한 값: {CATEGORIES}",
	)

	try:
	selected_app_version, manifest = resolve_app_version(req.app_version)
	except ValueError as e:
	raise HTTPException(status_code=400, detail=str(e))

	try:
	result = run_agent(
	query=req.query,
	top_n=req.top_n,
	threshold=SIMILARITY_THRESHOLD,
	category_filter=req.category_filter,
	use_reranker=req.use_reranker,
	use_hyde=req.use_hyde,
	allow_rewrite=req.allow_query_rewrite,
	app_version=selected_app_version,
	)
	except Exception as e:
	logger.error(f"검색 오류: {e}")
	raise HTTPException(status_code=500, detail=f"검색 중 오류 발생: {str(e)}")

	elapsed_ms = int((datetime.now() - start).total_seconds() * 1000)

	# 결과 변환
	menu_results = []
	for i, r in enumerate(result.get("search_results", []), 1):
	menu_results.append(MenuResult(
	rank=i,
	menu_id=r.get("menu_id", ""),
	menu_name=r.get("menu_name", r.get("menu_id", "")),
	menu_path=r.get("menu_path", ""),
	category=r.get("category"),
	similarity=round(r.get("similarity", 0.0), 4),
	similarity_pct=r.get("similarity_pct", f"{r.get('similarity', 0):.1%}"),
	confidence=round(r.get("confidence", 0.0), 4),
	))

	response = SearchResponse(
	query=req.query,
	refined_query=result.get("refined_query", req.query),
	detected_category=result.get("detected_category"),
	intent=result.get("intent", "기타"),
	results=menu_results,
	result_count=len(menu_results),
	quality_score=round(result.get("quality_score", 0.0), 4),
	is_satisfactory=result.get("is_satisfactory", False),
	retry_count=result.get("retry_count", 0),
	rewrite_history=result.get("rewrite_history", []),
	elapsed_ms=elapsed_ms,
	app_version=selected_app_version,
	dataset_version=str(manifest.get("dataset_version", "")),
	index_version=str(manifest.get("index_version", "")),
	applicable=bool(menu_results),
	)

	# 로그 기록
	try:
	_log_query(req, response)
	except Exception as e:
	logger.warning(f"로그 기록 실패: {e}")

	logger.info(
	f"검색 완료 \| query_hash={hashlib.sha256(req.query.encode('utf-8')).hexdigest()[:12]} \| "
	f"결과={len(menu_results)}개 \| 품질={response.quality_score:.2f} \| "
	f"재시도={response.retry_count} \| {elapsed_ms}ms"
	)

	return response


	@app.get(
	"/api/health",
	response_model=HealthResponse,
	summary="헬스체크",
	description="서버 상태 및 기본 정보를 확인합니다.",
	tags=["시스템"],
	)
	async def health(app_version: Optional[str] = None):
	try:
	selected_app_version, manifest = resolve_app_version(app_version)
	engine = _get_engine(selected_app_version)
	stats = engine.vectorstore.get_stats()
	return HealthResponse(
	status="ok",
	menu_count=stats.get("total_menus", 0),
	version=str(manifest.get("index_version", "1.0.0")),
	timestamp=datetime.now().isoformat(),
	)
	except Exception as e:
	raise HTTPException(status_code=503, detail=f"서버 오류: {str(e)}")


	@app.get(
	"/api/categories",
	response_model=CategoriesResponse,
	summary="카테고리 목록",
	description="검색 필터로 사용 가능한 카테고리 목록을 반환합니다.",
	tags=["메타"],
	)
	async def get_categories():
	return CategoriesResponse(categories=CATEGORIES)


	@app.get(
	"/api/stats",
	summary="검색 통계",
	description="누적 검색 쿼리 수 등 운영 통계를 반환합니다.",
	tags=["시스템"],
	)
	async def get_stats():
	total_queries = 0
	if QUERY_LOG_PATH.exists():
	with open(QUERY_LOG_PATH, "r", encoding="utf-8") as f:
	total_queries = sum(1 for _ in f)

	return {
	"total_queries": total_queries,
	"log_path": str(QUERY_LOG_PATH),
	"timestamp": datetime.now().isoformat(),
	}


	# ── 직접 실행 ─────────────────────────────────────────────────────────────────
	if __name__ == "__main__":
	import uvicorn
	uvicorn.run(
	"api.main:app",
	host="0.0.0.0",
	port=8000,
	reload=False,
	log_level="info",
	)