[TRI-187] 배포 25.03.06 10:18

app/__init__.py

@@ -1,30 +0,0 @@
-# app/__init__.py
-from flask import Flask
-from flask_restx import Api
-from app.router.recommendation_endpoint import api as recommendation_ns
-
-def create_app():
-    app = Flask(__name__)
-
-    # 앱 설정, 예: config.py에서 설정값 로드 (필요시)
-    # app.config.from_object('app.config')
-
-    # Swagger UI는 기본적으로 /swagger에서 제공 (doc 인자를 조정 가능)
-    api = Api(
-        app,
-        version="1.0",
-        title="Recommendation API",
-        description="API for restaurant recommendation",
-        doc="/swagger"  # Swagger UI 경로
-    )
-
-    # 추천 API 네임스페이스 등록
-    api.add_namespace(recommendation_ns, path="/recommend")
-
-    # 전역 예외 핸들러 (선택 사항)
-    @app.errorhandler(Exception)
-    def handle_exception(e):
-        app.logger.error(f"Unhandled Exception: {e}", exc_info=True)
-        return {"error": "Internal Server Error", "message": str(e)}, 500
-
-    return app

app/router/recommendation_api.py

@@ -0,0 +1,100 @@
+import os
+import time
+import json
+import logging
+from fastapi import APIRouter, HTTPException
+from app.config import UPLOAD_DIR, FEEDBACK_DIR
+from app.schema.recommendation_schema import UserData, RecommendationItem, CATEGORY_MAPPING
+from app.services.preprocess.data_loader import load_and_merge_json_files
+from app.services.preprocess.preprocessor import preprocess_data
+from app.services.model_trainer import train_model
+from app.services.model_trainer.recommendation import generate_recommendations
+from typing import List, Dict, Any
+
+# 라우터 설정
+logger = logging.getLogger("recommendation_api")
+router = APIRouter()
+
+# 글로벌 변수 초기화
+globals_dict = {}
+
+# 초기 데이터 로딩 및 모델 학습
+def initialize_model():
+    global globals_dict
+    json_directory = str(UPLOAD_DIR)
+
+    try:
+        df_raw = load_and_merge_json_files(json_directory)
+        df_final = preprocess_data(df_raw)  # 병합된 DataFrame 전달
+        globals_dict = train_model(df_final)
+        logger.info("모델 초기화 성공")
+    except Exception as e:
+        logger.error(f"Error during initialization: {e}", exc_info=True)
+        globals_dict = {}
+
+# 서버 시작 시 모델 초기화
+initialize_model()
+
+@router.post("", 
+             response_model=Dict[str, Any],  # 구체적인 응답 모델이 필요하다면 별도로 정의
+             responses={
+                 200: {"description": "Success"},
+                 400: {"description": "Bad Request"},
+                 500: {"description": "Internal Server Error"}
+             })
+
+async def recommend(user_data: UserData):
+    """사용자 데이터를 받아 추천 결과를 생성하고, 결과를 파일로 저장합니다."""
+    try:
+        user_id = user_data.user_id
+        preferred_categories = user_data.preferred_categories
+
+        if not user_id or not preferred_categories:
+            raise HTTPException(status_code=400, detail="user_id와 preferred_categories를 입력해주세요.")
+
+        # 선호 카테고리 이름을 해당 번호로 변환
+        preferred_ids = [CATEGORY_MAPPING.get(cat) for cat in preferred_categories if cat in CATEGORY_MAPPING]
+
+        if not preferred_ids:
+            raise HTTPException(status_code=400, detail="유효한 선호 카테고리를 입력해주세요.")
+
+        df_model = globals_dict.get("df_model")
+        if df_model is None:
+            raise HTTPException(status_code=500, detail="모델 데이터가 초기화되지 않았습니다.")
+
+        filtered_df = df_model[df_model["category_id"].isin(preferred_ids)].copy()
+        if filtered_df.empty:
+            raise HTTPException(status_code=400, detail="해당 선호 카테고리에 해당하는 식당 데이터가 없습니다.")
+
+        # 추천 결과 생성 (generate_recommendations는 JSON 문자열을 반환)
+        result_json = generate_recommendations(
+            filtered_df, 
+            globals_dict["stacking_reg"], 
+            globals_dict["model_features"], 
+            user_id,
+            globals_dict["scaler"]
+        )
+
+        # 추천 결과를 FEEDBACK_DIR에 파일로 저장
+        timestamp = int(time.time())
+        feedback_filename = f"recommendation_{user_id}_{timestamp}.json"
+        feedback_filepath = os.path.join(str(FEEDBACK_DIR), feedback_filename)
+
+        try:
+            os.makedirs(str(FEEDBACK_DIR), exist_ok=True)
+            with open(feedback_filepath, "w", encoding="utf-8") as f:
+                f.write(result_json)
+            logger.info(f"추천 결과가 {feedback_filepath}에 저장되었습니다.")
+        except Exception as file_err:
+            logger.error(f"추천 결과 저장 실패: {file_err}", exc_info=True)
+
+        # JSON 문자열을 Python 객체로 변환하여 반환
+        result_data = json.loads(result_json)
+        return result_data
+
+    except HTTPException:
+        # 이미 생성된 HTTPException은 그대로 다시 발생시킴
+        raise
+    except Exception as e:
+        logger.error(f"Error in recommendation endpoint: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=str(e))

app/schema/recommendation_schema.py

@@ -1,7 +1,7 @@
 # app/models/recommendation_schema.py
 
-from pydantic import BaseModel, conlist
-from typing import List
+from pydantic import BaseModel, Field
+from typing import List, Annotated
 
 class RecommendationItem(BaseModel):
     category_id: int
@@ -28,5 +28,5 @@ class RecommendationItem(BaseModel):
 
 class UserData(BaseModel):
     user_id: str
-    # 선호 카테고리는 최소 1개, 최대 3개
-    preferred_categories: conlist(str, min_items=1, max_items=3)
+    # 선호 카테고리는 최소 1개, 최대 3개 (Pydantic v2 방식)
+    preferred_categories: Annotated[List[str], Field(min_length=1, max_length=3)]

app/services/model_trainer/train_model.py

@@ -5,6 +5,16 @@
 from .model_evaluation import evaluate_model
 import numpy as np
 import logging
+import warnings
+import atexit
+import shutil
+import os
+import tempfile
+from joblib import parallel_backend
+
+# joblib 경고 필터링
+warnings.filterwarnings("ignore", message="resource_tracker")
+warnings.filterwarnings("ignore", message="There appear to be")
 
 logger = logging.getLogger(__name__)