[Feat] 키워드 매핑 기반 추천 개발

.github/workflows/github-action.yml

@@ -24,10 +24,14 @@ jobs:
         username: ${{ secrets.DOCKER_USERNAME }}
         password: ${{ secrets.DOCKER_PASSWORD }}
 
-    # 4. Create .env file from GitHub Secrets
+    # 4-1. Create .env file from GitHub Secrets
     - name: Create .env file
       run: echo "${{ secrets.ENV }}" > ./src/.env
 
+    # 4-2. Create keywords_mapping.py file from GitHub Secrets
+    - name: Create keywords_mapping.py file
+      run: echo "${{ secrets.KEYWORDS_MAPPING }}" > ./src/keywords_mapping.py
+
     # 5. Build and push Docker image
     - name: Build & push Docker image
       uses: docker/build-push-action@v6

.gitignore

@@ -162,3 +162,5 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+src/keywords_mapping.py

src/core/models.py

@@ -10,6 +10,7 @@ class Song(BaseModel):
     tag: str
     lyrics: str
     start_time: str
+    total_time: str
     cover_path: str
     genre: Optional[str]
     youtube_path: str

src/domain/song/schemas.py

@@ -1,5 +1,4 @@
-from pydantic import BaseModel
-from datetime import datetime
+from pydantic import Field, BaseModel
 from typing import List, Optional
 
 # 노래 응답 스키마
@@ -9,7 +8,9 @@ class SongUnitResponse(BaseModel):
     category: str
     lyrics: str
     cover_path: Optional[str] = None
-    youtube_path: str
+    youtube_path: Optional[str] = None
+    recommend_time: str = Field(alias="recommend_time")
+    total_time: str = Field(alias="total_time")
 
     class Config:
         orm_mode = True

src/domain/song/services.py

@@ -16,6 +16,14 @@ async def get_songs_by_category(category: Optional[str], skip: int, limit: int)
     songs = await get_many("song", filter_query, skip, limit)
     return normalize_lyrics(songs)
 
+# 태그별 노래 목록 가져오기
+async def get_songs_by_tag(tag: str, skip: int, limit: int) -> List[dict]:
+    filter_query = {"tag": {"$ne": "SPECIAL"}}
+    if tag:
+        filter_query["tag"] = {"$eq": tag, "$ne": "SPECIAL"}
+    songs = await get_many("song", filter_query, skip, limit)
+    return normalize_lyrics(songs)
+
 # 카테고리에 맞는 총 노래 수를 반환
 async def get_total_songs_count(category: Optional[str]) -> int:
     filter_query = {"category": category} if category else {}
@@ -43,6 +51,17 @@ async def get_random_song_by_category(category: str) -> dict:
         raise CustomException(ErrorCode.SONG_NOT_FOUND)
     return normalize_lyrics(result)[0]
 
+async def get_random_song_by_tag(tag: str) -> dict:
+    result = await db["song"].aggregate([
+        {"$match": {"tag": tag}},
+        {"$sample": {"size": 1}}
+    ]).to_list(length=1)
+    # 결과가 비어 있는 경우
+    if not result:
+        raise CustomException(ErrorCode.SONG_NOT_FOUND)
+    return normalize_lyrics(result)[0]
+
+
 # 공통 메서드: normalize_lyrics (역슬래시 이스케이프 방지)
 def normalize_lyrics(songs: List[dict]) -> List[dict]:
     for song in songs:

src/domain/song/tag.py

