searchHybridAsync.py

import os
import torch
import pandas as pd
import logging
import faiss
import numpy as np
import time
import gensim
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from datasets import load_dataset
from huggingface_hub import login, hf_hub_download, HfApi, create_repo
from keybert import KeyBERT
from sentence_transformers import SentenceTransformer
from joblib import Parallel, delayed
from tqdm import tqdm
import tempfile
import re
import sys
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# ✅ 로그 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ✅ 스레드 풀 설정 (비동기 작업을 위한)
thread_pool = ThreadPoolExecutor(max_workers=os.cpu_count() or 4)

# ✅ FastAPI 인스턴스 생성
app = FastAPI(title="🚀 KeyBERT + Word2Vec 기반 FAISS 검색 API", version="1.2")

# ✅ GPU 사용 여부 확인
device = "cuda" if torch.cuda.is_available() else "cpu"
logger.info(f"🚀 실행 디바이스: {device.upper()}")

# ✅ Hugging Face 로그인
HF_API_TOKEN = os.getenv("HF_API_TOKEN")
if HF_API_TOKEN:
    logger.info("🔑 Hugging Face API 로그인 중...")
    login(token=HF_API_TOKEN)
else:
    logger.error("❌ HF_API_TOKEN이 설정되지 않았습니다. 일부 기능이 제한될 수 있습니다.")

# ✅ Word2Vec 모델 로드
word2vec_model = None
try:
    logger.info("🔄 Word2Vec 모델 로드 중...")
    MODEL_REPO = "aikobay/item-model"
    model_path = hf_hub_download(repo_id=MODEL_REPO, filename="item_vectors.bin", repo_type="dataset")
    word2vec_model = gensim.models.KeyedVectors.load_word2vec_format(model_path, binary=True)
    logger.info(f"✅ Word2Vec 모델 로드 완료! 단어 수: {len(word2vec_model.key_to_index)}")
except Exception as e:
    logger.error(f"❌ Word2Vec 모델 로드 실패: {e}")

# ✅ KeyBERT 모델 로드
logger.info("🔄 KeyBERT 모델 로드 중...")
kw_model = KeyBERT("paraphrase-multilingual-MiniLM-L12-v2")
original_embedding_model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
logger.info("✅ KeyBERT 모델 로드 완료!")

# ✅ 한국어 특화 임베딩 모델로 교체
embedding_model = None
try:
    logger.info("🔄 한국어 특화 임베딩 모델로 교체 시도...")
    # 한국어 특화 모델 로드 시도 (실패시 기존 모델 유지)
    embedding_model = SentenceTransformer("jhgan/ko-sroberta-multitask")
    logger.info("✅ 한국어 특화 임베딩 모델 로드 완료!")
except Exception as e:
    logger.warning(f"⚠️ 한국어 특화 모델 로드 실패, 기존 모델 유지: {e}")
    embedding_model = original_embedding_model

# ✅ 진행 중인 경매 상품 데이터 로드
async def load_huggingface_jsonl(dataset_name, split="train"):
    """Hugging Face Hub에서 데이터셋 비동기 로드"""
    try:
        # 스레드 풀에서 실행하여 비동기 처리
        loop = asyncio.get_event_loop()
        
        def _load_dataset():
            repo_id = f"aikobay/{dataset_name}"
            dataset = load_dataset(repo_id, split=split)
            return dataset.to_pandas().dropna()
        
        # 스레드 풀에서 비동기로 실행
        df = await loop.run_in_executor(thread_pool, _load_dataset)
        return df
    except Exception as e:
        logger.error(f"❌ 데이터 로드 중 오류 발생: {e}")
        return pd.DataFrame()

# 초기 데이터 로드 - 비동기 함수를 동기적으로 호출하여 시작 시 로드
active_sale_items = None
try:
    # 비동기 함수를 시작 시 실행하기 위한 임시 이벤트 루프 사용
    loop = asyncio.new_event_loop()
    active_sale_items = loop.run_until_complete(load_huggingface_jsonl("initial_saleitem_dataset"))
    loop.close()
    
    if active_sale_items.empty:
        logger.error("❌ 데이터셋이 비어 있습니다. 프로그램을 종료합니다.")
        exit(1)
    logger.info(f"✅ 경매 상품 데이터 로드 완료! 총 {len(active_sale_items)}개 상품")
