오탈자 처리

app/database/session.py

@@ -126,7 +126,9 @@ async def get_session() -> AsyncGenerator[AsyncSession, None]:
         try:
             yield session
         except Exception as e:
+            import traceback
             await session.rollback()
+            logger.error(traceback.format_exc())
             logger.error(
                 f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, "
                 f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms"

app/lyric/api/routers/v1/lyric.py

@@ -292,7 +292,6 @@ async def generate_lyric(
             "promotional_expression_example" : promotional_expressions[request_body.language],
             "timing_rules" : timing_rules["60s"], # 아직은 선택지 하나
         }
-        estimated_prompt = lyric_prompt.build_prompt(lyric_input_data) 
 
         step1_elapsed = (time.perf_counter() - step1_start) * 1000
         #logger.debug(f"[generate_lyric] Step 1 완료 - 프롬프트 {len(prompt)}자 ({step1_elapsed:.1f}ms)")

app/social/api/routers/v1/__init__.py

@@ -4,5 +4,5 @@ Social API Routers v1
 
 from app.social.api.routers.v1.oauth import router as oauth_router
 from app.social.api.routers.v1.upload import router as upload_router
-
-__all__ = ["oauth_router", "upload_router"]
+from app.social.api.routers.v1.seo import router as seo_router
+__all__ = ["oauth_router", "upload_router", "seo_router"]

app/social/api/routers/v1/seo.py

@@ -0,0 +1,131 @@
+
+import logging, json
+
+from redis.asyncio import Redis
+
+from config import social_oauth_settings, db_settings
+from app.social.constants import YOUTUBE_SEO_HASH
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.social.schemas import (
+    YoutubeDescriptionRequest,
+    YoutubeDescriptionResponse, 
+)
+
+from app.database.session import get_session
+from app.user.dependencies import get_current_user
+from app.user.models import User
+from app.home.models import Project, MarketingIntel
+from fastapi import APIRouter, BackgroundTasks, Depends, Query
+from fastapi import HTTPException, status
+from app.utils.prompts.prompts import yt_upload_prompt
+from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
+
+redis_seo_client = Redis(
+    host=db_settings.REDIS_HOST,
+    port=db_settings.REDIS_PORT,
+    db=0,
+    decode_responses=True,
+)
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/seo", tags=["Social SEO"])
+
+@router.post(
+    "/youtube",
+    response_model=YoutubeDescriptionResponse,
+    summary="유튜브 SEO descrption 생성",
+    description="유튜브 업로드 시 사용할 descrption을 SEO 적용하여 생성",
+)
+async def youtube_seo_description(
+        request_body: YoutubeDescriptionRequest,
+        current_user: User = Depends(get_current_user),
+        session: AsyncSession = Depends(get_session),
+        ) -> YoutubeDescriptionResponse:
+    
+    # TODO : 나중에 Session Task_id 검증 미들웨어 만들면 추가해주세요.
+
+    logger.info(
+        f"[youtube_seo_description] Try Cache - user: {current_user.user_uuid} / task_id : {request_body.task_id}"
+    )
+    cached = await get_yt_seo_in_redis(request_body.task_id)
+    if cached: # redis hit
+        return cached
+    
+    logger.info(
+        f"[youtube_seo_description] Cache miss - user: {current_user.user_uuid} "
+    )
+    updated_seo = await make_youtube_seo_description(request_body.task_id, current_user, session)
+    await set_yt_seo_in_redis(request_body.task_id, updated_seo)
+
+    return updated_seo
+
+async def make_youtube_seo_description(
+        task_id: str,
+        current_user: User,
+        session: AsyncSession,
+    ) -> YoutubeDescriptionResponse:
+    
+    logger.info(
+        f"[make_youtube_seo_description] START - user: {current_user.user_uuid} "
+    )
+    try:
+        project_query = await session.execute(
+            select(Project)
+            .where(
+                Project.task_id == task_id,
+                Project.user_uuid == current_user.user_uuid)
+               .order_by(Project.created_at.desc())
+            .limit(1)
+        )
+
+        project = project_query.scalar_one_or_none()
+        marketing_query = await session.execute(
+            select(MarketingIntel)
+            .where(MarketingIntel.id == project.marketing_intelligence)
+        )
+        marketing_intelligence = marketing_query.scalar_one_or_none()
+
+        hashtags = marketing_intelligence.intel_result["target_keywords"]
+        
+        yt_seo_input_data = {
+            "customer_name" : project.store_name,
+            "detail_region_info" : project.detail_region_info,
+            "marketing_intelligence_summary" : json.dumps(marketing_intelligence.intel_result, ensure_ascii=False),
+            "language" : project.language,
+            "target_keywords" : hashtags
+        }
+        chatgpt = ChatgptService()
+        yt_seo_output = await chatgpt.generate_structured_output(yt_upload_prompt, yt_seo_input_data)
+        result_dict = {
+            "title" : yt_seo_output.title,
+            "description" : yt_seo_output.description,
+            "keywords": hashtags
+        }
+
+        result = YoutubeDescriptionResponse(**result_dict)
+        return result
+
+    except Exception as e:
+        logger.error(f"[youtube_seo_description] EXCEPTION - error: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
+        )
+    
+
+async def get_yt_seo_in_redis(task_id:str) -> YoutubeDescriptionResponse | None:
+    field = f"task_id:{task_id}"
+    yt_seo_info = await redis_seo_client.hget(YOUTUBE_SEO_HASH, field)
+    if yt_seo_info:
+        yt_seo = json.loads(yt_seo_info)
+    else:
+        return None
+    return YoutubeDescriptionResponse(**yt_seo)
+
+async def set_yt_seo_in_redis(task_id:str, yt_seo : YoutubeDescriptionResponse) -> None:
+    field = f"task_id:{task_id}"
+    yt_seo_info = json.dumps(yt_seo.model_dump(), ensure_ascii=False)
+    await redis_seo_client.hsetex(YOUTUBE_SEO_HASH, field, yt_seo_info, ex=3600)
+    return
+

app/social/api/routers/v1/upload.py

@@ -4,10 +4,11 @@
 소셜 미디어 영상 업로드 관련 엔드포인트를 제공합니다.
 """
 
-import logging
+import logging, json
 from typing import Optional
 
 from fastapi import APIRouter, BackgroundTasks, Depends, Query
+from fastapi import HTTPException, status
 from sqlalchemy import select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -403,16 +404,12 @@ async def cancel_upload(
     upload = result.scalar_one_or_none()
 
     if not upload:
-        from fastapi import HTTPException, status
-
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail="업로드 정보를 찾을 수 없습니다.",
         )
 
     if upload.status != UploadStatus.PENDING.value:
-        from fastapi import HTTPException, status
-
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="대기 중인 업로드만 취소할 수 있습니다.",
@@ -424,4 +421,4 @@ async def cancel_upload(
     return MessageResponse(
         success=True,
         message="업로드가 취소되었습니다.",
-    )
+    )

app/social/constants.py

@@ -94,6 +94,8 @@ YOUTUBE_SCOPES = [
     "https://www.googleapis.com/auth/userinfo.profile",  # 사용자 프로필
 ]
 
+YOUTUBE_SEO_HASH = "SEO_Describtion_YT"
+
 # =============================================================================
 # Instagram/Facebook OAuth Scopes (추후 구현)
 # =============================================================================

app/social/schemas.py

@@ -276,6 +276,32 @@ class SocialUploadHistoryResponse(BaseModel):
         }
     )
 
+class YoutubeDescriptionRequest(BaseModel):
+    """유튜브 SEO Description 제안 (자동완성) Request 모델"""
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "task_id" :  "019c739f-65fc-7d15-8c88-b31be00e588e"
+            }
+        }
+    )
+    task_id: str = Field(..., description="작업 고유 식별자")
+
+class YoutubeDescriptionResponse(BaseModel):
+    """유튜브 SEO Description 제안 (자동완성) Response 모델"""
+    title:str = Field(..., description="유튜브 영상 제목 - SEO/AEO 최적화")
+    description : str = Field(..., description="제안된 유튜브 SEO Description")
+    keywords : list[str] = Field(..., description="해시태그 리스트")
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "title" : "여기에 더미 타이틀",
+                "description": "여기에 더미 텍스트",
+                "keywords": ["여기에", "더미", "해시태그"]
+            }
+        }
+    )
 
 # =============================================================================
 # 공통 응답 스키마

app/utils/chatgpt_prompt.py

@@ -82,7 +82,7 @@ class ChatgptService:
         self,
         prompt : Prompt,
         input_data : dict,
-    ) -> str:
+    ) -> BaseModel:
         prompt_text = prompt.build_prompt(input_data)
 
         logger.debug(f"[ChatgptService] Generated Prompt (length: {len(prompt_text)})")

app/utils/prompts/prompts.py

@@ -52,6 +52,14 @@ lyric_prompt = Prompt(
     prompt_model = prompt_settings.LYRIC_PROMPT_MODEL
 )
 
+yt_upload_prompt = Prompt(
+    prompt_template_path=os.path.join(prompt_settings.PROMPT_FOLDER_ROOT, prompt_settings.YOUTUBE_PROMPT_FILE_NAME),
+    prompt_input_class = YTUploadPromptInput,
+    prompt_output_class = YTUploadPromptOutput,
+    prompt_model = prompt_settings.YOUTUBE_PROMPT_MODEL
+)
+
 def reload_all_prompt():
     marketing_prompt._reload_prompt()
-    lyric_prompt._reload_prompt()
+    lyric_prompt._reload_prompt()
+    yt_upload_prompt._reload_prompt()

app/utils/prompts/schemas/__init__.py

@@ -1,2 +1,3 @@
 from .lyric import LyricPromptInput, LyricPromptOutput
-from .marketing import MarketingPromptInput, MarketingPromptOutput
+from .marketing import MarketingPromptInput, MarketingPromptOutput
+from .youtube import YTUploadPromptInput, YTUploadPromptOutput

app/utils/prompts/schemas/youtube.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional
+
+# Input 정의
+class YTUploadPromptInput(BaseModel):
+    customer_name : str = Field(..., description = "마케팅 대상 사업체 이름")
+    detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세")
+    marketing_intelligence_summary : Optional[str] = Field(None, description = "마케팅 분석 정보 보고서")
+    language : str= Field(..., description = "영상 언어")
+    target_keywords: List[str] = Field(..., description="태그 키워드 리스트")
+
+# Output 정의
+class YTUploadPromptOutput(BaseModel):
+    title:str = Field(..., description="유튜브 영상 제목 - SEO/AEO 최적화")
+    description: str = Field(..., description = "유튜브 영상 설명 - SEO/AEO 최적화")
+    

config.py

@@ -184,6 +184,8 @@ class PromptSettings(BaseSettings):
     LYRIC_PROMPT_FILE_NAME : str = Field(default="lyric_prompt.txt")
     LYRIC_PROMPT_MODEL : str = Field(default="gpt-5-mini")
 
+    YOUTUBE_PROMPT_FILE_NAME : str = Field(default="yt_upload_prompt.txt")
+    YOUTUBE_PROMPT_MODEL : str = Field(default="gpt-5-mini")
     model_config = _base_config
 
 
@@ -598,9 +600,9 @@ class SocialUploadSettings(BaseSettings):
 
 
 prj_settings = ProjectSettings()
+cors_settings = CORSSettings()
 apikey_settings = APIKeySettings()
 db_settings = DatabaseSettings()
-cors_settings = CORSSettings()
 crawler_settings = CrawlerSettings()
 naver_api_settings = NaverAPISettings()
 azure_blob_settings = AzureBlobSettings()

main.py

@@ -21,6 +21,7 @@ from app.sns.api.routers.v1.sns import router as sns_router
 from app.video.api.routers.v1.video import router as video_router
 from app.social.api.routers.v1.oauth import router as social_oauth_router
 from app.social.api.routers.v1.upload import router as social_upload_router
+from app.social.api.routers.v1.seo import router as social_seo_router
 from app.utils.cors import CustomCORSMiddleware
 from config import prj_settings
 
@@ -360,6 +361,7 @@ app.include_router(video_router)
 app.include_router(archive_router)  # Archive API 라우터 추가
 app.include_router(social_oauth_router, prefix="/social")  # Social OAuth 라우터 추가
 app.include_router(social_upload_router, prefix="/social")  # Social Upload 라우터 추가
+app.include_router(social_seo_router, prefix="/social")  # Social Upload 라우터 추가
 app.include_router(sns_router)  # SNS API 라우터 추가
 
 # DEBUG 모드에서만 테스트 라우터 등록