[NL-59] 로또 번호 추천 테이블 생성 및 프롬프트 작성, 번호 추천/조회 API 구현 (#9)

* fix: github action deploy image prune 명령 추가

* fix: 스크립트 실행 오류 수정

* feat: 로또 번호 추천 프롬프트 추가

* fix: 사주명식 - 가장 강한/약한 오행 계산 및 저장 추가

* fix: parser 오류 수정

* fix: 프롬프트 수정

* feat: 사용자별 로또 번호 추천/조회 API 생성

* feat: alembic migration

* fix: typeddict -> base model

* Update src/lotto/service.py

Co-authored-by: Copilot <[email protected]>

* Update src/lotto/service.py

Co-authored-by: Copilot <[email protected]>

* feat: 강한/상충되는 기운 필드 추가

---------

Co-authored-by: Gukhee Jo <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

.github/workflows/deploy.yml

@@ -50,3 +50,6 @@ jobs:
 
             # 상태 확인
             docker compose -f docker-compose.dev.yml ps fastapi
+            
+            # 사용되지 않는 이미지 정리
+            docker system prune

docker-compose.dev.yml

@@ -20,9 +20,9 @@ services:
     image: satto-registry.kr.ncr.ntruss.com/satto-server-fastapi:latest
     pull_policy: always
     container_name: fastapi-app
-    depends_on:
-      mysql:
-        condition: service_healthy
+#    depends_on:
+#      mysql:
+#        condition: service_healthy
     volumes:
       - ${WRITABLE_DIR}/logs/app:/data/logs/app
     command: >

migrations/scripts/insert_lotto_data.py

@@ -15,6 +15,11 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+# 프로젝트 루트 Python 경로에 추가
+project_root = Path(__file__).parent.parent.parent
+if str(project_root) not in sys.path:
+    sys.path.insert(0, str(project_root))
+
 # .env 파일 로드
 from dotenv import load_dotenv
 load_dotenv()
@@ -199,4 +204,4 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main()) 
+    asyncio.run(main())

migrations/versions/de9ee05d86b4_add_lotto_recommendations.py

