199 lines
7.4 KiB
Python
199 lines
7.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,
|
|
content_type: str = "video",
|
|
) -> YoutubeDescriptionResponse:
|
|
"""
|
|
유튜브 SEO description 생성
|
|
|
|
Redis 캐시 확인 후 miss이면 GPT로 생성하고 캐싱.
|
|
|
|
content_type="ssul" 이면 task_id 자리에 ssul_content.id(문자열)가 온다.
|
|
캐시 키에 종류 접두를 붙인다 — ADO2 task_id(UUID)와 썰박스 id(숫자)는
|
|
형식이 달라 실제로 겹치진 않지만, 형식 우연에 기대지 않는다.
|
|
"""
|
|
cache_key = f"ssul:{task_id}" if content_type == "ssul" else task_id
|
|
logger.info(
|
|
f"[SEO_SERVICE] Try Cache - user: {current_user.user_uuid} / key: {cache_key}"
|
|
)
|
|
|
|
cached = await self._get_from_redis(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
logger.info(f"[SEO_SERVICE] Cache miss - user: {current_user.user_uuid}")
|
|
if content_type == "ssul":
|
|
result = await self._generate_ssul_seo(task_id, current_user, session)
|
|
else:
|
|
result = await self._generate_seo_description(task_id, current_user, session)
|
|
await self._set_to_redis(cache_key, result)
|
|
|
|
return result
|
|
|
|
async def _generate_ssul_seo(
|
|
self,
|
|
content_id: str,
|
|
current_user: User,
|
|
session: AsyncSession,
|
|
) -> YoutubeDescriptionResponse:
|
|
"""썰박스 콘텐츠용 SEO 생성 — ADO2 와 **다른 프롬프트**(시트 ssul_upload)를 쓴다.
|
|
|
|
ADO2 는 업장 마케팅 분석 보고서 기반의 광고 영상 SEO 지만, 썰박스는
|
|
병맛 역사 썰툰이라 톤이 완전히 다르다. 태그도 GPT 가 함께 만든다
|
|
(ADO2 처럼 재사용할 마케팅 분석 target_keywords 가 없다).
|
|
"""
|
|
from app.ssulbox.constants import SCENARIO_NAMES
|
|
from app.ssulbox.models import SsulContent
|
|
from app.utils.prompts.prompts import get_ssul_upload_prompt
|
|
|
|
try:
|
|
content = (
|
|
await session.execute(
|
|
select(SsulContent).where(
|
|
SsulContent.id == int(content_id),
|
|
SsulContent.user_uuid == current_user.user_uuid,
|
|
SsulContent.is_deleted.is_(False),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
if content is None:
|
|
raise HTTPException(
|
|
status_code=404, detail="콘텐츠를 찾을 수 없습니다."
|
|
)
|
|
|
|
input_data = {
|
|
"store_name": content.store_name or "",
|
|
"region": content.region or "",
|
|
"scenario_name": SCENARIO_NAMES.get(content.scenario, content.scenario),
|
|
}
|
|
|
|
chatgpt = ChatgptService(timeout=180)
|
|
out = await chatgpt.generate_structured_output(
|
|
get_ssul_upload_prompt(), input_data
|
|
)
|
|
|
|
return YoutubeDescriptionResponse(
|
|
title=out.title,
|
|
description=out.description,
|
|
keywords=out.keywords,
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[SEO_SERVICE] SSUL EXCEPTION - error: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"썰박스 SEO 생성에 실패했습니다. : {str(e)}",
|
|
)
|
|
|
|
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,
|
|
"industry": project.industry or "", # 크롤 시 분류해 Project에 저장한 업종 enum
|
|
}
|
|
|
|
# 업종 분기는 프롬프트 내부 {industry}로 처리하므로 단일 프롬프트 사용
|
|
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()
|