except Exception as e:
    logger.error(f"❌ 상품 데이터 로드 실패: {e}")
    exit(1)

# ✅ FAISS 인덱스 초기화
faiss_index = None
indexed_items = []

# ✅ 멀티코어 벡터화 함수 
async def encode_texts_parallel(texts, batch_size=512):
    """멀티 프로세싱을 활용한 벡터화 속도 최적화 (비동기 지원)"""
    num_cores = os.cpu_count()  # CPU 개수 확인
    logger.info(f"🔄 멀티코어 벡터화 진행 (코어 수: {num_cores})")

    def encode_batch(batch):
        return embedding_model.encode(batch, convert_to_numpy=True)

    # 배치 단위로 병렬 처리
    text_batches = [texts[i:i + batch_size] for i in range(0, len(texts), batch_size)]
    
    # 병렬 처리를 비동기적으로 실행
    loop = asyncio.get_event_loop()
    embeddings = await loop.run_in_executor(
        thread_pool, 
        lambda: Parallel(n_jobs=num_cores)(delayed(encode_batch)(batch) for batch in text_batches)
    )

    return np.vstack(embeddings).astype("float32")

# ✅ FAISS 인덱스 저장 함수 (Hugging Face Hub)
async def save_faiss_index():
    """FAISS 인덱스를 Hugging Face Hub에 저장 (비동기 지원)"""
    global faiss_index, indexed_items
    
    if faiss_index is None or not indexed_items:
        logger.error("❌ 저장할 FAISS 인덱스가 없습니다.")
        return False
    
    try:
        # 레포지토리 ID
        repo_id = os.getenv("HF_INDEX_REPO", "aikobay/saleitem_faiss_index")
        
        # 비동기 작업을 위한 루프
        loop = asyncio.get_event_loop()
        
        # 비동기 작업으로 실행
        def _save_index():
            # HfApi 객체 생성
            api = HfApi()
            
            # 레포지토리 존재 여부 확인 및 생성
            try:
                api.repo_info(repo_id=repo_id, repo_type="dataset")
                logger.info(f"✅ 기존 레포지토리 사용: {repo_id}")
            except Exception:
                logger.info(f"🔄 레포지토리가 존재하지 않아 새로 생성합니다: {repo_id}")
                create_repo(
                    repo_id=repo_id,
                    repo_type="dataset",
                    private=True,
                    exist_ok=True
                )
                logger.info(f"✅ 레포지토리 생성 완료: {repo_id}")
            
            # 임시 파일로 먼저 로컬에 저장
            with tempfile.TemporaryDirectory() as temp_dir:
                index_path = os.path.join(temp_dir, "faiss_index.bin")
                items_path = os.path.join(temp_dir, "indexed_items.txt")
                
                # FAISS 인덱스 저장
                faiss.write_index(faiss_index, index_path)
                
                # 아이템 목록 저장
                with open(items_path, "w", encoding="utf-8") as f:
                    f.write("\n".join(indexed_items))
                
                # README 파일 생성
                readme_path = os.path.join(temp_dir, "README.md")
                with open(readme_path, "w", encoding="utf-8") as f:
                    f.write(f"""# FAISS 인덱스 저장소
이 저장소는 상품 검색을 위한 FAISS 인덱스와 관련 데이터를 포함하고 있습니다.
- 최종 업데이트: {pd.Timestamp.now()}
- 인덱스 항목 수: {len(indexed_items)}
- 모델: KeyBERT + Word2Vec
이 저장소는 'aikobay/initial_saleitem_dataset'의 상품 데이터를 기반으로 생성된 벡터 인덱스를 저장하기 위해 자동 생성되었습니다.
""")
                
                # 파일 업로드
                for file_path, file_name in [
                    (index_path, "faiss_index.bin"),
                    (items_path, "indexed_items.txt"),
                    (readme_path, "README.md")
                ]:
                    api.upload_file(
                        path_or_fileobj=file_path,
                        path_in_repo=file_name,
                        repo_id=repo_id,
                        repo_type="dataset"
                    )
                
                logger.info(f"✅ FAISS 인덱스가 Hugging Face Hub에 저장되었습니다. 레포: {repo_id}")
                return True
            
        # 스레드 풀에서 비동기적으로 실행
        result = await loop.run_in_executor(thread_pool, _save_index)
        return result
            
    except Exception as e:
        logger.error(f"❌ FAISS 인덱스 Hub 저장 중 오류 발생: {e}")
        
        # 로컬에 백업 저장 시도
        try:
            loop = asyncio.get_event_loop()
            
            def _local_backup():
                local_path = os.path.join(os.getcwd(), "faiss_index.bin")
                faiss.write_index(faiss_index, local_path)
                with open("indexed_items.txt", "w", encoding="utf-8") as f:
                    f.write("\n".join(indexed_items))
                logger.info(f"✅ FAISS 인덱스가 로컬에 백업 저장되었습니다: {local_path}")
                return True
            
            result = await loop.run_in_executor(thread_pool, _local_backup)
            return result
        except Exception as local_err:
            logger.error(f"❌ 로컬 백업 저장도 실패: {local_err}")
            return False