@@ -0,0 +1,41 @@
+"""add lotto_recommendations
+
+Revision ID: de9ee05d86b4
+Revises: 05818a33a6e5
+Create Date: 2025-08-10 00:59:05.406541
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'de9ee05d86b4'
+down_revision: Union[str, Sequence[str], None] = '05818a33a6e5'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('lotto_recommendations',
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('user_id', sa.String(length=255), nullable=False),
+    sa.Column('round', sa.Integer(), nullable=False),
+    sa.Column('content', sa.JSON(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), nullable=False),
+    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('lotto_recommendations')
+    # ### end Alembic commands ###

src/four_pillars/common/calculator.py

@@ -1,8 +1,10 @@
 from datetime import datetime, date
 from pathlib import Path
 from typing import List, Tuple, Dict
+from collections import Counter
 
 from src.four_pillars.entities.schemas import FourPillar
+from src.four_pillars.entities.enums import FiveElements
 
 
 class FourPillarsCalculator:
@@ -14,6 +16,39 @@ class FourPillarsCalculator:
     def __init__(self):
         self._init_kanshi_data()
         self._init_setsuiri_data()
+        self._init_five_elements_mapping()
+
+    def _init_five_elements_mapping(self):
+        """천간과 지지의 오행 매핑 초기화"""
+        # 천간 오행 매핑
+        self.jikkan_five_elements = {
+            "甲": FiveElements.WOOD,   # 갑 - 목
+            "乙": FiveElements.WOOD,   # 을 - 목
+            "丙": FiveElements.FIRE,   # 병 - 화
+            "丁": FiveElements.FIRE,   # 정 - 화
+            "戊": FiveElements.EARTH,  # 무 - 토
+            "己": FiveElements.EARTH,  # 기 - 토
+            "庚": FiveElements.METAL,  # 경 - 금
+            "辛": FiveElements.METAL,  # 신 - 금
+            "壬": FiveElements.WATER,  # 임 - 수
+            "癸": FiveElements.WATER,  # 계 - 수
+        }
+        
+        # 지지 오행 매핑
+        self.jyunishi_five_elements = {
+            "寅": FiveElements.WOOD,   # 인 - 목
+            "卯": FiveElements.WOOD,   # 묘 - 목
+            "巳": FiveElements.FIRE,   # 사 - 화
+            "午": FiveElements.FIRE,   # 오 - 화
+            "辰": FiveElements.EARTH,  # 진 - 토
+            "戌": FiveElements.EARTH,  # 술 - 토
+            "丑": FiveElements.EARTH,  # 축 - 토
+            "未": FiveElements.EARTH,  # 미 - 토
+            "申": FiveElements.METAL,  # 신 - 금
+            "酉": FiveElements.METAL,  # 유 - 금
+            "亥": FiveElements.WATER,  # 해 - 수
+            "子": FiveElements.WATER,  # 자 - 수
+        }
 
     def _init_kanshi_data(self):
         """60간지 배열과 해시 초기화"""
@@ -115,6 +150,40 @@ def _calculate_kanshi(
 
         return [year_pillar, month_pillar, day_pillar, time_pillar]
 
+    def _analyze_five_elements(self, pillars: List[str]) -> Tuple[List[FiveElements], List[FiveElements]]:
+        """사주팔자의 오행 분석"""
+        element_counts = Counter()
+        
+        # 모든 오행을 0으로 초기화
+        all_elements = [FiveElements.WOOD, FiveElements.FIRE, FiveElements.EARTH, FiveElements.METAL, FiveElements.WATER]
+        for element in all_elements:
+            element_counts[element] = 0
+        
+        for pillar in pillars:
+            if pillar is None:
+                continue
+                
+            # 천간과 지지 분리
+            jikkan = pillar[0]  # 첫 번째 글자는 천간
+            jyunishi = pillar[1]  # 두 번째 글자는 지지
+            
+            # 천간 오행 추가
+            if jikkan in self.jikkan_five_elements:
+                element_counts[self.jikkan_five_elements[jikkan]] += 1
+            
+            # 지지 오행 추가
+            if jyunishi in self.jyunishi_five_elements:
+                element_counts[self.jyunishi_five_elements[jyunishi]] += 1
+        
+        # 가장 많은 오행과 가장 적은 오행 찾기
+        max_count = max(element_counts.values())
+        min_count = min(element_counts.values())
+        
+        strong_elements = [element for element, count in element_counts.items() if count == max_count]
+        weak_elements = [element for element, count in element_counts.items() if count == min_count]
+        
+        return strong_elements, weak_elements
+
     def calculate_four_pillars(self, birth_date: datetime) -> FourPillar:
         """사주 계산 메인 함수"""
         year = birth_date.year
@@ -129,13 +198,21 @@ def calculate_four_pillars(self, birth_date: datetime) -> FourPillar:
             minute = None
 
         pillars = self._calculate_kanshi(year, month, day, hour, minute)
+        
+        # 오행 분석
+        strong_elements, weak_elements = self._analyze_five_elements(pillars)
+        
+        # FiveElements enum을 문자열로 변환
+        strong_elements_str = [element.value for element in strong_elements] if strong_elements else None
+        weak_elements_str = [element.value for element in weak_elements] if weak_elements else None
+        
         result: FourPillar = {
             "year_pillar": pillars[0],  # 년주
             "month_pillar": pillars[1],  # 월주
             "day_pillar": pillars[2],  # 일주
+            "time_pillar": pillars[3], # 시주
+            "strong_elements": strong_elements_str,
+            "weak_elements": weak_elements_str
         }
 
-        if pillars[3] is not None:
-            result["time_pillar"] = pillars[3]
-
         return result

src/four_pillars/entities/enums.py

@@ -0,0 +1,10 @@
+from enum import Enum
+
+
+class FiveElements(str, Enum):
+    """오행 (五行)"""
+    WOOD = "목(木)"  # 목
+    FIRE = "화(火)"  # 화
+    EARTH = "토(土)"  # 토
+    METAL = "금(金)"  # 금
+    WATER = "수(水)"  # 수

src/four_pillars/entities/schemas.py

@@ -1,10 +1,11 @@
-from typing import Optional
-
+from typing import Optional, List
 from typing_extensions import TypedDict
 
 
-class FourPillar(TypedDict):
-    year_pillar: str
-    month_pillar: str
-    day_pillar: str
-    time_pillar: Optional[str] = None
+class FourPillar(TypedDict, total=False):
+    year_pillar: str  # 년주
+    month_pillar: str  # 월주
+    day_pillar: str  # 일주
+    time_pillar: Optional[str]  # 시주
+    strong_elements: Optional[List[str]]  # 가장 많은 오행
+    weak_elements: Optional[List[str]]  # 가장 적은 오행

src/hcx_client/common/parser.py

@@ -1,33 +1,40 @@
 import json
 import re
 
+from src.common.logger import logger
+
 
 class Parser:
     @staticmethod
-    def parse_four_pillar(content: str):
-        # 마크다운 JSON 블록에서 JSON 추출
-        json_match = re.search(r"```json\s*(\{.*?\})\s*```", content, re.DOTALL)
-        if json_match:
-            try:
-                json_str = json_match.group(1)
-                json_data = json.loads(json_str)
-                return json_data
-            except json.JSONDecodeError as e:
-                raise ValueError(f"JSON 파싱 오류: {e}")
-            except Exception as e:
-                raise ValueError(f"모델 검증 오류: {e}")
+    def parse_json(content: str):
+        try:
+            # 마크다운 JSON 블록에서 JSON 추출
+            json_match = re.search(r"```json\s*(\{.*?\})\s*```", content, re.DOTALL)
+            if json_match:
+                try:
+                    json_str = json_match.group(1)
+                    json_data = json.loads(json_str)
+                    return json_data
+                except json.JSONDecodeError as e:
+                    raise ValueError(f"JSON 파싱 오류: {e}")
+                except Exception as e:
+                    raise ValueError(f"모델 검증 오류: {e}")
+
+            # 일반 코드 블록에서 JSON 추출 (```json이 아닌 경우)
+            code_match = re.search(r"```\s*(\{.*?\})\s*```", content, re.DOTALL)
+            if code_match:
+                try:
+                    json_str = code_match.group(1)
+                    json_data = json.loads(json_str)
+                    return json_data
+                except json.JSONDecodeError as e:
+                    raise ValueError(f"JSON 파싱 오류: {e}")
+                except Exception as e:
+                    raise ValueError(f"모델 검증 오류: {e}")
+
+            return json.loads(content)
 
-        # 일반 코드 블록에서 JSON 추출 (```json이 아닌 경우)
-        code_match = re.search(r"```\s*(\{.*?\})\s*```", content, re.DOTALL)
-        if code_match:
-            try:
-                json_str = code_match.group(1)
-                json_data = json.loads(json_str)
-                return json_data
-            except json.JSONDecodeError as e:
-                raise ValueError(f"JSON 파싱 오류: {e}")
-            except Exception as e:
-                raise ValueError(f"모델 검증 오류: {e}")
+        except Exception as e:
+            logger.info(f"[Parser] JSON 파싱 실패: {str(e)}")
+            raise ValueError(f"마크다운 파싱 오류: {e}")
 
-        # JSON 블록을 찾지 못한 경우
-        raise ValueError("마크다운 JSON 블록을 찾을 수 없습니다.")

src/hcx_client/prompts/fortune.yaml

@@ -32,3 +32,52 @@ four_pillar:
     성별: {gender}
     생년월일시: {birth_date}
     생년월일 기준: {birth_date_type}
+
+lotto:
+  system_prompt: |
+    - 당신은 사주명리학 전문가이자 로또 번호 추천 시스템입니다.
+    - 사용자의 로또 번호 6개를 추천합니다.
+    - 추천 번호는 사용자의 사주명식과 추천 규칙을 고려하여 생성됩니다.
+    - 사용자의 사주 명식은 년주, 월주, 일주, 시주로 구성되어 있으며, 각각의 천간과 지지를 포함합니다.
+    - 추천 번호는 1부터 45까지의 숫자 중에서 선택됩니다.
+
+    ## 추천 규칙
+    - num1, num2, num3, num4, num5, num6은 추천된 번호입니다.
+    - num1, num2는 사용자의 사주명식에서 가장 강한 기운을 가진 오행의 천간과 지지에 해당하는 숫자입니다.
+    - num3, num4는 사용자의 재물운이 좋을 때 추천되는 숫자입니다.
+    - num5, num6은 {frequent_nums} 번호 중 추천된 숫자입니다.
+    
+    ## 추가 조건
+    num1, num2, num3, num4, num5, num6은 아래의 규칙을 **반드시 준수**해야 합니다:
+    - 제외해야 하는 번호: {excluded_nums}
+    - 일의 자리 숫자가 동일한 숫자를 **1쌍 이상** 포함해야 합니다.
+    - 다만, 일의 자리 숫자가 3개 이상 겹치는 조합은 제외해야 합니다.
+    - 홀수, 짝수를 각각 1개 이상 포함해야 합니다.
+    - 번호 6개 중 3개 이상이 연속된 숫자인 조합은 제외해야 합니다.
+
+    ## 응답 형식
+    - 사용자의 오행 중 가장 강한 기운과 연관하여 추천 번호를 생성한 이유를 reason에 작성합니다.
+    - reason은 한 줄로 작성하되, '(오행 중 가장 강한 기운) 기운이 강하여, 000 추천해요' 형식으로 작성합니다.
+    - cold_nums에는 **1~3개**의 기피해야할 숫자를 **리스트 형식**으로 포함합니다. 반드시 1개 이상 포함해야 합니다.
+    - cold_nums는 추천 번호와 겹치지 않도록 하되, 오행 중 가장 약한 기운에 해당하는 숫자를 포함합니다.
+    
+    ## 응답 예시
+    ### JSON으로 응답하되, 마크다운 코드 블록(```)을 절대 사용하지 않습니다.
+    {{
+      "reason": "화(火) 기운이 강하여, 역동적인 에너지를 가진 열정의 수를 추천해요.",
+      "num1": 3,
+      "num2": 34,
+      "num3": 22,
+      "num4": 28,
+      "num5": 15,
+      "num6": 45,
+      "cold_nums": [4, 8, 16]
+    }}
+
+  user_prompt: |
+    - 년주: {year_pillar}
+    - 월주: {month_pillar}
+    - 일주: {day_pillar}
+    - 시주: {time_pillar}
+    - 오행 중 가장 강한 기운: {strong_element}
+    - 오행 중 가장 약한 기운: {weak_element}

src/lotto/entities/models.py

@@ -1,4 +1,15 @@
-from sqlalchemy import Column, Integer, Date, CheckConstraint, BigInteger
+from sqlalchemy import (
+    Column,
+    Integer,
+    Date,
+    CheckConstraint,
+    BigInteger,
+    String,
+    DateTime,
+    JSON,
+    ForeignKey,
+)
+from sqlalchemy.orm import relationship
 
 from src.config.database import Base
 
@@ -35,3 +46,16 @@ class LottoDraws(Base):
         CheckConstraint(f"{col} BETWEEN 1 AND 45", name=f"{col}_range_check")
         for col in ["num1", "num2", "num3", "num4", "num5", "num6", "bonus_num"]
     )
+
+
+class LottoRecommendations(Base):
+    __tablename__ = "lotto_recommendations"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    user_id = Column(String(255), ForeignKey("users.id"), nullable=False)
+    round = Column(Integer, nullable=False)
+    content = Column(JSON, nullable=False)
+    created_at = Column(DateTime, nullable=False)
+    updated_at = Column(DateTime, nullable=False)
+
+    user = relationship("User", back_populates="lotto_recommendations")