[NL-56] 로또 결과/통계 테이블 추가, 데이터 삽입 스크립트 작성, 회차/번호별 통계 리스트 조회 API 구현 (#8)

* feat: lotto draws, statistics table 생성 및 alembic migration

* feat: 로또 회차별 리스트 조회/번호별 통계 조회 API 구현

* fix: prize amount type 변경

* feat: lotto data insert 스크립트 추가

* fix: 매핑 수정

* Update src/lotto/entities/models.py

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

---------

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

Author: 3 people

migrations/env.py

@@ -29,6 +29,7 @@
 
 # 모든 도메인 모델을 import하여 메타데이터에 포함해야 함
 from src.users.entities.models import User
+from src.lotto.entities.models import LottoStatistics, LottoDraws
 
 # 모델 메타데이터 설정
 target_metadata = Base.metadata

migrations/scripts/insert_lotto_data.py

@@ -0,0 +1,202 @@
+#!/usr/bin/env python3
+"""
+로또 데이터를 동행복권 API에서 가져와서 데이터베이스에 저장하는 스크립트
+"""
+
+import asyncio
+import json
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, Any, Optional
+
+import httpx
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+# .env 파일 로드
+from dotenv import load_dotenv
+load_dotenv()
+
+from src.config.database import Mysql
+from src.config.config import db_config
+from src.lotto.entities.models import LottoDraws, LottoStatistics
+
+
+class LottoDataImporter:
+    def __init__(self):
+        self.db = Mysql(db_config)
+        self.base_url = "https://dhlottery.co.kr/common.do"
+        
+    async def fetch_lotto_data(self, client: httpx.AsyncClient, drw_no: int) -> Optional[Dict[str, Any]]:
+        """특정 회차의 로또 데이터를 가져옵니다."""
+        params = {
+            "method": "getLottoNumber",
+            "drwNo": drw_no
+        }
+        
+        try:
+            response = await client.get(self.base_url, params=params)
+            if response.status_code == 200:
+                data = response.json()
+                if data.get("returnValue") == "success":
+                    return data
+                else:
+                    print(f"회차 {drw_no}: API 응답 오류 - {data}")
+                    return None
+            else:
+                print(f"회차 {drw_no}: HTTP 오류 - {response.status_code}")
+                return None
+        except Exception as e:
+            print(f"회차 {drw_no}: 요청 오류 - {e}")
+            return None
+    
+    def parse_lotto_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        """API 응답 데이터를 데이터베이스 모델에 맞게 파싱합니다."""
+        return {
+            "round": data["drwNo"],
+            "draw_date": datetime.strptime(data["drwNoDate"], "%Y-%m-%d").date(),
+            "num1": data["drwtNo1"],
+            "num2": data["drwtNo2"],
+            "num3": data["drwtNo3"],
+            "num4": data["drwtNo4"],
+            "num5": data["drwtNo5"],
+            "num6": data["drwtNo6"],
+            "bonus_num": data["bnusNo"],
+            "first_prize_amount": data["firstWinamnt"],
+            "total_winners": data["firstPrzwnerCo"]
+        }
+    
+    async def save_lotto_draw(self, db_session: AsyncSession, lotto_data: Dict[str, Any]) -> bool:
+        """로또 추첨 데이터를 데이터베이스에 저장합니다."""
+        try:
+            # 이미 존재하는지 확인
+            existing = await db_session.execute(
+                select(LottoDraws).where(LottoDraws.round == lotto_data["round"])
+            )
+            if existing.scalar_one_or_none():
+                print(f"회차 {lotto_data['round']}: 이미 존재함")
+                return False
+            
+            # 새로운 로또 추첨 데이터 생성
+            lotto_draw = LottoDraws(**lotto_data)
+            db_session.add(lotto_draw)
+            await db_session.commit()
+            print(f"회차 {lotto_data['round']}: 저장 완료")
+            return True
+        except Exception as e:
+            await db_session.rollback()
+            print(f"회차 {lotto_data['round']}: 저장 오류 - {e}")
+            return False
+    
+    async def update_statistics(self, db_session: AsyncSession, lotto_data: Dict[str, Any]) -> None:
+        """로또 통계 데이터를 업데이트합니다."""
+        try:
+            # 메인 번호들 (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"]
+            
+            # 각 번호에 대한 통계 업데이트
+            for num in range(1, 46):
+                # 기존 통계 조회
+                stat = await db_session.execute(
+                    select(LottoStatistics).where(LottoStatistics.num == num)
+                )
+                stat = stat.scalar_one_or_none()
+                
+                if not stat:
+                    # 새로운 통계 생성
+                    stat = LottoStatistics(
+                        num=num,
+                        main_count=0,
+                        bonus_count=0,
+                        total_count=0
+                    )
+                    db_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
+                
+                # 마지막 출현 정보 업데이트
+                if num in main_numbers or num == bonus_number:
+                    stat.last_round = lotto_data["round"]
+                    stat.last_date = lotto_data["draw_date"]
+            
+            await db_session.commit()
+            print(f"회차 {lotto_data['round']}: 통계 업데이트 완료")
+            
+        except Exception as e:
+            await db_session.rollback()
+            print(f"회차 {lotto_data['round']}: 통계 업데이트 오류 - {e}")
+    
+    async def import_lotto_data(self, start_drw_no: int = 1, end_drw_no: int = 1183):
+        """지정된 범위의 로또 데이터를 가져와서 저장합니다."""
+        print(f"로또 데이터 가져오기 시작: 회차 {start_drw_no} ~ {end_drw_no}")
+        
+        async with httpx.AsyncClient() as client:
+            async with self.db.session() as db_session:
+                success_count = 0
+                error_count = 0
+                
+                for drw_no in range(start_drw_no, end_drw_no + 1):
+                    print(f"처리 중: 회차 {drw_no}")
+                    
+                    # API에서 데이터 가져오기
+                    api_data = await self.fetch_lotto_data(client, drw_no)
+                    if not api_data:
+                        error_count += 1
+                        continue
+                    
+                    # 데이터 파싱
+                    lotto_data = self.parse_lotto_data(api_data)
+                    
+                    # 데이터베이스에 저장
+                    if await self.save_lotto_draw(db_session, lotto_data):
+                        # 통계 업데이트
+                        await self.update_statistics(db_session, lotto_data)
+                        success_count += 1
+                    else:
+                        error_count += 1
+                    
+                    # API 호출 간격 조절 (서버 부하 방지)
+                    await asyncio.sleep(0.1)
+                
+                print(f"\n가져오기 완료:")
+                print(f"성공: {success_count}개")
+                print(f"실패: {error_count}개")
+    
+    async def close(self):
+        """데이터베이스 연결을 종료합니다."""
+        await self.db.close()
+
+
+async def main():
+    """메인 함수"""
+    importer = LottoDataImporter()
+    
+    try:
+        # 전체 회차 데이터 가져오기 (1~1183)
+        await importer.import_lotto_data(1, 1183)
+        
+        # 또는 특정 범위만 가져오기
+        # await importer.import_lotto_data(1100, 1183)
+        
+    except KeyboardInterrupt:
+        print("\n사용자에 의해 중단되었습니다.")
+    except Exception as e:
+        print(f"오류 발생: {e}")
+    finally:
+        await importer.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main()) 