# ✅ FAISS 인덱스 로드 함수 (Hugging Face Hub)
async def load_faiss_index():
    """Hugging Face Hub에서 FAISS 인덱스를 로드 (비동기 지원)"""
    global faiss_index, indexed_items
    
    # 레포지토리 ID
    repo_id = os.getenv("HF_INDEX_REPO", "aikobay/saleitem_faiss_index")
    
    try:
        # 비동기 작업을 위한 루프
        loop = asyncio.get_event_loop()
        
        # 비동기 작업으로 실행
        def _load_index():
            # 레포지토리 존재 확인
            api = HfApi()
            try:
                api.repo_info(repo_id=repo_id, repo_type="dataset")
                logger.info(f"✅ FAISS 인덱스 레포지토리 확인: {repo_id}")
            except Exception as repo_err:
                logger.warning(f"⚠️ 레포지토리가 존재하지 않습니다: {repo_err}")
                raise FileNotFoundError("Hub 레포지토리가 존재하지 않습니다")
            
            # Hub에서 파일 다운로드
            index_path = hf_hub_download(
                repo_id=repo_id,
                filename="faiss_index.bin",
                repo_type="dataset"
            )
            
            items_path = hf_hub_download(
                repo_id=repo_id,
                filename="indexed_items.txt",
                repo_type="dataset"
            )
            
            # 파일 로드
            loaded_index = faiss.read_index(index_path)
            with open(items_path, "r", encoding="utf-8") as f:
                loaded_items = f.read().splitlines()
                
            return loaded_index, loaded_items
        
        # 스레드 풀에서 비동기적으로 실행
        loaded_index, loaded_items = await loop.run_in_executor(thread_pool, _load_index)
        
        # 전역 변수에 할당
        faiss_index = loaded_index
        indexed_items = loaded_items
        
        logger.info(f"✅ FAISS 인덱스가 Hub에서 로드되었습니다. 총 {len(indexed_items)}개 상품")
        return True
        
    except Exception as e:
        logger.warning(f"⚠️ Hub에서 FAISS 인덱스 로드 중 오류 발생: {e}")
        
        # 로컬 파일 확인
        try:
            loop = asyncio.get_event_loop()
            
            def _load_local():
                local_index_path = "faiss_index.bin"
                local_items_path = "indexed_items.txt"
                
                if os.path.exists(local_index_path) and os.path.exists(local_items_path):
                    loaded_index = faiss.read_index(local_index_path)
                    with open(local_items_path, "r", encoding="utf-8") as f:
                        loaded_items = f.read().splitlines()
                    return loaded_index, loaded_items
                else:
                    logger.warning("⚠️ 로컬 FAISS 인덱스 파일이 존재하지 않습니다.")
                    return None, None
            
            # 스레드 풀에서 비동기적으로 실행
            result = await loop.run_in_executor(thread_pool, _load_local)
            
            if result[0] is not None:
                faiss_index, indexed_items = result
                logger.info(f"✅ 로컬 FAISS 인덱스 로드 성공. 총 {len(indexed_items)}개 상품")
                return True
            else:
                return False
                
        except Exception as local_err:
            logger.error(f"❌ 로컬 FAISS 인덱스 로드 중 오류: {local_err}")
            return False