@@ -0,0 +1,24 @@
+from enum import Enum
+
+# 태그 enum 정의
+class TagEnum(Enum):
+    SPECIAL = "SPECIAL"
+    ENDURANCE = "ENDURANCE"
+    CONFIDENCE = "CONFIDENCE"
+    BODY_HEALTH = "BODY_HEALTH"
+    MENTAL_HEALTH = "MENTAL_HEALTH"
+    RETRY = "RETRY"
+    YOUTH = "YOUTH"
+    NEW_START = "NEW_START"
+    MONEY = "MONEY"
+    LOTTO = "LOTTO"
+    HOUSE = "HOUSE"
+    LOVE_START = "LOVE_START"
+    LOVE_KEEP = "LOVE_KEEP"
+    HAPPINESS = "HAPPINESS"
+    SUCCESS = "SUCCESS"
+    LUCK = "LUCK"
+
+# 태그 목록 가져오기
+def get_tag_enum_list():
+    return [tag.value for tag in TagEnum]

src/domain/wish/gpt_categorize.py

@@ -5,8 +5,8 @@
 import logging
 import json
 
-from src.domain.song.category import get_category_enum_list
-from src.domain.song.services import get_songs_by_category
+from src.domain.song.tag import get_tag_enum_list
+from src.domain.song.services import get_songs_by_tag
 from src.exceptions.custom_exceptions import CustomException
 from src.exceptions.error_codes import ErrorCode
 
@@ -15,24 +15,24 @@
 openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
 
 async def categorize_wish(content: str) -> dict:
-    # 카테고리 Enum 목록 가져오기
-    category_enum = get_category_enum_list()
+    # 태그 Enum 목록 가져오기
+    tag_enum = get_tag_enum_list()
 
-    # 카테고리별 노래 목록 가져오기
+    # 태그별 노래 목록 가져오기
     total_songs = []