migrations/versions/05818a33a6e5_fix_lotto_prize_amount_type.py

@@ -0,0 +1,38 @@
+"""fix lotto prize amount type
+
+Revision ID: 05818a33a6e5
+Revises: 47e81b19f851
+Create Date: 2025-08-05 23:53:32.318980
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision: str = '05818a33a6e5'
+down_revision: Union[str, Sequence[str], None] = '47e81b19f851'
+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.alter_column('lotto_draws', 'first_prize_amount',
+               existing_type=mysql.INTEGER(),
+               type_=sa.BigInteger(),
+               existing_nullable=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('lotto_draws', 'first_prize_amount',
+               existing_type=sa.BigInteger(),
+               type_=mysql.INTEGER(),
+               existing_nullable=False)
+    # ### end Alembic commands ###

migrations/versions/47e81b19f851_add_lotto_draws_and_statistics.py

@@ -0,0 +1,67 @@
+"""add lotto draws and statistics
+
+Revision ID: 47e81b19f851
+Revises: ec45a4f91988
+Create Date: 2025-08-05 23:23:51.373015
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '47e81b19f851'
+down_revision: Union[str, Sequence[str], None] = 'ec45a4f91988'
+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_draws',
+    sa.Column('round', sa.Integer(), autoincrement=False, nullable=False),
+    sa.Column('draw_date', sa.Date(), nullable=False),
+    sa.Column('num1', sa.Integer(), nullable=False),
+    sa.Column('num2', sa.Integer(), nullable=False),
+    sa.Column('num3', sa.Integer(), nullable=False),
+    sa.Column('num4', sa.Integer(), nullable=False),
+    sa.Column('num5', sa.Integer(), nullable=False),
+    sa.Column('num6', sa.Integer(), nullable=False),
+    sa.Column('bonus_num', sa.Integer(), nullable=False),
+    sa.Column('first_prize_amount', sa.Integer(), nullable=False),
+    sa.Column('total_winners', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), nullable=False),
+    sa.CheckConstraint('bonus_num BETWEEN 1 AND 45', name='bonus_num_range_check'),
+    sa.CheckConstraint('num1 BETWEEN 1 AND 45', name='num1_range_check'),
+    sa.CheckConstraint('num2 BETWEEN 1 AND 45', name='num2_range_check'),
+    sa.CheckConstraint('num3 BETWEEN 1 AND 45', name='num3_range_check'),
+    sa.CheckConstraint('num4 BETWEEN 1 AND 45', name='num4_range_check'),
+    sa.CheckConstraint('num5 BETWEEN 1 AND 45', name='num5_range_check'),
+    sa.CheckConstraint('num6 BETWEEN 1 AND 45', name='num6_range_check'),
+    sa.PrimaryKeyConstraint('round')
+    )
+    op.create_table('lotto_statistics',
+    sa.Column('num', sa.Integer(), autoincrement=False, nullable=False),
+    sa.Column('main_count', sa.Integer(), nullable=True),
+    sa.Column('bonus_count', sa.Integer(), nullable=True),
+    sa.Column('total_count', sa.Integer(), nullable=True),
+    sa.Column('last_round', sa.Integer(), nullable=True),
+    sa.Column('last_date', sa.Date(), nullable=True),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), nullable=False),
+    sa.CheckConstraint('num BETWEEN 1 AND 45', name='num_range_check'),
+    sa.PrimaryKeyConstraint('num')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('lotto_statistics')
+    op.drop_table('lotto_draws')
+    # ### end Alembic commands ###