# ✅ FAISS 인덱스 구축
async def rebuild_faiss_index():
    """FAISS 인덱스를 새롭게 구축 (비동기 지원)"""
    global faiss_index, indexed_items, active_sale_items

    logger.info("🔄 FAISS 인덱스를 재구축 중...")
    
    # 최신 상품 데이터 로드
    active_sale_items = await load_huggingface_jsonl("initial_saleitem_dataset")
    if active_sale_items.empty:
        logger.error("❌ 상품 데이터를 로드할 수 없습니다.")
        raise RuntimeError("상품 데이터 로드 실패")

    # 상품명 목록 추출
    item_names = active_sale_items["ITEMNAME"].tolist()
    indexed_items = item_names
    logger.info(f"🔹 총 {len(item_names)}개 상품 벡터화 시작...")

    # 벡터화 및 인덱스 구축 - 코사인 유사도 사용
    item_vectors = await encode_texts_parallel(item_names)
    
    # 벡터 정규화 (코사인 유사도를 위해)
    norms = np.linalg.norm(item_vectors, axis=1, keepdims=True)
    normalized_vectors = item_vectors / norms
    
    # Inner Product 기반 인덱스 사용 (코사인 유사도를 위해)
    loop = asyncio.get_event_loop()
    
    def _build_index():
        index = faiss.IndexFlatIP(item_vectors.shape[1])
        index.add(normalized_vectors)
        return index
    
    faiss_index = await loop.run_in_executor(thread_pool, _build_index)

    logger.info(f"✅ FAISS 인덱스 구축 완료! 총 {len(indexed_items)}개 항목.")
    
    # 구축 후 Hub에 저장
    await save_faiss_index()
    return True

# ✅ FAISS 인덱스 상태 확인 및 필요시에만 구축
async def check_faiss_index():
    """FAISS 인덱스가 존재하는지 확인하고 없으면 구축 (비동기 지원)"""
    global faiss_index
    
    if faiss_index is None:
        # Hub에서 로드 시도
        if not await load_faiss_index():
            # 로드 실패 시 새로 구축
            logger.warning("⚠️ 저장된 인덱스가 없어 새로 구축합니다.")
            await rebuild_faiss_index()
    
    # 모든 과정 후에도 인덱스가 None이면 오류
    if faiss_index is None:
        raise RuntimeError("FAISS 인덱스 초기화에 실패했습니다.")

# ✅ KeyBERT 기반 핵심 키워드 추출
async def extract_keywords(query: str, top_n: int = 3):
    """KeyBERT를 사용한 핵심 키워드 추출 (비동기 지원)"""
    loop = asyncio.get_event_loop()
    
    def _extract():
        return kw_model.extract_keywords(query, keyphrase_ngram_range=(1,2), top_n=top_n)
    
    keywords = await loop.run_in_executor(thread_pool, _extract)
    return [k[0] for k in keywords]

