castad
/
o2o-castad-backend
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

유튜브 자동 seo description redis 적용

main
jaehwang 2026-02-20 08:19:45 +00:00
parent b3354d4ad1
commit 172586e699
7 changed files with 141 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/database/session.py
View File

@ -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"

4
app/social/api/routers/v1/__init__.py
View File

@ -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"]

131
app/social/api/routers/v1/seo.py Normal file
View File

@ -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="/autodescription", tags=["Social SEO"])
@router.post(
"",
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_inteligence)
)
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

64
app/social/api/routers/v1/upload.py
View File

@ -23,17 +23,12 @@ from app.social.schemas import (
SocialUploadRequest,
SocialUploadResponse,
SocialUploadStatusResponse,
YoutubeDescriptionRequest,
YoutubeDescriptionResponse,
)
from app.social.services import social_account_service
from app.social.worker.upload_task import process_social_upload
from app.user.dependencies import get_current_user
from app.user.models import User
from app.home.models import Project, MarketingIntel
from app.video.models import Video
from app.utils.prompts.prompts import yt_upload_prompt
from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
logger = logging.getLogger(__name__)
@ -427,62 +422,3 @@ async def cancel_upload(
success=True,
message="업로드가 취소되었습니다.",
)
@router.post(
"/autodescription",
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:
logger.info(
f"[youtube_seo_description] START - user: {current_user.user_uuid} "
)
try:
task_id = request_body.task_id
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_inteligence)
)
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)
response = YoutubeDescriptionResponse(
title=yt_seo_output.title,
description=yt_seo_output.description,
keywords = hashtags
)
return response
except Exception as e:
logger.error(f"[youtube_seo_description] EXCEPTION - error: {e}")
raise HTTPException(
status_code=500,
detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
)

2
app/social/constants.py
View File

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

2
config.py
View File

@ -600,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()

2
main.py
View File

@ -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 모드에서만 테스트 라우터 등록