poetry.lock

@@ -19,7 +19,8 @@ dependencies = [
     "cryptography (>=45.0.5,<46.0.0)",
     "gunicorn (>=23.0.0,<24.0.0)",
     "httpx (>=0.28.1,<0.29.0)",
-    "pyyaml (>=6.0.2,<7.0.0)"
+    "pyyaml (>=6.0.2,<7.0.0)",
+    "greenlet (>=3.2.3,<4.0.0)"
 ]
 
 

pyproject.toml

@@ -0,0 +1,6 @@
+from enum import Enum
+
+
+class SortType(str, Enum):
+    FREQUENCY = "frequency"  # 빈도순
+    NUMBER = "number"        # 번호순

src/lotto/__init__.py

@@ -0,0 +1,37 @@
+from sqlalchemy import Column, Integer, Date, CheckConstraint, BigInteger
+
+from src.config.database import Base
+
+
+class LottoStatistics(Base):
+    __tablename__ = "lotto_statistics"
+
+    num = Column(Integer, primary_key=True, autoincrement=False)
+    main_count = Column(Integer, default=0)  # 메인 번호로 나온 횟수
+    bonus_count = Column(Integer, default=0)  # 보너스 번호로 나온 횟수
+    total_count = Column(Integer, default=0)  # 전체 출현 횟수
+    last_round = Column(Integer)  # 마지막 출현 회차
+    last_date = Column(Date)  # 마지막 출현 날짜
+
+    __table_args__ = (CheckConstraint("num BETWEEN 1 AND 45", name="num_range_check"),)
+
+
+class LottoDraws(Base):
+    __tablename__ = "lotto_draws"
+
+    round = Column(Integer, primary_key=True, autoincrement=False)
+    draw_date = Column(Date, nullable=False)
+    num1 = Column(Integer, nullable=False)
+    num2 = Column(Integer, nullable=False)
+    num3 = Column(Integer, nullable=False)
+    num4 = Column(Integer, nullable=False)
+    num5 = Column(Integer, nullable=False)
+    num6 = Column(Integer, nullable=False)
+    bonus_num = Column(Integer, nullable=False)
+    first_prize_amount = Column(BigInteger, nullable=False)
+    total_winners = Column(Integer, nullable=False)
+
+    __table_args__ = tuple(
+        CheckConstraint(f"{col} BETWEEN 1 AND 45", name=f"{col}_range_check")
+        for col in ["num1", "num2", "num3", "num4", "num5", "num6", "bonus_num"]
+    )