o2o-castad-backend/app/social/services/seo_service.py

"""
유튜브 SEO 서비스

ADO2 영상은 제목/설명/해시태그를 video 테이블에 저장합니다.
썰박스는 별도 프롬프트로 생성합니다.
"""

import json
import logging

from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.home.models import MarketingIntel, Project
from app.social.schemas import YoutubeDescriptionResponse
from app.social.services.sns_metadata import apply_sns_metadata, has_stored_sns_metadata
from app.ssulbox.models import SsulContent
from app.user.models import User
from app.video.models import Video

logger = logging.getLogger(__name__)


class SeoService:
    """유튜브 SEO 비즈니스 로직 서비스"""

    async def get_youtube_seo_description(
        self,
        current_user: User,
        session: AsyncSession,
        video_id: int | None = None,
        task_id: str | None = None,
    ) -> YoutubeDescriptionResponse:
        """저장된 SNS 메타데이터를 반환하거나, 없으면 생성 후 video에 저장합니다."""
        video = None
        if video_id is not None:
            video = await self._get_owned_video(video_id, current_user.user_uuid, session)
        elif task_id:
            video = await self._get_owned_video_by_task(
                task_id, current_user.user_uuid, session
            )

        if video is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="해당하는 영상을 찾을 수 없습니다.",
            )

        logger.info(
            f"[SEO_SERVICE] Load metadata - user: {current_user.user_uuid} / video_id: {video.id}"
        )

        if has_stored_sns_metadata(video):
            return self._response_from_row(video)

        result = await self.generate_and_save_for_video(video.id, session)
        if result is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"video_id '{video.id}'에 해당하는 영상을 찾을 수 없습니다.",
            )
        await session.commit()
        return result

    async def get_ssul_seo(
        self,
        content_id: int,
        current_user: User,
        session: AsyncSession,
    ) -> YoutubeDescriptionResponse:
        """썰박스 콘텐츠용 SEO 생성 — ADO2 와 다른 프롬프트(시트 ssul_upload)를 쓴다."""
        from app.ssulbox.constants import SCENARIO_NAMES
        from app.utils.prompts.chatgpt_prompt import ChatgptService
        from app.utils.prompts.prompts import get_ssul_upload_prompt

        try:
            content = (
                await session.execute(
                    select(SsulContent).where(
                        SsulContent.id == 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="콘텐츠를 찾을 수 없습니다."
                )

            if has_stored_sns_metadata(content):
                return self._response_from_row(content)

            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
            )
            result = YoutubeDescriptionResponse(
                title=out.title,
                description=out.description,
                keywords=out.keywords,
            )
            apply_sns_metadata(content, result.title, result.description, result.keywords)
            await session.commit()
            logger.info(f"[SEO_SERVICE] Saved ssul metadata - content_id: {content_id}")
            return result

        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_and_save_for_video(
        self,
        video_id: int,
        session: AsyncSession,
    ) -> YoutubeDescriptionResponse | None:
        """GPT로 SEO를 생성해 지정한 video 행에 저장합니다. 워커/온디맨드 공용."""
        video_result = await session.execute(select(Video).where(Video.id == video_id))
        video = video_result.scalar_one_or_none()
        if video is None:
            logger.warning(f"[SEO_SERVICE] Video NOT FOUND - video_id: {video_id}")
            return None

        if has_stored_sns_metadata(video):
            return self._response_from_row(video)

        result = await self._generate_seo_description(video.task_id, session)
        apply_sns_metadata(video, result.title, result.description, result.keywords)
        await session.flush()
        logger.info(f"[SEO_SERVICE] Saved metadata - video_id: {video_id}")
        return result

    async def _get_owned_video(
        self,
        video_id: int,
        user_uuid: str,
        session: AsyncSession,
    ) -> Video | None:
        result = await session.execute(
            select(Video)
            .join(Project, Project.id == Video.project_id)
            .where(
                Video.id == video_id,
                Project.user_uuid == user_uuid,
                Video.is_deleted.is_(False),
            )
        )
        return result.scalar_one_or_none()

    async def _get_owned_video_by_task(
        self,
        task_id: str,
        user_uuid: str,
        session: AsyncSession,
    ) -> Video | None:
        result = await session.execute(
            select(Video)
            .join(Project, Project.id == Video.project_id)
            .where(
                Video.task_id == task_id,
                Project.user_uuid == user_uuid,
                Video.is_deleted.is_(False),
            )
            .order_by(Video.created_at.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def _generate_seo_description(
        self,
        task_id: str,
        session: AsyncSession,
    ) -> YoutubeDescriptionResponse:
        """GPT를 사용하여 SEO description 생성"""
        from app.utils.prompts.chatgpt_prompt import ChatgptService
        from app.utils.prompts.prompts import yt_upload_prompt

        logger.info(f"[SEO_SERVICE] Generating SEO - task_id: {task_id}")

        try:
            project_result = await session.execute(
                select(Project)
                .where(Project.task_id == task_id)
                .order_by(Project.created_at.desc())
                .limit(1)
            )
            project = project_result.scalar_one_or_none()
            if project is None:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"task_id '{task_id}'에 해당하는 Project를 찾을 수 없습니다.",
                )

            marketing_result = await session.execute(
                select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
            )
            marketing_intelligence = marketing_result.scalar_one_or_none()
            if marketing_intelligence is None:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail="마케팅 인텔리전스를 찾을 수 없습니다.",
                )

            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 "",
            }

            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 HTTPException:
            raise
        except Exception as e:
            logger.error(f"[SEO_SERVICE] EXCEPTION - error: {e}")
            raise HTTPException(
                status_code=500,
                detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
            )

    def _response_from_row(self, row) -> YoutubeDescriptionResponse:
        return YoutubeDescriptionResponse(
            title=row.title or "",
            description=row.description or "",
            keywords=list(row.hashtags or []),
        )


seo_service = SeoService()