Merge pull request #21 from SynergyX-AI-Pattern/feat/#16_stock_search_by_image

Soomxn · web-flow · commit c19bf4e90c81 · 2025-08-15T19:15:05.000+09:00
diff --git a/.gitignore b/.gitignore
@@ -361,4 +361,7 @@ $RECYCLE.BIN/
 # Windows shortcuts
 *.lnk
 
+# API Keys
+keys/
+
 # End of https://www.toptal.com/developers/gitignore/api/pycharm,python,venv,windows,macos,virtualenv
diff --git a/app/api/v1/endpoints/stocks/__init__.py b/app/api/v1/endpoints/stocks/__init__.py
diff --git a/app/api/v1/endpoints/stocks/image_search.py b/app/api/v1/endpoints/stocks/image_search.py
@@ -0,0 +1,40 @@
+from fastapi import APIRouter, UploadFile, File, Depends
+from app.api_payload.code.error_status import ErrorStatus
+from app.db.session import get_db
+from sqlalchemy.orm import Session
+from app.api_payload.code.success_status import SuccessStatus
+from app.core.response import success_response
+from app.exceptions.base import APIException
+from app.schemas.base_response import BaseResponse
+from app.schemas.image_search import ImageSearchResponse
+from app.services.image_search_service import ImageSearchService
+
+router = APIRouter()
+
+@router.post(
+    "/search-by-image",
+    response_model=BaseResponse,
+    summary="이미지 기반 종목 검색",
+    description="""
+                업로드된 이미지로부터 브랜드를 인식하여 종목 정보를 반환합니다.
+                최대 5MB 파일만 업로드 가능합니다.
+    """,
+    tags=["Stock"]
+)
+async def search_stock_by_image(
+        image: UploadFile = File(...),
+        db: Session = Depends(get_db)
+):
+    contents = await image.read()
+    MAX_IMAGE_SIZE = 5 * 1024 * 1024  # 5MB
+
+    # 파일 크기 제한(5MB)
+    if len(contents) > MAX_IMAGE_SIZE:
+        raise APIException(ErrorStatus.FILE_TOO_LARGE)
+
+    result: ImageSearchResponse = ImageSearchService.search_stock_by_image(contents, db)
+
+    return success_response(
+        data=result,
+        status=SuccessStatus.IMAGE_SEARCH_SUCCESS
+    )
diff --git a/app/api/v1/endpoints/stocks/stock.py b/app/api/v1/endpoints/stocks/stock.py
diff --git a/app/api/v1/routers/__init__.py b/app/api/v1/routers/__init__.py
@@ -1,10 +1,11 @@
 from fastapi import APIRouter
-from app.api.v1.routers import stock_router, backtest_router, pattern_detection_router
+from app.api.v1.routers import stock_router, backtest_router, pattern_detection_router, emotion_diary_router
 
 router = APIRouter()
 
 router.include_router(stock_router.router, prefix="/v1")
 router.include_router(backtest_router.router, prefix="/v1")
 router.include_router(pattern_detection_router.router, prefix="/v1")
+router.include_router(emotion_diary_router.router, prefix="/v1")
 
 # 아래에 추가 ex. router.include_router(user_router.router, prefix="/v1")
diff --git a/app/api/v1/routers/stock_router.py b/app/api/v1/routers/stock_router.py
@@ -1,5 +1,6 @@
 from fastapi import APIRouter
-from app.api.v1.endpoints import stock
+from app.api.v1.endpoints.stocks import stock, image_search
 
 router = APIRouter()
 router.include_router(stock.router, prefix="/stocks", tags=["Stock"])
+router.include_router(image_search.router, prefix="/stocks", tags=["Stock"])
diff --git a/app/api_payload/code/error_status.py b/app/api_payload/code/error_status.py
@@ -7,6 +7,9 @@ class ErrorStatus(Enum):
     VALIDATION_ERROR = ("VALIDATION_ERROR", "유효성 검사 실패", status.HTTP_400_BAD_REQUEST)
     MODEL_ERROR = ("MODEL_ERROR", "모델 예측 중 오류 발생", status.HTTP_500_INTERNAL_SERVER_ERROR)
     NOTIFICATION_SEND_FAILED = ("NOTIFICATION_SEND_FAILED", "알림 전송에 실패했습니다.", status.HTTP_500_INTERNAL_SERVER_ERROR)
+    GPT_API_ERROR = ("GPT_API_ERROR", "GPT API 호출에 실패했습니다.", status.HTTP_502_BAD_GATEWAY)
+    FILE_TOO_LARGE = ("FILE_TOO_LARGE", "업로드한 파일의 크기가 너무 큽니다.", status.HTTP_413_REQUEST_ENTITY_TOO_LARGE)
+    GPT_RESPONSE_PARSE_ERROR = ("GPT_RESPONSE_PARSE_ERROR", "GPT 응답을 JSON으로 파싱할 수 없습니다.", status.HTTP_500_INTERNAL_SERVER_ERROR)
 
     # 종목
     STOCK_NOT_FOUND = ("STOCK_NOT_FOUND", "해당 종목을 찾을 수 없습니다.", status.HTTP_404_NOT_FOUND)
diff --git a/app/api_payload/code/success_status.py b/app/api_payload/code/success_status.py
@@ -9,6 +9,8 @@ class SuccessStatus(Enum):
     BACKTEST_EXECUTED = ("BACKTEST_EXECUTED", "백테스팅 실행 성공", status.HTTP_200_OK)
     STOCK_PREDICTED = ("STOCK200", "15일 예측 성공", status.HTTP_200_OK)
     PATTERN_DETECTION_EXECUTED = ("PATTERN_DETECTION_EXECUTED", "패턴 감지 성공", status.HTTP_200_OK)
+    IMAGE_SEARCH_SUCCESS = ("IMAGE_SEARCH_SUCCESS", "이미지로 종목 검색 성공", status.HTTP_200_OK)
+    DIARY_ANALYSIS_SUCCESS = ("DIARY_ANALYSIS_SUCCESS", "감정 분석 성공", status.HTTP_200_OK)
 
     def __init__(self, code, message, http_status):
         self.code = code
diff --git a/app/core/config.py b/app/core/config.py
@@ -1,12 +1,14 @@
 from pydantic_settings import BaseSettings
 from typing import Literal
-from pydantic import HttpUrl
+from pydantic import HttpUrl, SecretStr
 
 class Settings(BaseSettings):
     DATABASE_URL: str
     ENV: Literal["local", "prod"] = "prod"  # 기본값 prod
     DISCORD_WEBHOOK_URL: str
     SPRING_SERVER_BASE_URL: HttpUrl
+    GOOGLE_APPLICATION_CREDENTIALS: str
+    OPENAI_API_KEY: SecretStr
 
     class Config:
         env_file = ".env"
diff --git a/app/crud/stock.py b/app/crud/stock.py
@@ -1,3 +1,5 @@
+from difflib import get_close_matches
+from sqlalchemy import func
 from sqlalchemy.orm import Session
 from app.models.stock import Stock
 
@@ -7,3 +9,33 @@ def get_stock_by_id(db: Session, stock_id: int) -> Stock | None:
     stock_id 기준으로 종목 정보를 조회합니다.
     """
     return db.query(Stock).filter(Stock.id == stock_id).first()
+
+def get_stock_by_name(db: Session, name: str) -> Stock | None:
+    """
+    종목명 기준으로 종목 정보를 조회합니다.
+    """
+
+    # 입력 정규화
+    # 공백 제거 및 대소문자 무시
+    normalized_name = name.replace(" ", "").lower()
+
+    # 1차 : 정확히 일치할 경우 반환
+    stock = (
+        db.query(Stock)
+        .filter(func.replace(func.lower(Stock.name), " ", "") == normalized_name)
+        .first()
+    )
+    if stock:
+        return stock
+
+    # 2차 : 유사도 기반 매칭 결과 반환
+    all_stock_names = db.query(Stock.name).all()
+    name_list = [str(row[0]) for row in all_stock_names]
+    close_matches = get_close_matches(name, name_list, n=1, cutoff=0.6) # 유사도 60% 이상
+
+    if close_matches:
+        return db.query(Stock).filter(Stock.name == close_matches[0]).first()
+
+    # 없을 경우 None 리턴
+    return None
+
diff --git a/app/schemas/image_search.py b/app/schemas/image_search.py
@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class ImageSearchResponse(BaseModel):
+    id: Optional[int]
+
+    model_config = {
+        "from_attributes": True,
+        "json_schema_extra": {
+            "example": {
+                "id": 1
+            }
+        }
+    }
diff --git a/app/services/external/__init__.py b/app/services/external/__init__.py
diff --git a/app/services/external/gpt_service.py b/app/services/external/gpt_service.py
@@ -0,0 +1,44 @@
+import logging
+from openai import OpenAI
+
+from app.api_payload.code.error_status import ErrorStatus
+from app.core.config import settings
+
+from app.exceptions.base import APIException
+
+logger = logging.getLogger(__name__)
+
+
+# GPT 클라이언트 생성
+client = OpenAI(api_key=settings.OPENAI_API_KEY.get_secret_value())
+
+
+class GPTService:
+    @staticmethod
+    def infer_brand_name_from_keywords(keywords: list[str]) -> str:
+        """
+        키워드로 기업명을 추론합니다.
+        """
+        if not keywords:
+            return "모름"
+
+        prompt = (
+            "아래 키워드를 보고 어떤 KOSPI 100 상장 종목과 가장 관련이 있는지 추론해줘.\n"
+            "반드시 한국 증시에 상장된 **KOSPI 100 종목명 중 하나**로만 대답해야 해.\n"
+            "예를 들어, '삼성전자', '삼성바이오로직스', 'LG화학'처럼 종목명만 정확하게 대답해줘.\n\n"
+            f"키워드 목록: {', '.join(keywords)}\n\n"
+            "설명 없이 종목명만 한 줄로 정확하게 대답해. 그 외 다른 말은 하지 마."
+            "키워드가 제공되지 않았거나 판단할 정보가 없는 경우 '모름' 이라고 말해줘."
+        )
+
+        try:
+            response = client.chat.completions.create(
+                model="gpt-4o",
+                messages=[{"role": "user", "content": prompt}],
+                temperature=0.2
+            )
+            return response.choices[0].message.content.strip()
+        except Exception as e:
+            logger.error(f"[GPTService] GPT API 호출 실패: {e}")
+            raise APIException(ErrorStatus.GPT_API_ERROR)
+
diff --git a/app/services/external/vision_service.py b/app/services/external/vision_service.py
@@ -0,0 +1,92 @@
+import logging, os, re
+from google.cloud import vision
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+# Google Cloud Vision API 클라이언트 (지연 초기화)
+client: vision.ImageAnnotatorClient | None = None
+
+class VisionService:
+    """
+    Google Cloud Vision API 연동 서비스입니다.
+    이미지에서 라벨, 텍스트, 로고, 웹 정보를 추출합니다.
+    """
+
+    @staticmethod
+    def extract_keywords(image_bytes: bytes) -> dict[str, list[str]]:
+        """
+        이미지에서 키워드를 추출합니다.
+        """
+        global client
+        if client is None:
+            os.environ.setdefault("GOOGLE_APPLICATION_CREDENTIALS", settings.GOOGLE_APPLICATION_CREDENTIALS)
+            client = vision.ImageAnnotatorClient()
+
+        # 이미지 바이트 데이터를 이미지 객체로 변환
+        image = vision.Image(content=image_bytes)
+
+        # 딕셔너리 초기화
+        result = {"label": [], "text": [], "logo": [], "web": []}
+
+        logger.debug("[VisionService] Vision 분석 시작")
+
+        # 라벨
+        try:
+            label_res = client.label_detection(image=image)
+            result["label"] = [l.description for l in label_res.label_annotations]
+        except Exception as e:
+            logger.warning(f"[VisionService] 라벨 분석 실패: {e}")
+
+        # 텍스트
+        try:
+            text_res = client.text_detection(image=image)
+            result["text"] = [t.description for t in text_res.text_annotations]
+        except Exception as e:
+            logger.warning(f"[VisionService] 텍스트 분석 실패: {e}")
+
+        # 로고
+        try:
+            logo_res = client.logo_detection(image=image)
+            result["logo"] = [l.description for l in logo_res.logo_annotations]
+        except Exception as e:
+            logger.warning(f"[VisionService] 로고 분석 실패: {e}")
+
+        # 웹
+        try:
+            web_res = client.web_detection(image=image)
+            if web_res.web_detection and web_res.web_detection.web_entities:
+                result["web"] = [w.description for w in web_res.web_detection.web_entities if w.description]
+        except Exception as e:
+            logger.warning(f"[VisionService] 웹 분석 실패: {e}")
+
+        # 디버그용 로그 출력
+        logger.debug(
+            f"[VisionService] Vision 분석 완료: "
+            f"label={result['label'][:3]}, text={result['text'][:3]}, "
+            f"logo={result['logo'][:3]}, web={result['web'][:3]}"
+        )
+        return result
+
+    @staticmethod
+    def normalize_keywords(keyword_sources: dict[str, list[str]]) -> list[str]:
+        """
+        키워드를 정규화합니다.
+        """
+        # 중복 제거 set 사용
+        all_keywords = set()
+
+        for source_keywords in keyword_sources.values():
+            for keyword in source_keywords:
+                # 특수문자 제거 (영문/한글/숫자/공백만 허용)
+                cleaned = re.sub(r'[^a-zA-Z가-힣0-9\s]', ' ', keyword)
+
+                # 앞뒤 공백 제거 후 단어 단위로 분할
+                parts = cleaned.strip().split()
+
+                # 전체 키워드 집합에 추가
+                all_keywords.update(parts)
+
+        return list(all_keywords)
+
diff --git a/app/services/image_search_service.py b/app/services/image_search_service.py
@@ -0,0 +1,38 @@
+import logging
+from sqlalchemy.orm import Session
+from app.crud.stock import get_stock_by_name
+from app.exceptions.base import APIException
+from app.api_payload.code.error_status import ErrorStatus
+from app.schemas.image_search import ImageSearchResponse
+from app.services.external.gpt_service import GPTService
+from app.services.external.vision_service import VisionService
+
+logger = logging.getLogger(__name__)
+
+
+class ImageSearchService:
+    @staticmethod
+    def search_stock_by_image(contents: bytes, db: Session) -> ImageSearchResponse:
+        """
+        이미지 → Vision api → GPT api → 종목 검색
+        """
+        # 키워드 추출
+        keywords_dict = VisionService.extract_keywords(contents)
+
+        # 키워드 정규화
+        keyword_list = VisionService.normalize_keywords(keywords_dict)
+
+        # 종목 추론
+        brand_name = GPTService.infer_brand_name_from_keywords(keyword_list)
+
+        # 종목 검색
+        logger.debug(f"[ImageSearchService] {brand_name}으로 종목 검색 시작")
+        stock = get_stock_by_name(db, brand_name)
+        if not stock:
+            logger.warning(f"[ImageSearchService] {brand_name}와 일치하는 종목을 찾을 수 없습니다.")
+            raise APIException(ErrorStatus.STOCK_NOT_FOUND)
+        logger.info(f"[ImageSearchService] 매칭된 종목 : {stock.name}")
+
+        return ImageSearchResponse(
+            id = stock.id
+        )
diff --git a/app/tests/services/test_image_search_service.py b/app/tests/services/test_image_search_service.py