# ✅ Word2Vec 기반 키워드 확장 함수
async def expand_keywords_with_word2vec(keywords: list, max_new=5):
    """Word2Vec 모델을 사용한 키워드 확장 (비동기 지원)"""
    if word2vec_model is None:
        logger.warning("⚠️ Word2Vec 모델이 로드되지 않아 확장을 수행하지 않습니다.")
        return keywords
    
    expanded_keywords = list(keywords)  # 복사본 생성
    
    try:
        loop = asyncio.get_event_loop()
        
        def _expand():
            result = list(keywords)
            
            for keyword in keywords:
                # 단어가 모델에 있는지 확인
                if keyword in word2vec_model:
                    # 유사 단어 찾기
                    similar_words = word2vec_model.most_similar(keyword, topn=max_new)
                    result.extend([word for word, _ in similar_words])
                elif len(keyword.split()) > 1:
                    # 복합어인 경우 개별 단어로 시도
                    for word in keyword.split():
                        if word in word2vec_model and len(word) > 1:
                            similar_words = word2vec_model.most_similar(word, topn=2)
                            result.extend([w for w, _ in similar_words])
            
            # 중복 제거
            return list(set(result))
        
        expanded_keywords = await loop.run_in_executor(thread_pool, _expand)
        logger.info(f"🔍 Word2Vec 확장 키워드: {expanded_keywords}")
        return expanded_keywords
    
    except Exception as e:
        logger.error(f"❌ Word2Vec 키워드 확장 중 오류 발생: {e}")
        return keywords

# ✅ FAISS 검색 함수
async def search_faiss_with_keywords(query: str, top_k: int = 5, keywords=None, expanded_keywords=None):
    """키워드 기반 FAISS 검색 수행 (비동기 + 배치 인코딩 적용)"""
    await check_faiss_index()
    start_time = time.time()
    
    # 키워드 추출
    if keywords is None:
        keywords = await extract_keywords(query)
        logger.info(f"🔍 KeyBERT 추출 키워드: {keywords}")
    
    # 키워드 확장
    if expanded_keywords is None:
        expanded_keywords = await expand_keywords_with_word2vec(keywords)
    
    loop = asyncio.get_event_loop()

    # ✅ 원본 쿼리 + 확장 키워드 모두 한 번에 배치 인코딩
    texts_to_encode = [query] + expanded_keywords

    def _encode_batch():
        vectors = embedding_model.encode(texts_to_encode, convert_to_numpy=True)
        norms = np.linalg.norm(vectors, axis=1, keepdims=True)
        return (vectors / norms).astype("float32")

    all_vectors = await loop.run_in_executor(thread_pool, _encode_batch)

    query_vector = np.array([all_vectors[0]])
    keyword_vectors = all_vectors[1:]  # 나머지는 확장 키워드용

    # ✅ 원본 쿼리 FAISS 검색
    def _search_query():
        return faiss_index.search(query_vector, top_k * 2)

    distances, query_indices = await loop.run_in_executor(thread_pool, _search_query)

    scored_results = {}

    for i, dist in zip(query_indices[0], distances[0]):
        if i < len(indexed_items):
            item_name = indexed_items[i]
            scored_results[item_name] = 2.0 * dist  # 쿼리는 가중치 2배

    # ✅ 확장 키워드 벡터들에 대해 배치 검색
    """배치처리 수정
    async def batch_keyword_search(vectors):
        tasks = []

        for vec in vectors:
            keyword_vector = np.array([vec])

            def _search():
                return faiss_index.search(keyword_vector, top_k)

            tasks.append(loop.run_in_executor(thread_pool, _search))

        return await asyncio.gather(*tasks)

    keyword_results = await batch_keyword_search(keyword_vectors)
    """
    def _batch_search_faiss():
        # 빈 배열 체크
        if len(keyword_vectors) == 0:
            return []
            
        # 모든 키워드 벡터를 하나의 배치로 결합 (N개 벡터 x D 차원)
        batch_vectors = np.vstack(keyword_vectors)
        # 한 번에 배치 검색 수행
        distances, indices = faiss_index.search(batch_vectors, top_k)
        # 벡터별 결과 분리하여 반환
        return [(distances[i], indices[i]) for i in range(len(keyword_vectors))]
    
    # 스레드 풀에서 배치 검색 실행
    keyword_results = await loop.run_in_executor(thread_pool, _batch_search_faiss)
    


    # ✅ 점수 누적 처리
    """ 배치처리 수정
    for result in keyword_results:
        k_distances, k_indices = result
        for i, dist in zip(k_indices[0], k_distances[0]):
            if i < len(indexed_items):
                item_name = indexed_items[i]
                if item_name in scored_results:
                    scored_results[item_name] += 0.5 * dist
                else:
                    scored_results[item_name] = 0.5 * dist
    """
    # ✅ 점수 누적 처리
    for k_distances, k_indices in keyword_results:
        for i, dist in zip(k_indices, k_distances):
            if i < len(indexed_items):
                item_name = indexed_items[i]
                if item_name in scored_results:
                    scored_results[item_name] += 0.5 * dist
                else:
                    scored_results[item_name] = 0.5 * dist

    # ✅ 점수 정렬 및 필터링
    sorted_results = sorted(scored_results.items(), key=lambda x: x[1], reverse=True)
    recommendations = []
    min_score_threshold = 0.3

    for item_name, score in sorted_results:
        if score < min_score_threshold:
            continue
        try:
            item_seq = active_sale_items.loc[active_sale_items["ITEMNAME"] == item_name, "ITEMSEQ"].values[0]
            recommendations.append({"ITEMSEQ": item_seq, "ITEMNAME": item_name, "score": float(score)})
        except Exception:
            continue

    # ✅ 직접 매칭 보완
    if len(recommendations) < top_k:
        def _find_direct_matches():
            matches = []
            for item_name in indexed_items:
                if query.lower() in item_name.lower():
                    try:
                        item_seq = active_sale_items.loc[active_sale_items["ITEMNAME"] == item_name, "ITEMSEQ"].values[0]
                        if not any(r["ITEMNAME"] == item_name for r in recommendations):
                            matches.append({"ITEMSEQ": item_seq, "ITEMNAME": item_name, "score": 1.0})
                    except:
                        continue
            return matches

        direct_matches = await loop.run_in_executor(thread_pool, _find_direct_matches)
        recommendations.extend(direct_matches)

    logger.info(f"🔍 검색 완료 | 걸린 시간: {time.time() - start_time:.4f}초 | 결과 수: {len(recommendations)}")
    return recommendations[:top_k]


