feat: 로또 번호 업데이트 스케줄러 추가 (#20)

* fix: 불필요 필드 제거

* fix: apscheduler package 추가

* feat: 로또 번호 업데이트 스케줄러 추가

* feat: 로또 회차별 조회 시 유저의 참여 여부 표기 추가

* fix: 스케줄러 요일 추가

---------

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

Author: 9keyyyy

poetry.lock

@@ -20,7 +20,8 @@ dependencies = [
     "gunicorn (>=23.0.0,<24.0.0)",
     "httpx (>=0.28.1,<0.29.0)",
     "pyyaml (>=6.0.2,<7.0.0)",
-    "greenlet (>=3.2.3,<4.0.0)"
+    "greenlet (>=3.2.3,<4.0.0)",
+    "apscheduler (>=3.11.0,<4.0.0)"
 ]
 
 

pyproject.toml

@@ -0,0 +1,7 @@
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+from src.common.scheduler.tasks import update_next_lotto_draw
+
+scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
+
+scheduler.add_job(update_next_lotto_draw, "cron", day_of_week="sat", hour="21", minute="0")

requirements.txt

@@ -0,0 +1,26 @@
+from src.common.logger import logger
+from src.config.config import db_config
+from src.config.database import Mysql
+from src.lotto.repository import LottoRepository
+from src.lotto.service import LottoService
+
+
+async def update_next_lotto_draw():
+    """다음 회차 로또 데이터를 동행복권 API에서 가져와서 데이터베이스에 저장합니다."""
+    db = Mysql(db_config)
+    
+    try:
+        async with db.session() as session:
+            repository = LottoRepository(session)
+            lotto_service = LottoService(repository)
+            success = await lotto_service.update_next_lotto_draw()
+            await session.commit()
+            if success:
+                logger.info("로또 데이터 업데이트가 성공적으로 완료되었습니다.")
+            else:
+                logger.warning("로또 데이터 업데이트가 실패했습니다.")
+                
+    except Exception as e:
+        logger.error(f"로또 데이터 업데이트 중 오류 발생: {e}")
+    finally:
+        await db.close()

src/common/scheduler/scheduler.py

@@ -3,13 +3,15 @@
 from fastapi import FastAPI
 
 from src.common.logger import start_logging
+from src.common.scheduler.scheduler import scheduler
 from src.config.config import db_config
 from src.config.database import Mysql
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     try:
+        scheduler.start()
         app.state.log_processor_task = await start_logging()
         app.state.mysql = Mysql(db_config)
         yield

src/common/scheduler/tasks.py

@@ -1,15 +1,12 @@
-import asyncio
-import traceback
+from collections import Counter
 from datetime import datetime, date
 from pathlib import Path
-from typing import List, Tuple, Dict
-from collections import Counter
+from typing import List, Tuple
 
-from src.four_pillars.entities.schemas import FourPillar, FourPillarDetail, PillarInfo
 from src.four_pillars.entities.enums import FiveElements, TenGods
+from src.four_pillars.entities.schemas import FourPillar, FourPillarDetail, PillarInfo
 from src.hcx_client.client import HCXClient
 from src.hcx_client.common.utils import HCXUtils
-from src.hcx_client.common.parser import Parser
 
 
 class FourPillarsCalculator:
@@ -413,16 +410,11 @@ 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)
         result: FourPillar = {
             "year_pillar": pillars[0],  # 년주
             "month_pillar": pillars[1],  # 월주
             "day_pillar": pillars[2],  # 일주
             "time_pillar": pillars[3],  # 시주
-            "strong_elements": strong_elements,
-            "weak_elements": weak_elements,
         }
 
         return result

src/config/lifespan.py

@@ -19,8 +19,6 @@ class FourPillar(TypedDict, total=False):
     month_pillar: str  # 월주
     day_pillar: str  # 일주
     time_pillar: Optional[str]  # 시주
-    strong_elements: Optional[List[str]]  # 가장 많은 오행
-    weak_elements: Optional[List[str]]  # 가장 적은 오행
 
 
 class FourPillarDetail(CommonBase):

src/four_pillars/common/calculator.py

@@ -1,4 +1,4 @@
-from datetime import date, datetime
+from datetime import date
 from typing import List, Optional
 
 from pydantic import BaseModel, Field, field_validator