-    for category in category_enum:
-        songs = await get_songs_by_category(category, skip=0, limit=100)
-        category_songs = [{"title": song["title"], "idx": song["song_index"]} for song in songs]
+    for tag in tag_enum:
+        songs = await get_songs_by_tag(tag, skip=0, limit=100)
+        tag_songs = [{"title": song["title"], "idx": song["song_index"]} for song in songs]
         total_songs.append({
-            "category": category,
-            "songs": category_songs
+            "category": tag,
+            "songs": tag_songs
         })
 
     # 프롬프트
     prompt = f'''
     당신은 사용자의 소원을 듣고 그것을 다음 8가지 카테고리 중 하나로 분류한 뒤, 제목을 바탕으로 가장 적합한 노래를 골라줍니다.
     소원의 핵심 내용을 이해하고 단 한 곡을 선택해야 합니다.
-    * category 값은 다음 중 하나입니다: {", ".join(category_enum)}
+    * category 값은 다음 중 하나입니다: {", ".join(tag_enum)}
 
     * OUTPUT 형태 및 자료형:
     {{
@@ -65,7 +65,6 @@ async def categorize_wish(content: str) -> dict:
         response_json = json.loads(message_content)
 
         # Extracting required data
-        # category = response_json.get("category")
         song_index = response_json.get("idx")
 
         # Validating the response
@@ -74,7 +73,6 @@ async def categorize_wish(content: str) -> dict:
 
         # Returning the result
         return {
-            # "category": category,
             "song_index": song_index
         }
 
@@ -86,4 +84,4 @@ async def categorize_wish(content: str) -> dict:
     except Exception as e:
         error_message = f"Exception occurred: {e}\n{traceback.format_exc()}"
         logging.error(error_message)
-        raise CustomException(ErrorCode.EXTERNAL_SERVICE_ERROR)
+        raise CustomException(ErrorCode.EXTERNAL_SERVICE_ERROR)

src/domain/wish/schemas.py

@@ -1,6 +1,4 @@
 from pydantic import BaseModel, Field
-from typing import Optional
-from bson import ObjectId
 from datetime import datetime
 
 # 소원 생성 요청 스키마
@@ -9,7 +7,7 @@ class WishCreate(BaseModel):
     content: str
     is_displayed: bool = Field(alias="is_displayed")
 
-# 랜덤4개 소원 응답 스키마
+# 랜덤 4개 소원 응답 스키마
 class WishRandomResponse(BaseModel):
     nickname: str
     content: str

src/domain/wish/services.py

@@ -1,60 +1,86 @@
 from src.core.database import db
-from src.core.crud import insert, get_by_id, get_many
+from src.core.crud import insert, get_by_id
 from datetime import datetime
 from typing import List
 from src.exceptions.custom_exceptions import CustomException
 from src.exceptions.error_codes import ErrorCode
 
 from src.domain.wish.gpt_categorize import categorize_wish
-from src.domain.song.services import get_song_by_song_index
+from src.domain.song.services import get_song_by_song_index, get_random_song_by_tag
 
-# 소원의 카테고리 분류 및 노래 추천
+from src.keywords_mapping import keyword_to_tag
+
+# 소원 처리 메인 함수
 async def process_wish(wish):
+    validate_wish(wish)
+
+    existing_wish = await find_existing_wish(wish)
+    if existing_wish:
+        return await handle_existing_wish(existing_wish)
+
+    # 키워드 기반 태그 검색 및 추천
+    keyword_recommendation = await recommend_by_keyword(wish.content)
+    if keyword_recommendation:
+        return keyword_recommendation
+
+    # AI로 태그+제목 기반 추천
+    return await categorize_and_recommend(wish)
+
+## 유효성 검사 함수
+def validate_wish(wish):
     if not wish.content:
         raise CustomException(ErrorCode.MISSING_PARAMETER)
-
-    # 양옆 공백 제거
     wish.content = wish.content.strip()
     wish.nickname = wish.nickname.strip()
-
-    # 이미 동일한 내용의 소원이 존재하는지 확인
-    existing_wish = await db["wish"].find_one({
+
+## 기존 소원 찾기
+async def find_existing_wish(wish):
+    return await db["wish"].find_one({
         "nickname": wish.nickname,
         "content": wish.content
     })
+
+## 기존 소원 처리
+async def handle_existing_wish(existing_wish):
+    existing_wish_song = await db["song"].find_one({
+        "_id": existing_wish["song_id"]
+    })
+    song_id = str(existing_wish["_id"])
+    return {
+        "wish_id": song_id,
+        "nickname": existing_wish["nickname"],
+        "wish": existing_wish["content"],
+        "category": existing_wish_song["category"],
+        "recommended_song": format_song_data(existing_wish_song),
+        "wishes_count": await count_wishes_by_song_id(song_id)
+    }
+
+## 키워드 기반 추천
+async def recommend_by_keyword(content):
+    for keyword, tag in keyword_to_tag.items():
+        if keyword in content:
+            recommended_song = await get_random_song_by_tag(tag)
+            if recommended_song:
+                return {
+                    "wish_id": None,
+                    "nickname": None,
+                    "wish": content,
+                    "category": recommended_song["category"],
+                    "recommended_song": format_song_data(recommended_song),
+                    "wishes_count": await count_wishes_by_song_id(recommended_song["_id"])
+                }
+    return None
+
+## 태그 분류 및 추천
+async def categorize_and_recommend(wish):
+    category_data = await categorize_wish(wish.content)
 
-    if existing_wish:
-        existing_wish_song = await db["song"].find_one({
-            "_id": existing_wish["song_id"]
-        })
-        song_id = str(existing_wish["_id"])
-        return {
-            "wish_id": song_id,
-            "nickname": existing_wish["nickname"],
-            "wish": existing_wish["content"],
-            "category": existing_wish_song["category"],  # category는 기존 소원에서 가져와야 할 수도 있음
-            "recommended_song": {
-                "title": existing_wish_song["title"],
-                "artist": existing_wish_song["artist"],
-                "lyrics": existing_wish_song["lyrics"],
-                "cover_path": existing_wish_song["cover_path"],
-                "recommend_time": existing_wish_song["start_time"],
-                "youtube_path": existing_wish_song["youtube_path"],
-            },
-            "wishes_count": await count_wishes_by_song_id(song_id)
-        }
-
-
-    # 카테고리 분류 및 추천 노래 가져오기
-    category_data = await categorize_wish(wish.content)  # {"category": "...", "song_index": ...}
     recommended_song = await get_song_by_song_index(category_data["song_index"])
-    
+
     # 노래 추천 시점 계산
-    start_time = datetime.strptime(recommended_song["start_time"], "%H:%M:%S")
-    midnight = datetime.strptime("00:00:00", "%H:%M:%S")
-    recommend_time = midnight - start_time
+    recommend_time = calculate_recommend_time(recommended_song["start_time"])
 
-    # 소원 데이터 생성 및 반환
+    # 소원 데이터 생성
     created_wish = await create_wish(
         nickname=wish.nickname,
         content=wish.content,
@@ -69,16 +95,33 @@ async def process_wish(wish):
         "wish": created_wish["content"],
         "category": recommended_song["category"],
         "recommended_song": {
-            "title": recommended_song["title"],
-            "artist": recommended_song["artist"],
-            "lyrics": recommended_song["lyrics"],
-            "cover_path": recommended_song["cover_path"],
+            **format_song_data(recommended_song),
             "recommend_time": str(recommend_time),
-            "youtube_path": recommended_song["youtube_path"],
         },
-        "wishes_count": count_wishes_by_song_id(song_id)
+        "wishes_count": await count_wishes_by_song_id(song_id)
+    }
+
+## 노래 데이터 포맷팅
+def format_song_data(song):
+    if "lyrics" in song and isinstance(song["lyrics"], str):
+            song["lyrics"] = song["lyrics"].replace("\\n", "\n")
+    return {
+        "title": song["title"],
+        "artist": song["artist"],
+        "lyrics": song["lyrics"],
+        "cover_path": song["cover_path"],
+        "recommend_time": song["start_time"],
+        "youtube_path": song["youtube_path"]
     }
 
+
+## 추천 시간 계산
+def calculate_recommend_time(start_time):
+    start_time_dt = datetime.strptime(start_time, "%H:%M:%S")
+    midnight = datetime.strptime("00:00:00", "%H:%M:%S")
+    return midnight - start_time_dt
+
+
 # 소원 생성
 async def create_wish(nickname: str, content: str, song_id: str, is_displayed: bool) -> str:
     wish_data = {
@@ -92,23 +135,28 @@ async def create_wish(nickname: str, content: str, song_id: str, is_displayed: b
     created_wish = await db["wish"].find_one({"_id": result.inserted_id})
     return created_wish
 
-# 특정 소원 가져오기
-async def get_wish_by_id(_id: str) -> dict:
-    return await get_by_id("wish", _id)
 
 # 랜덤 소원 가져오기
 async def get_random_wishes(limit: int = 4) -> List[dict]:
-    # MongoDB의 $sample 사용 (별도 crud 함수 필요할 수도 있음)
     wishes = await db["wish"].aggregate([{"$sample": {"size": limit}}]).to_list(length=limit)
     for wish in wishes:
         song = await get_by_id("song", wish["song_id"])
         if song:
-            wish["category"] = song.get("category", "unknown")
+            category = song.get("category")
+            if category is None:
+                wish["category"] = "WEALTH" # 에러를 발생시키는 대신 임의의 카테고리 값을 삽입
+            wish["category"] = category
         else:
-            raise CustomException(ErrorCode.SONG_CATEGORY_IS_MISSING)
+            wish["category"] = "HAPPINESS" # 에러를 발생시키는 대신 임의의 카테고리 값을 삽입
     return wishes
 
+
+# 특정 소원 가져오기
+async def get_wish_by_id(_id: str) -> dict:
+    return await get_by_id("wish", _id)
+
+
 # 특정 노래의 wish 갯수 세기
 async def count_wishes_by_song_id(_id: str) -> int:
-    wishes = await get_many("wish", {"song_id": _id}, 0, 0)  # 모든 wish를 가져옴
-    return len(wishes)
+    count = await db.wish.count_documents({"song_id": _id})
+    return count