# ✅ API 요청 모델
class RecommendRequest(BaseModel):
    search_query: str
    top_k: int = 5
    use_expansion: bool = True  # 키워드 확장 사용 여부

# ✅ 추천 API 엔드포인트
@app.post("/api/recommend")
async def recommend(request: RecommendRequest, background_tasks: BackgroundTasks):
    """Word2Vec 기반 FAISS 검색/추천 API (비동기 처리)"""
    try:
        # 로그에 요청 정보 기록
        logger.info(f"📝 검색 요청: '{request.search_query}' (top_k: {request.top_k}, 확장: {request.use_expansion})")
        
        # 키워드 추출 (비동기)
        keywords = await extract_keywords(request.search_query)
        
        # 키워드 확장 사용 여부에 따라 처리 (비동기)
        if request.use_expansion and word2vec_model is not None:
            expanded_keywords = await expand_keywords_with_word2vec(keywords)
        else:
            expanded_keywords = keywords
            logger.info(f"🔍 키워드 확장 없이 진행: {keywords}")
        
        # FAISS 검색 수행 (비동기)
        recommendations = await search_faiss_with_keywords(
            request.search_query, 
            request.top_k,
            keywords,
            expanded_keywords
        )
        
        # 결과 로깅
        if recommendations:
            logger.info(f"🔍 검색 결과: {[r['ITEMNAME'] for r in recommendations]}")
        else:
            logger.warning(f"⚠️ 검색 결과 없음: '{request.search_query}'")
        
        # 백그라운드에서 인덱스 상태 확인 태스크 추가 (사용자 응답은 지연되지 않음)
        background_tasks.add_task(check_index_health)
        
        return {
            "query": request.search_query, 
            "recommendations": recommendations,
            "keywords": keywords,
            "expanded_keywords": expanded_keywords
        }
    except Exception as e:
        logger.error(f"❌ 추천 처리 중 오류: {str(e)}")
        raise HTTPException(status_code=500, detail=f"추천 오류: {str(e)}")