@@ -19,6 +19,7 @@ class LottoDraw(CommonBase):
     bonus_num: int
     first_prize_amount: int
     total_winners: int
+    has_recommendation: bool = False
 
 
 class LottoDrawList(CommonBase):

src/four_pillars/entities/schemas.py

@@ -15,16 +15,42 @@ def __init__(self, session: AsyncSession = Depends(get_db_session)):
         self.session = session
 
     async def get_lotto_draws(
-        self, cursor: Optional[int] = None, limit: int = 10
+        self,
+        user_id: Optional[str] = None,
+        cursor: Optional[int] = None,
+        limit: int = 10,
     ) -> Tuple[List[LottoDraws], Optional[int]]:
-        query = select(LottoDraws).order_by(desc(LottoDraws.round))
+        if user_id:
+            query = (
+                select(
+                    LottoDraws,
+                    (LottoRecommendations.id.is_not(None)).label("has_recommendation"),
+                )
+                .outerjoin(
+                    LottoRecommendations,
+                    (LottoDraws.round == LottoRecommendations.round)
+                    & (LottoRecommendations.user_id == user_id),
+                )
+                .order_by(desc(LottoDraws.round))
+            )
+        else:
+            query = select(LottoDraws).order_by(desc(LottoDraws.round))
+
         if cursor:
             query = query.where(LottoDraws.round < cursor)
 
         query = query.limit(limit)
 
         result = await self.session.execute(query)
-        draws = result.scalars().all()
+
+        if user_id:
+            rows = result.fetchall()
+            draws = []
+            for draw, has_recommendation in rows:
+                draw.has_recommendation = bool(has_recommendation)
+                draws.append(draw)
+        else:
+            draws = result.scalars().all()
 
         next_cursor = draws[-1].round if len(draws) == limit else None
         return draws, next_cursor
@@ -57,6 +83,66 @@ async def get_latest_round(self) -> Optional[int]:
         latest = result.scalar_one_or_none()
         return latest
 
+    async def create_lotto_draw(self, lotto_data: dict) -> LottoDraws:
+        """로또 추첨 데이터를 생성합니다."""
+        lotto_draw = LottoDraws(**lotto_data)
+        self.session.add(lotto_draw)
+        await self.session.flush()
+        await self.session.refresh(lotto_draw)
+        return lotto_draw
+
+    async def get_lotto_draw_by_round(self, round: int) -> Optional[LottoDraws]:
+        """특정 회차의 로또 추첨 데이터를 조회합니다."""
+        query = select(LottoDraws).where(LottoDraws.round == round)
+        result = await self.session.execute(query)
+        return result.scalar_one_or_none()
+
+    async def update_lotto_statistics(self, lotto_data: dict) -> None:
+        """로또 통계 데이터를 업데이트합니다."""
+        # 메인 번호들 (1-6번)과 보너스 번호
+        main_numbers = [
+            lotto_data["num1"],
+            lotto_data["num2"],
+            lotto_data["num3"],
+            lotto_data["num4"],
+            lotto_data["num5"],
+            lotto_data["num6"],
+        ]
+        bonus_number = lotto_data["bonus_num"]
+
+        # 당첨된 번호들만 수집 (중복 제거)
+        winning_numbers = set(main_numbers + [bonus_number])
+
+        # 당첨된 번호들의 통계만 조회
+        stats_query = select(LottoStatistics).where(
+            LottoStatistics.num.in_(winning_numbers)
+        )
+        result = await self.session.execute(stats_query)
+        existing_stats = {stat.num: stat for stat in result.scalars().all()}
+
+        # 각 당첨 번호에 대해 통계 업데이트
+        for num in winning_numbers:
+            if num in existing_stats:
+                stat = existing_stats[num]
+            else:
+                # 새로운 통계 생성
+                stat = LottoStatistics(
+                    num=num, main_count=0, bonus_count=0, total_count=0
+                )
+                self.session.add(stat)
+
+            # 카운트 업데이트
+            if num in main_numbers:
+                stat.main_count += 1
+                stat.total_count += 1
+            elif num == bonus_number:
+                stat.bonus_count += 1
+                stat.total_count += 1
+
+            # 마지막 출현 정보 업데이트
+            stat.last_round = lotto_data["round"]
+            stat.last_date = lotto_data["draw_date"]
+
     async def create_lotto_recommendation(
         self, user_id: str, round: int, content: dict
     ) -> LottoRecommendations: