DEPLOY v1.0.0

.github/workflows/workflow.yml

@@ -37,6 +37,14 @@ jobs:
       - name: Checkout Code
         uses: actions/checkout@v4
 
+      - name: Transfer deploy file to EC2
+        uses: appleboy/scp-action@master
+        with:
+          host: ${{ secrets.REMOTE_HOST }}
+          username: ${{ secrets.REMOTE_USER }}
+          key: ${{ secrets.SSH_PRIVATE_KEY }}
+          source: ./deployment/
+          target: /home/ubuntu/deployment
 
   deploy:
     name: Deploy
@@ -52,8 +60,8 @@ jobs:
           port: 22
           script: |
             # 1. cd
-            mkdir -p ~/dearbelly
-            cd ~/dearbelly
+            mkdir -p ~/dearbelly/deployment
+            cd ~/dearbelly/deployment
 
             # 2. .env file
             echo "${{ secrets.ENV }}" > .env
@@ -69,5 +77,5 @@ jobs:
             docker pull ${{ secrets.ECR_URI }}/dearbelly-cv:latest
 
             # 6. docker start
-            # TODO : 스크립트 작성
-            docker run -d --name app-blue -p 8000:8000 ${{ secrets.ECR_URI }}/dearbelly-cv:latest
+            sudo chmod +x deploy.sh
+            source deploy.sh

.gitignore

@@ -40,7 +40,7 @@ MANIFEST
 pip-log.txt
 pip-delete-this-directory.txt
 
-# Unit test / coverage reports
+# Unit tests / coverage reports
 htmlcov/
 .tox/
 .nox/

app/api/endpoints/predictions.py

@@ -4,30 +4,43 @@
 import redis.asyncio as redis
 import uuid
 from datetime import datetime
+import json
 
 router = APIRouter()
 
 def get_redis_client(request: Request) -> redis.Redis:
     return request.app.state.redis_client
 
+"""
+테스트를 위한 임시적인 API
+"""
 @router.post("/predict", status_code=202)
 async def create_prediction_job(job_request: JobRequest, redis_client: redis.Redis = Depends(get_redis_client)):
-    correlation_id = str(uuid.uuid4())
+    correlationId = str(uuid.uuid4())
     job = ImageJob(
         correlationId=correlationId,
-        presignedUrl=job_request.presigned_url,
+        presignedUrl=job_request.presignedUrl,
         replyQueue=settings.STREAM_RESULT,
-        callbackUrl=None,
-        contentType="image/jpeg",
+        contentType=job_request.contentType,
         createdAt=datetime.utcnow().isoformat(),
         ttlSec=3600,
     )
 
-    await redis_client.xadd(
+    # 타입에 맞도록 넣어주기
+    correlationId = job.correlationId
+    payload = json.dumps(job.dict())
+    print("분석 결과 발행 시작...")
+    entry_id = await redis_client.xadd(
         settings.STREAM_JOB,
-        {"json": job.model_dump_json()},
+        {
+            "type": "image_results",
+            "payload": payload,
+            "correlationId": correlationId,
+        },
         maxlen=10_000,
         approximate=True,
     )
 
-    return {"job_id": correlation_id}
+    print("분석 결과 발행 완료...")
+
+    return {"job_id": correlationId}

app/core/config.py

@@ -5,14 +5,11 @@
 load_dotenv()
 
 redis_url = os.getenv("REDIS_SERVER_URL")
-bucket_name = os.getenv("BUCKET_NAME")
-s3_region = os.getenv("S3_REGION")
-
+openai_key = os.getenv("OPENAI_API_KEY")
 
 class Settings(BaseSettings):
     REDIS_URL: str = redis_url
-    S3_REGION: str = bucket_name
-    BUCKET_NAME: str = s3_region
+    OPENAI_KEY: str = openai_key
 
     STREAM_JOB: str = "image.jobs" # SpringBoot에서 job 발행 (FastAPI에서 listen)
     STREAM_RESULT: str = "image.results"  # FastAPI에서 결과 발행 (SpringBoot에서 listen)

app/core/lifespan.py

@@ -1,26 +1,24 @@
 
 import asyncio
 from contextlib import asynccontextmanager
-import redis.asyncio as redis
+
+import anyio.to_thread
+import redis
 from fastapi import FastAPI
 
+from app.worker.redis_client import redis_client
 from app.core.config import settings
 from app.worker.worker import JobWorker
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True)
+    limiter = anyio.to_thread.current_default_thread_limiter()
+    limiter.total_tokens = 100
 
-    try:
-        await redis_client.xgroup_create(
-            name=settings.STREAM_JOB,
-            groupname=settings.GROUP_NAME,
-            id="$",
-            mkstream=True
-        )
-    except redis.ResponseError as e:
-        if "BUSYGROUP" not in str(e):
-            raise
+    redis_client.xgroup_create(
+        stream_name=settings.STREAM_JOB,
+        group_name=settings.GROUP_NAME,
+    )
 
     worker = JobWorker(redis_client)
     worker_task = asyncio.create_task(worker.run())

app/schemas/job.py

@@ -2,20 +2,25 @@
 from pydantic import BaseModel, Field
 
 class JobRequest(BaseModel):
-    presigned_url: str = Field(..., description="다운로드할 이미지의 Presigned URL")
+    presignedUrl: str = Field(..., description="다운로드할 이미지의 Presigned URL")
 
+"""
+FastAPI가 image.results Stream에 발행하는 메시지
+"""
 class JobResult(BaseModel):
-    pill_name: str
-    correlation_id: str
-    label: str
-    confidence: float
-    finished_at: str
+    correlationId: str
+    pillName: str
+    isSafe: int
+    description: str
+    finishedAt: str
 
+"""
+Spring으로부터 받는 Job(image.jobs 구독)
+"""
 class ImageJob(BaseModel):
-    correlation_id: str = Field(alias="correlationId")
-    presigned_url: str = Field(alias="presignedUrl")
-    reply_queue: str = Field(alias="replyQueue")
-    callback_url: str | None = Field(alias="callbackUrl")
-    content_type: str = Field(alias="contentType")
-    created_at: str = Field(alias="createdAt")
-    ttl_sec: int = Field(alias="ttlSec")
+    correlationId: str = Field(alias="correlationId")
+    presignedUrl: str = Field(alias="presignedUrl")
+    replyQueue: str = Field(alias="replyQueue")
+    contentType: str = Field(alias="contentType")
+    createdAt: str = Field(alias="createdAt")
+    ttlSec: int = Field(alias="ttlSec")

app/services/openai_service.py

@@ -0,0 +1,66 @@
+from app.core.config import settings
+from openai import OpenAI
+import json
+
+client = OpenAI(api_key=settings.OPENAI_KEY)
+
+class PregnancySafetyChecker:
+    def __init__(self, client: OpenAI):
+        self.client = client
+
+    """
+    - isSafe: 안전하면 1, 안전하지 않으면 0
+    - description: 복용 가능 여부 설명
+    """
+    def ask_chatgpt_about_pregnancy_safety(self, pill_name: str) -> tuple[str, int]:
+        prompt = f"""
+        약 이름: {pill_name}
+        질문: 이 약은 임산부가 복용해도 안전한가요? 복용 가능 여부와 주의사항을 알려주세요.
+        description 안에는 문장마다 \\n 을 적용하세요.
+        결과를 JSON 형식으로 정확히 반환하세요. 설명이나 다른 텍스트를 절대 덧붙이지 마세요.
+        스키마:
+        {{
+            "description": "복용 가능 여부 및 주의사항에 대한 설명",
+            "isSafe": 1 또는 0
+        }}
+        """
+
+        response = self.client.chat.completions.create(
+            model="gpt-4o-mini",
+            messages=[{"role": "user", "content": prompt}],
+            temperature=0,
+            max_tokens=600,
+            response_format={"type": "json_object"}
+        )
+        print("GPT Asking 성공...")
+        raw = response.choices[0].message.content.strip()
+
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError:
+            start = raw.find("{")
+            end = raw.rfind("}")
+            if start != -1 and end != -1 and start < end:
+                data = json.loads(raw[start:end+1])
+            else:
+                # 디버깅
+                preview = raw[:200].replace("\n", "\\n")
+                raise ValueError(f"응답이 유효한 JSON이 아닙니다. preview='{preview}'")
+
+        description = data.get("description")
+        isSafe = data.get("isSafe")
+
+        if isinstance(isSafe, bool):
+            isSafe = 1 if isSafe else 0
+        elif isinstance(isSafe, str):
+            isSafe = 1 if isSafe.strip() in {"1", "true", "True"} else 0
+        elif not isinstance(isSafe, int):
+            isSafe = 0
+
+        if not isinstance(description, str):
+            description = ""  # 안전장치
+
+        return description, int(isSafe)
+
+
+checker = PregnancySafetyChecker(client)

app/services/predictor_service.py

@@ -6,6 +6,7 @@
 from PIL import Image
 import json
 from pathlib import Path
+from io import BytesIO
 
 class LightCNN(nn.Module):
     def __init__(self, num_classes):
@@ -53,8 +54,8 @@ def _load_model(self, model_path: Path) -> LightCNN:
         model.eval()
         return model
 
-    def predict(self, image_path: Path) -> tuple[str, str, float]:
-        image = Image.open(image_path).convert('RGB')
+    def predict(self, stream_file: BytesIO) -> tuple[str, str, float]:
+        image = Image.open(stream_file).convert('RGB')
         input_tensor = self.transform(image).unsqueeze(0).to(self.device)
 
         with torch.no_grad():

app/services/s3_service.py

@@ -1,27 +1,15 @@
 
-import boto3
-from botocore.config import Config as BotoConfig
-from pathlib import Path
 import requests
-
-from app.core.config import settings
+from io import BytesIO
 
 class S3Service:
     def __init__(self):
-        self.client = boto3.client(
-            "s3",
-            region_name=settings.S3_REGION,
-            config=BotoConfig(
-                retries={"max_attempts": 5, "mode": "standard"},
-                read_timeout=30,
-                connect_timeout=5,
-            ),
-        )
+        pass
 
-    def download_file_from_presigned_url(self, presigned_url: str, destination: Path):
+    def download_file_from_presigned_url(self, presigned_url: str) -> BytesIO:
         response = requests.get(presigned_url)
         response.raise_for_status()
-        with open(destination, "wb") as f:
-            f.write(response.content)
+        # response의 content를 BytesIO로 감싸 반환
+        return BytesIO(response.content)
 
 s3_service = S3Service()