# 인덱스 상태 확인 함수 (백그라운드 태스크용)
async def check_index_health():
    """인덱스 상태를 주기적으로 확인하는 백그라운드 태스크"""
    try:
        # 인덱스 사용 상태 확인
        if faiss_index is None:
            logger.warning("⚠️ 백그라운드 체크: FAISS 인덱스가 로드되지 않았습니다.")
            await check_faiss_index()
        
        # 추가적인 상태 확인 로직을 여기에 구현할 수 있음
        logger.debug("✅ 인덱스 상태 확인 완료")
    except Exception as e:
        logger.error(f"❌ 백그라운드 인덱스 체크 중 오류: {str(e)}")

# ✅ 유사 단어 검색 API
@app.post("/api/similar_words")
async def similar_words(word: str, top_k: int = 10):
    """Word2Vec 모델을 사용한 유사 단어 검색 API (비동기 지원)"""
    try:
        if word2vec_model is None:
            return {"error": "Word2Vec 모델이 로드되지 않았습니다."}
        
        loop = asyncio.get_event_loop()
        
        def _get_similar():
            if word not in word2vec_model:
                return []
            
            similar = word2vec_model.most_similar(word, topn=top_k)
            return [{"word": w, "similarity": float(s)} for w, s in similar]
        
        result = await loop.run_in_executor(thread_pool, _get_similar)
        
        if not result:
            return {"word": word, "similar_words": [], "message": "단어가 모델에 없습니다."}
        
        return {"word": word, "similar_words": result}
    except Exception as e:
        logger.error(f"❌ 유사 단어 검색 중 오류: {str(e)}")
        raise HTTPException(status_code=500, detail=f"유사 단어 검색 오류: {str(e)}")

# ✅ FAISS 인덱스 갱신 API (명시적으로 요청할 때만 실행)
@app.post("/api/update_index")
async def update_index(background_tasks: BackgroundTasks):
    """FAISS 인덱스를 새롭게 구축 (명시적 요청 시에만, 비동기 처리)"""
    try:
        # 인덱스 재구축을 백그라운드 태스크로 실행
        background_tasks.add_task(rebuild_and_log_index)
        return {"message": "✅ FAISS 인덱스 업데이트가 백그라운드에서 시작되었습니다."}
    except Exception as e:
        logger.exception("❌ [API] 인덱스 업데이트 처리 중 예외 발생")
        raise HTTPException(status_code=500, detail=f"인덱스 업데이트 실패: {str(e)}")

# 백그라운드 작업용 인덱스 재구축 함수
async def rebuild_and_log_index():
    """백그라운드에서 인덱스를 재구축하고 결과를 로깅"""
    try:
        logger.info("🔄 백그라운드에서 인덱스 재구축 시작")
        start_time = time.time()
        await rebuild_faiss_index()
        elapsed = time.time() - start_time
        logger.info(f"✅ 백그라운드 인덱스 재구축 완료! 소요 시간: {elapsed:.2f}초")
    except Exception as e:
        logger.error(f"❌ 백그라운드 인덱스 재구축 중 오류: {str(e)}")

