Merge pull request #11 from MapChillE/fix/UBLE-80-elasticsearch

[Fix] elasticsearch 연동 및 location 추가

Author: hayong39

app/api/recommend.py

@@ -1,16 +1,20 @@
 from fastapi import APIRouter, Depends, HTTPException, Query
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, joinedload
 from sentence_transformers import SentenceTransformer
-from sqlalchemy import text
+from sqlalchemy import text, func
 from app.database.connection import get_db
-from app.models import Store
+from app.models import Store, Brand
 from app.services.recommend_service import HybridRecommender
 from geoalchemy2.functions import ST_DWithin, ST_SetSRID, ST_MakePoint, ST_Distance
+from geoalchemy2 import Geometry
+from geoalchemy2.shape import to_shape
 import logging
 from app.services.collect_user_data import collect_user_data
 from collections import defaultdict
 from app.database.redis_client import r
 import json
+from app.database.es import es
+
 
 router = APIRouter()
 model = SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
@@ -23,6 +27,12 @@ def get_min_rank(benefits: list) -> str:
             return b.rank
     return "NONE"
 
+def extract_lat_lng(store):
+    if store.location is None:
+        return None, None
+    point = to_shape(store.location)
+    return point.y, point.x
+
 @router.get("/recommend")
 def recommend(
     user_id:int, 
@@ -33,7 +43,7 @@ def recommend(
     ):
 
     # 1. 사용자 정보 수집
-    categories, histories, bookmarks, clicks, searches = collect_user_data(user_id, db)
+    categories, histories, bookmarks, clicks, searches = collect_user_data(user_id, db, es)
 
     if not (categories or histories or bookmarks or clicks or searches):
         raise HTTPException(status_code=404, detail="사용자 정보가 부족합니다.")
@@ -111,7 +121,7 @@ def hybrid_recommend(
         logger.error(f"Redis 캐시 확인 중 오류: {e}")
 
     # 1. 사용자 텍스트 정보 수집
-    categories, histories, bookmarks, clicks, searches = collect_user_data(user_id, db)
+    categories, histories, bookmarks, clicks, searches = collect_user_data(user_id, db, es)
 
     if not (categories or histories or bookmarks or clicks or searches):
         raise HTTPException(status_code=404, detail="사용자 정보가 부족합니다.")
@@ -126,17 +136,20 @@ def hybrid_recommend(
     logger.debug(f"Recommendation results for user {user_id}: {results}")
 
     # 4. 위치 기반 필터링: 추천 브랜드 매장 중 반경 km 이내
-    nearby_stores = db.query(Store).filter(
-    Store.brand_id.in_(recommended_brand_ids),
-    ST_DWithin(Store.location, ST_SetSRID(ST_MakePoint(lng, lat), 4326), radius_km * 1000)
+    store_query = db.query(Store).options(
+        joinedload(Store.brand).joinedload(Brand.category),
+        joinedload(Store.brand).joinedload(Brand.benefits)
+    ).filter(
+        Store.brand_id.in_(recommended_brand_ids),
+        func.ST_DWithin(Store.location, func.ST_SetSRID(func.ST_MakePoint(lng, lat), 4326), radius_km * 1000)
     ).order_by(
-        ST_Distance(Store.location, ST_SetSRID(ST_MakePoint(lng, lat), 4326))
+        func.ST_Distance(Store.location, func.ST_SetSRID(func.ST_MakePoint(lng, lat), 4326))
     ).all()
 
     # 매장 중 하나씩 결과 연결
     store_map = {}
     brand_store_map = defaultdict(list)
-    for store in nearby_stores:
+    for store in store_query:
         brand_store_map[store.brand_id].append(store)
     store_map = {bid: stores[0] for bid, stores in brand_store_map.items()}
 
@@ -147,10 +160,13 @@ def hybrid_recommend(
             continue
         store = store_map[brand_id]
         brand = store.brand
+        lat, lng = extract_lat_lng(store)
 
         item = {
             "storeId": store.id,
             "storeName": store.name,
+            "latitude": lat,
+            "longitude": lng,
             "category": brand.category.name if brand.category else None,
             "description": brand.description,
             "isVIPcock": brand.rank_type in ("VIP", "VIP_NORMAL"),

app/database/es.py

@@ -0,0 +1,30 @@
+from elasticsearch import Elasticsearch
+from dotenv import load_dotenv
+import os
+import logging
+
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+
+ELASTICSEARCH_URL = os.getenv("ELASTICSEARCH_URL")
+ES_ID = os.getenv("ES_ID")
+ES_PW = os.getenv("ES_PW")
+
+if not all([ELASTICSEARCH_URL, ES_ID, ES_PW]):
+        raise ValueError("필수 Elasticsearch 환경 변수가 설정되지 않았습니다.")
+
+es = Elasticsearch(
+            ELASTICSEARCH_URL, 
+            basic_auth=(ES_ID, ES_PW),
+            verify_certs=True
+    )   
+
+#연결 확인
+try:
+        if not es.ping():
+                raise ConnectionError("Elasticsearch 서버에 연결할 수 없습니다.")
+        logger.info("Elasticsearch 연결 성공")
+except Exception as e:
+        logger.error(f"Elasticsearch 연결 실패: {e}")
+        raise

app/services/collect_user_data.py

@@ -1,8 +1,14 @@
 from sqlalchemy.orm import Session
 from sqlalchemy import text
+from elasticsearch import Elasticsearch
+from elasticsearch.helpers import scan
+import logging
 
-def collect_user_data(user_id: int, db: Session) -> tuple[list, list, list, list, list]:
-    """Collect user-related text data from various sources."""
+logger = logging.getLogger(__name__)
+
+def collect_user_data(user_id: int, db: Session, es: Elasticsearch) -> tuple[list, list, list, list, list]:
+    
+    #RDB에서 가져오는 데이터
     categories = db.execute(text("""
         SELECT c.name FROM user_category uc
         JOIN category c ON uc.category_id = c.id
@@ -21,14 +27,35 @@ def collect_user_data(user_id: int, db: Session) -> tuple[list, list, list, list
         WHERE bm.user_id = :user_id
     """), {"user_id": user_id}).scalars().all()
 
-    clicks = db.execute(text("""
-        SELECT s.name FROM store_click_log cl
-        JOIN store s ON cl.store_id = s.id
-        WHERE cl.user_id = :user_id
-    """), {"user_id": user_id}).scalars().all()
+    # es에서 가져오는 클릭 로그
+    clicks = []
+    try:
+        for doc in scan(es, index="store-click-log", query={
+            "query": {
+                "term": {
+                    "userId": user_id
+                }
+            }
+        }):
+            store_name = doc["_source"].get("storeName")
+            if store_name:
+                clicks.append(store_name)
+    except Exception as e:
+        logger.error(f"클릭 로그 조회 실패 (user_id: {user_id}: {e})")
 
-    searches = db.execute(text("""
-        SELECT keyword FROM search_log WHERE user_id = :user_id
-    """), {"user_id": user_id}).scalars().all()
-    
+    # es에서 가져오는 검색 로그
+    searches = []
+    try:
+        for doc in scan(es, index="search-log", query={
+            "query": {
+                "term": {
+                    "userId": user_id
+                }
+            }       
+        }):
+            keyword = doc["_source"].get("searchKeyword")
+            if keyword:
+                searches.append(keyword)
+    except Exception as e:
+        logger.error(f"검색 로그 조회 실패 (user_id: {user_id}: {e})")
     return categories, histories, bookmarks, clicks, searches

app/services/recommend_service.py

@@ -4,6 +4,13 @@
 from implicit.als import AlternatingLeastSquares
 from sqlalchemy.orm import Session
 from app.models import BrandClickLog, StoreClickLog, BrandEmbedding, Store
+from elasticsearch import Elasticsearch
+from elasticsearch.helpers import scan
+from app.database.es import es
+import logging
+
+logger = logging.getLogger(__name__)
+
 
 class HybridRecommender:
     def __init__(self):
@@ -12,15 +19,28 @@ def __init__(self):
         self.item_factors = None
         self.item_id_to_index = {}
         self.index_to_item_id = {}
+        self.user_id_to_code = {}
+        self.code_to_user_id = {}
+        self.es = es
+
+    # 로그 가져오는 함수
+    def get_logs_from_es(self, index_name: str):
+        logs = []
+        try:
+            for doc in scan(self.es, index=index_name):
+                logs.append(doc["_source"])
+        except Exception as e:
+            logger.error(f"Elasticsearch 로그 조회 실패 (index: {index_name}): {e}")
+        return logs
 
     def train_model(self, db: Session):
-        store_logs = db.query(StoreClickLog).all()
-        brand_logs = db.query(BrandClickLog).all()
+        store_logs = self.get_logs_from_es("store-click-log")
+        brand_logs = self.get_logs_from_es("brand-click-log")
 
         combined_data = []
 
-        # Store 클릭 로그 : brand_id로 매핑
-        store_ids = [log.store_id for log in store_logs]
+        # Store 클릭 로그 : store_id -> brand_id로 매핑
+        store_ids = [log["storeId"] for log in store_logs]
         stores_with_brands = db.query(Store.id, Store.brand_id).filter(
             Store.id.in_(store_ids),
             Store.brand_id.isnot(None)
@@ -29,13 +49,13 @@ def train_model(self, db: Session):
         store_to_brand = {store_id: brand_id for store_id, brand_id in stores_with_brands}
 
         for log in store_logs:
-            brand_id = store_to_brand.get(log.store_id)
+            brand_id = store_to_brand.get(log["storeId"])
             if brand_id:
-                combined_data.append({'user_id': log.user_id, 'brand_id': brand_id})
+                combined_data.append({'user_id': log["userId"], "brand_id": brand_id})
 
         # Brand 클릭 로그
         for log in brand_logs:
-            combined_data.append({'user_id': log.user_id, 'brand_id': log.brand_id}) 
+            combined_data.append({'user_id': log["userId"], "brand_id": log["brandId"]})
 
         if not combined_data:
             return

requirements.txt

@@ -1,21 +1,35 @@
 annotated-types==0.7.0
 anyio==4.9.0
+certifi==2025.7.9
+charset-normalizer==3.4.2
+click==8.2.1
+elastic-transport==8.17.1
+elasticsearch==8.12.1
 fastapi==0.116.1
+filelock==3.18.0
+fsspec==2025.5.1
 GeoAlchemy2==0.17.1
 h11==0.16.0
+hf-xet==1.1.5
+huggingface-hub==0.33.4
 idna==3.10
 implicit==0.7.2
+Jinja2==3.1.6
 joblib==1.5.1
+MarkupSafe==3.0.2
 mpmath==1.3.0
+networkx==3.5
 numpy==2.3.1
+packaging==25.0
 pandas==2.3.1
 pgvector==0.4.1
 pillow==11.3.0
-psycopg2-binary==2.9.10
+psycopg2==2.9.10
 pydantic==2.11.7
 pydantic_core==2.33.2
 python-dateutil==2.9.0.post0
 python-dotenv==1.1.1
+pytz==2025.2
 PyYAML==6.0.2
 redis==6.2.0
 regex==2024.11.6
@@ -24,6 +38,9 @@ safetensors==0.5.3
 scikit-learn==1.7.0
 scipy==1.16.0
 sentence-transformers==5.0.0
+setuptools==80.9.0
+shapely==2.1.1
+six==1.17.0
 sniffio==1.3.1
 SQLAlchemy==2.0.41
 starlette==0.47.1
@@ -35,4 +52,6 @@ tqdm==4.67.1
 transformers==4.53.2
 typing-inspection==0.4.1
 typing_extensions==4.14.1
+tzdata==2025.2
+urllib3==2.5.0
 uvicorn==0.35.0