130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
"""
|
|
유튜브 SEO 서비스
|
|
|
|
SEO description 생성 및 Redis 캐싱 로직을 처리합니다.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
|
|
from fastapi import HTTPException
|
|
from redis.asyncio import Redis
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from config import db_settings
|
|
from app.home.models import MarketingIntel, Project
|
|
from app.social.constants import YOUTUBE_SEO_HASH
|
|
from app.social.schemas import YoutubeDescriptionResponse
|
|
from app.user.models import User
|
|
from app.utils.prompts.chatgpt_prompt import ChatgptService
|
|
from app.utils.prompts.prompts import yt_upload_prompt
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
redis_seo_client = Redis(
|
|
host=db_settings.REDIS_HOST,
|
|
port=db_settings.REDIS_PORT,
|
|
db=0,
|
|
decode_responses=True,
|
|
)
|
|
|
|
|
|
class SeoService:
|
|
"""유튜브 SEO 비즈니스 로직 서비스"""
|
|
|
|
async def get_youtube_seo_description(
|
|
self,
|
|
task_id: str,
|
|
current_user: User,
|
|
session: AsyncSession,
|
|
) -> YoutubeDescriptionResponse:
|
|
"""
|
|
유튜브 SEO description 생성
|
|
|
|
Redis 캐시 확인 후 miss이면 GPT로 생성하고 캐싱.
|
|
"""
|
|
logger.info(
|
|
f"[SEO_SERVICE] Try Cache - user: {current_user.user_uuid} / task_id: {task_id}"
|
|
)
|
|
|
|
cached = await self._get_from_redis(task_id)
|
|
if cached:
|
|
return cached
|
|
|
|
logger.info(f"[SEO_SERVICE] Cache miss - user: {current_user.user_uuid}")
|
|
result = await self._generate_seo_description(task_id, current_user, session)
|
|
await self._set_to_redis(task_id, result)
|
|
|
|
return result
|
|
|
|
async def _generate_seo_description(
|
|
self,
|
|
task_id: str,
|
|
current_user: User,
|
|
session: AsyncSession,
|
|
) -> YoutubeDescriptionResponse:
|
|
"""GPT를 사용하여 SEO description 생성"""
|
|
logger.info(f"[SEO_SERVICE] Generating SEO - user: {current_user.user_uuid}")
|
|
|
|
try:
|
|
project_result = 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_result.scalar_one_or_none()
|
|
|
|
marketing_result = await session.execute(
|
|
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
|
)
|
|
marketing_intelligence = marketing_result.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(timeout=180)
|
|
yt_seo_output = await chatgpt.generate_structured_output(yt_upload_prompt, yt_seo_input_data)
|
|
|
|
return YoutubeDescriptionResponse(
|
|
title=yt_seo_output.title,
|
|
description=yt_seo_output.description,
|
|
keywords=hashtags,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"[SEO_SERVICE] EXCEPTION - error: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
|
|
)
|
|
|
|
async def _get_from_redis(self, 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:
|
|
return YoutubeDescriptionResponse(**json.loads(yt_seo_info))
|
|
return None
|
|
|
|
async def _set_to_redis(self, 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.hset(YOUTUBE_SEO_HASH, field, yt_seo_info)
|
|
await redis_seo_client.expire(YOUTUBE_SEO_HASH, 3600)
|
|
|
|
|
|
seo_service = SeoService()
|