# ✅ 인덱스 디버깅 API
@app.get("/api/debug_index")
async def debug_index(query: str, top_k: int = 20):
    """인덱스 디버깅을 위한 API (비동기 지원)"""
    try:
        await check_faiss_index()
        
        loop = asyncio.get_event_loop()
        
        # 원본 벡터 생성 (비동기)
        def _get_vector():
            vector = embedding_model.encode(query, convert_to_numpy=True).astype("float32")
            norm = np.linalg.norm(vector)
            normalized_vector = vector / norm
            return normalized_vector, norm
        
        normalized_vector, norm = await loop.run_in_executor(thread_pool, _get_vector)
        
        # 원본 쿼리로 검색 (비동기)
        def _search():
            return faiss_index.search(np.array([normalized_vector]), top_k)
        
        distances, indices = await loop.run_in_executor(thread_pool, _search)
        
        # 결과 매핑
        results = []
        for i, (idx, dist) in enumerate(zip(indices[0], distances[0])):
            if idx < len(indexed_items):
                item_name = indexed_items[idx]
                results.append({
                    "rank": i + 1,
                    "index": int(idx),
                    "item_name": item_name,
                    "distance/score": float(dist)
                })
        
        # 데이터셋에 해당 단어가 있는지 확인 (비동기)
        def _find_matches():
            contains = [item for item in indexed_items if query.lower() in item.lower()][:5]
            exact = [item for item in indexed_items if query.lower() == item.lower()]
            return contains, exact
        
        contains_query, exact_matches = await loop.run_in_executor(thread_pool, _find_matches)
        
        return {
            "query": query,
            "vector_norm": float(norm),
            "contains_query": contains_query,
            "exact_matches": exact_matches,
            "results": results
        }
    except Exception as e:
        logger.error(f"❌ 인덱스 디버깅 중 오류: {str(e)}")
        raise HTTPException(status_code=500, detail=f"인덱스 디버깅 오류: {str(e)}")

# ✅ 문자열 포함 검색 API
@app.get("/api/text_search")
async def text_search(query: str, top_k: int = 10):
    """단순 텍스트 포함 검색 API (비동기 지원)"""
    try:
        loop = asyncio.get_event_loop()
        
        # 비동기 검색 함수
        def _text_search():
            # 단순 텍스트 포함 검색
            matched_items = []
            for idx, item_name in enumerate(indexed_items):
                if query.lower() in item_name.lower():
                    try:
                        item_seq = active_sale_items.loc[active_sale_items["ITEMNAME"] == item_name, "ITEMSEQ"].values[0]
                        matched_items.append({"ITEMSEQ": item_seq, "ITEMNAME": item_name, "match_type": "contains"})
                    except (IndexError, KeyError):
                        continue

            # 정확히 일치하는 항목을 앞으로
            exact_matches = []
            partial_matches = []
            
            for item in matched_items:
                if query.lower() == item["ITEMNAME"].lower():
                    item["match_type"] = "exact"
                    exact_matches.append(item)
                else:
                    partial_matches.append(item)
            
            # 결합 및 제한
            return exact_matches + partial_matches
        
        # 비동기적으로 검색 실행
        results = await loop.run_in_executor(thread_pool, _text_search)
        
        logger.info(f"🔍 텍스트 검색 결과: {len(results)}개 찾음, 쿼리: '{query}'")
        
        return {
            "query": query,
            "total_matches": len(results),
            "results": results[:top_k]
        }
    except Exception as e:
        logger.error(f"❌ 텍스트 검색 중 오류: {str(e)}")
        raise HTTPException(status_code=500, detail=f"텍스트 검색 오류: {str(e)}")

# ✅ FastAPI 실행
if __name__ == "__main__":
    # 서버 시작 시 저장된 인덱스 로드 시도
    try:
        # 비동기 함수를 동기적으로 호출하기 위한 임시 이벤트 루프 사용
        loop = asyncio.new_event_loop()
        if not loop.run_until_complete(load_faiss_index()):
            logger.warning("⚠️ 기존 인덱스 로드에 실패했습니다. 즉시 새 인덱스를 구축합니다.")
            # 인덱스 즉시 재구축
            loop.run_until_complete(rebuild_faiss_index())
            logger.info("✅ FAISS 인덱스 생성 완료!")
        else:
            logger.info("✅ 기존 인덱스를 성공적으로 로드했습니다.")
        loop.close()
    except Exception as e:
        logger.error(f"❌ 인덱스 초기 구축 실패: {e}")
        logger.warning("⚠️ 인덱스 없이 시작합니다. 검색 기능이 제한될 수 있습니다.")
    
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)