diff --git a/apps/pre-processing-service/app/service/blog/blog_create_service.py b/apps/pre-processing-service/app/service/blog/blog_create_service.py new file mode 100644 index 00000000..29ce12b7 --- /dev/null +++ b/apps/pre-processing-service/app/service/blog/blog_create_service.py @@ -0,0 +1,345 @@ +import json +import logging +import os +from datetime import datetime +from typing import Dict, List, Optional, Any + +from openai import OpenAI +from dotenv import load_dotenv + +from app.model.schemas import RequestBlogCreate +from app.errors.BlogPostingException import * + +# 환경변수 로드 +load_dotenv(".env.dev") + + +class BlogContentService: + """RAG를 사용한 블로그 콘텐츠 생성 전용 서비스""" + + def __init__(self): + # OpenAI API 키 설정 + self.openai_api_key = os.getenv("OPENAI_API_KEY") + if not self.openai_api_key: + raise ValueError("OPENAI_API_KEY가 .env.dev 파일에 설정되지 않았습니다.") + + # 인스턴스 레벨에서 클라이언트 생성 + self.client = OpenAI(api_key=self.openai_api_key) + logging.basicConfig(level=logging.INFO) + self.logger = logging.getLogger(__name__) + + def generate_blog_content(self, request: RequestBlogCreate) -> Dict[str, Any]: + """ + 요청 데이터를 기반으로 블로그 콘텐츠 생성 + + Args: + request: RequestBlogCreate 객체 + + Returns: + Dict: {"title": str, "content": str, "tags": List[str]} 형태의 결과 + """ + try: + # 1. 콘텐츠 정보 정리 + content_context = self._prepare_content_context(request) + + # 2. 프롬프트 생성 + prompt = self._create_content_prompt(content_context, request) + + # 3. GPT를 통한 콘텐츠 생성 + generated_content = self._generate_with_openai(prompt) + + # 4. 콘텐츠 파싱 및 구조화 + return self._parse_generated_content(generated_content, request) + + except Exception as e: + self.logger.error(f"콘텐츠 생성 실패: {e}") + return self._create_fallback_content(request) + + def _prepare_content_context(self, request: RequestBlogCreate) -> str: + """요청 데이터를 콘텐츠 생성용 컨텍스트로 변환""" + context_parts = [] + + # 키워드 정보 추가 + if request.keyword: + context_parts.append(f"주요 키워드: {request.keyword}") + + # 상품 정보 추가 + if request.product_info: + context_parts.append("\n상품 정보:") + + # 상품 기본 정보 + if request.product_info.get("title"): + context_parts.append(f"- 상품명: {request.product_info['title']}") + + if request.product_info.get("price"): + context_parts.append(f"- 가격: {request.product_info['price']:,}원") + + if request.product_info.get("rating"): + context_parts.append(f"- 평점: {request.product_info['rating']}/5.0") + + # 상품 상세 정보 + if request.product_info.get("description"): + context_parts.append(f"- 설명: {request.product_info['description']}") + + # 상품 사양 (material_info 등) + if request.product_info.get("material_info"): + context_parts.append("- 주요 사양:") + specs = request.product_info["material_info"] + if isinstance(specs, dict): + for key, value in specs.items(): + context_parts.append(f" * {key}: {value}") + + # 상품 옵션 + if request.product_info.get("options"): + options = request.product_info["options"] + context_parts.append(f"- 구매 옵션 ({len(options)}개):") + for i, option in enumerate(options[:5], 1): # 최대 5개만 + if isinstance(option, dict): + option_name = option.get("name", f"옵션 {i}") + context_parts.append(f" {i}. {option_name}") + else: + context_parts.append(f" {i}. {option}") + + # 구매 링크 + if request.product_info.get("url") or request.product_info.get( + "product_url" + ): + url = request.product_info.get("url") or request.product_info.get( + "product_url" + ) + context_parts.append(f"- 구매 링크: {url}") + + return "\n".join(context_parts) if context_parts else "키워드 기반 콘텐츠 생성" + + def _create_content_prompt(self, context: str, request: RequestBlogCreate) -> str: + """콘텐츠 생성용 프롬프트 생성""" + + # 기본 키워드가 없으면 상품 제목에서 추출 + main_keyword = request.keyword + if ( + not main_keyword + and request.product_info + and request.product_info.get("title") + ): + main_keyword = request.product_info["title"] + + prompt = f""" +다음 정보를 바탕으로 매력적인 블로그 포스트를 작성해주세요. + +정보: +{context} + +작성 가이드라인: +- 스타일: 친근하면서도 신뢰할 수 있는, 정보 제공 중심 +- 길이: 1200자 내외의 적당한 길이 +- 톤: 독자의 관심을 끄는 자연스러운 어조 + +작성 요구사항: +1. SEO 친화적이고 클릭하고 싶은 매력적인 제목 +2. 독자의 관심을 끄는 도입부 +3. 핵심 특징과 장점을 구체적으로 설명 +4. 실제 사용 시나리오나 활용 팁 +5. 구매 결정에 도움이 되는 정보 + +⚠️ 주의: +- 절대로 마지막에 'HTML 구조는…' 같은 자기 평가 문장을 추가하지 마세요. +- 출력 시 ```나 ```html 같은 코드 블록 구문을 포함하지 마세요. +- 오직 HTML 태그만 사용하여 구조화된 콘텐츠를 작성해주세요. +(예:
,
{product_name}에 대한 상세한 정보를 소개합니다.
+ +판매가: {request.product_info['price']:,}원
\n" + + if request.product_info.get("material_info"): + content += "신중한 검토를 통해 만족스러운 구매 결정을 내리시기 바랍니다.
+""" + + return { + "title": title, + "content": content, + "tags": self._generate_tags(request), + } + + +# if __name__ == '__main__': +# # 테스트용 요청 데이터 +# test_request = RequestBlogCreate( +# keyword="아이폰 케이스", +# product_info={ +# "title": "아이폰 15 프로 투명 케이스", +# "price": 29900, +# "rating": 4.8, +# "description": "9H 강화 보호 기능을 제공하는 투명 케이스", +# "material_info": { +# "소재": "TPU + PC", +# "두께": "1.2mm", +# "색상": "투명", +# "호환성": "아이폰 15 Pro" +# }, +# "options": [ +# {"name": "투명"}, +# {"name": "반투명"}, +# {"name": "블랙"} +# ], +# "url": "https://example.com/iphone-case" +# } +# ) +# +# # 서비스 실행 +# service = BlogContentService() +# print("=== 블로그 콘텐츠 생성 테스트 ===") +# print(f"키워드: {test_request.keyword}") +# print(f"상품: {test_request.product_info['title']}") +# print("\n--- 생성 시작 ---") +# +# result = service.generate_blog_content(test_request) +# +# print(f"\n=== 생성 결과 ===") +# print(f"제목: {result['title']}") +# print(f"\n태그: {', '.join(result['tags'])}") +# print(f"\n내용:\n{result['content']}") +# print(f"\n글자수: {len(result['content'])}자") diff --git a/apps/pre-processing-service/app/service/product_blog_posting_service.py b/apps/pre-processing-service/app/service/product_blog_posting_service.py deleted file mode 100644 index 129c4666..00000000 --- a/apps/pre-processing-service/app/service/product_blog_posting_service.py +++ /dev/null @@ -1,387 +0,0 @@ -import json -import logging -import os -from datetime import datetime -from typing import Dict, List, Optional, Any -from dataclasses import dataclass -from enum import Enum - -from openai import OpenAI -from dotenv import load_dotenv - -from app.service.blog.blogger_blog_post_adapter import BloggerBlogPostAdapter -from app.errors.BlogPostingException import * - -# 환경변수 로드 -load_dotenv(".env.dev") - -client = OpenAI() - - -class PostingStatus(Enum): - PENDING = "pending" - PROCESSING = "processing" - SUCCESS = "success" - FAILED = "failed" - RETRY = "retry" - - -@dataclass -class ProductData: - """크롤링된 상품 데이터 모델""" - - tag: str - product_url: str - title: str - price: int - rating: float - options: List[Dict[str, Any]] - material_info: Dict[str, str] - product_images: List[str] - crawled_at: str - - @classmethod - def from_dict(cls, data: Dict) -> "ProductData": - """딕셔너리에서 ProductData 객체 생성""" - product_detail = data.get("product_detail", {}) - return cls( - tag=data.get("tag", ""), - product_url=product_detail.get("url", ""), - title=product_detail.get("title", ""), - price=product_detail.get("price", 0), - rating=product_detail.get("rating", 0.0), - options=product_detail.get("options", []), - material_info=product_detail.get("material_info", {}), - product_images=product_detail.get("product_images", []), - crawled_at=data.get("crawled_at", ""), - ) - - -@dataclass -class BlogPostContent: - """생성된 블로그 포스트 콘텐츠""" - - title: str - content: str - tags: List[str] - - -@dataclass -class BlogContentRequest: - """블로그 콘텐츠 생성 요청""" - - content_style: str = "informative" # "informative", "promotional", "review" - target_keywords: List[str] = None - include_pricing: bool = True - include_specifications: bool = True - content_length: str = "medium" # "short", "medium", "long" - - -class ProductContentGenerator: - """GPT를 활용한 상품 블로그 콘텐츠 생성""" - - def __init__(self): - # 환경변수에서 OpenAI API 키 로드 - self.openai_api_key = os.getenv("OPENAI_API_KEY") - if not self.openai_api_key: - raise ValueError("OPENAI_API_KEY가 .env.dev 파일에 설정되지 않았습니다.") - - client.api_key = self.openai_api_key - - def generate_blog_content( - self, product_data: ProductData, request: BlogContentRequest - ) -> BlogPostContent: - """상품 데이터를 기반으로 블로그 콘텐츠 생성""" - - # 1. 상품 정보 정리 - product_info = self._format_product_info(product_data, request) - - # 2. 프롬프트 생성 - prompt = self._create_blog_prompt(product_info, request) - - # 3. GPT를 통한 콘텐츠 생성 - try: - - response = client.chat.completions.create( - model="gpt-4o-mini", - messages=[ - { - "role": "system", - "content": "당신은 전문적인 블로그 콘텐츠 작성자입니다. 상품 리뷰와 정보성 콘텐츠를 매력적이고 SEO 친화적으로 작성합니다.", - }, - {"role": "user", "content": prompt}, - ], - temperature=0.7, - max_tokens=2000, - ) - - generated_content = response.choices[0].message.content - - # 4. 콘텐츠 파싱 및 구조화 - return self._parse_generated_content( - generated_content, product_data, request - ) - - except Exception as e: - logging.error(f"콘텐츠 생성 실패: {e}") - return self._create_fallback_content(product_data, request) - - def _format_product_info( - self, product_data: ProductData, request: BlogContentRequest - ) -> str: - """상품 정보를 텍스트로 포맷팅""" - info_parts = [ - f"상품명: {product_data.title}", - ] - - # 가격 정보 추가 - if request.include_pricing and product_data.price: - info_parts.append(f"가격: {product_data.price:,}원") - - # 평점 정보 추가 - if product_data.rating: - info_parts.append(f"평점: {product_data.rating}/5.0") - - # 사양 정보 추가 - if request.include_specifications and product_data.material_info: - info_parts.append("\n상품 사양:") - for key, value in product_data.material_info.items(): - info_parts.append(f"- {key}: {value}") - - # 옵션 정보 추가 - if product_data.options: - info_parts.append(f"\n구매 옵션 ({len(product_data.options)}개):") - for i, option in enumerate(product_data.options[:5], 1): # 처음 5개만 - info_parts.append(f"{i}. {option.get('name', 'N/A')}") - - # 구매 링크 - if product_data.product_url: - info_parts.append(f"\n구매 링크: {product_data.product_url}") - - return "\n".join(info_parts) - - def _create_blog_prompt( - self, product_info: str, request: BlogContentRequest - ) -> str: - """블로그 작성용 프롬프트 생성""" - - # 스타일별 가이드라인 - style_guidelines = { - "informative": "객관적이고 상세한 정보 제공 중심으로, 독자가 제품을 이해할 수 있도록 전문적으로 작성", - "promotional": "제품의 장점과 매력을 강조하며, 구매 의욕을 자극할 수 있도록 매력적으로 작성", - "review": "실제 사용 경험을 바탕으로 한 솔직한 평가와 추천 중심으로 작성", - } - - # 길이별 가이드라인 - length_guidelines = { - "short": "800자 내외의 간결한 내용", - "medium": "1200자 내외의 적당한 길이", - "long": "1500자 이상의 상세한 내용", - } - - style_guide = style_guidelines.get( - request.content_style, style_guidelines["informative"] - ) - length_guide = length_guidelines.get( - request.content_length, length_guidelines["medium"] - ) - - # 키워드 정보 - keywords_text = "" - if request.target_keywords: - keywords_text = f"\n포함할 키워드: {', '.join(request.target_keywords)}" - - prompt = f""" -다음 상품 정보를 바탕으로 매력적인 블로그 포스트를 작성해주세요. - -상품 정보: -{product_info} - -작성 가이드라인: -- 스타일: {style_guide} -- 길이: {length_guide} -- 톤: 친근하면서도 신뢰할 수 있는, 정보 제공 중심{keywords_text} - -작성 요구사항: -1. SEO 친화적이고 클릭하고 싶은 매력적인 제목 -2. 독자의 관심을 끄는 도입부 -3. 상품의 핵심 특징과 장점을 구체적으로 설명 -4. 실제 사용 시나리오나 활용 팁 -5. 구매 결정에 도움이 되는 정보 - -⚠️ 주의: -- 절대로 마지막에 '이 HTML 구조는…' 같은 자기 평가 문장을 추가하지 마세요. -- 출력 시 ```나 ```html 같은 코드 블록 구문을 포함하지 마세요. -- 오직 HTML 태그만 사용하여 구조화된 콘텐츠를 작성해주세요. -(예:,
{product_data.title}에 대한 상세한 정보를 소개합니다.
- -판매가: {product_data.price:,}원
-""" - - if product_data.material_info: - content += "상품 구매는 여기에서 가능합니다.
-""" - - return BlogPostContent( - title=title, - content=content, - tags=[product_data.tag] if product_data.tag else ["상품정보"], - ) - - -class ProductBlogPostingService: - """상품 데이터를 Blogger에 포스팅하는 메인 서비스""" - - def __init__(self): - self.content_generator = ProductContentGenerator() - self.blogger_service = BloggerBlogPostAdapter() - - def post_product_to_blogger( - self, product_data: ProductData, request: BlogContentRequest - ) -> dict: - """상품 데이터를 Blogger에 포스팅""" - try: - # 1. GPT를 통한 콘텐츠 생성 - blog_content = self.content_generator.generate_blog_content( - product_data, request - ) - - # 2. Blogger에 포스팅 - self.blogger_service.post_content( - title=blog_content.title, - content=blog_content.content, - tags=blog_content.tags, - ) - - # 3. 성공 결과 반환 - return { - "status": "success", - "platform": "blogger", - "title": blog_content.title, - "tags": blog_content.tags, - "posted_at": datetime.now().isoformat(), - "product_tag": product_data.tag, - } - - except Exception as e: - logging.error(f"Blogger 포스팅 실패: {e}") - # ProductData 객체 기준으로 처리 - return { - "status": "failed", - "error": str(e), - "platform": "blogger", - "attempted_at": datetime.now().isoformat(), - "product_tag": getattr(product_data, "tag", "unknown"), - } - - # def batch_post_products(self, products_data: List[Dict], request: BlogContentRequest) -> List[Dict[str, Any]]: - # """여러 상품을 일괄 포스팅""" - # results = [] - # - # for product_data in products_data: - # result = self.post_product_to_blogger(product_data, request) - # results.append(result) - # - # # API 호출 제한을 고려한 딜레이 - # import time - # time.sleep(3) # 3초 대기 - # - # return results diff --git a/apps/pre-processing-service/app/test/test_blog_create_service.py b/apps/pre-processing-service/app/test/test_blog_create_service.py new file mode 100644 index 00000000..d32e4e9e --- /dev/null +++ b/apps/pre-processing-service/app/test/test_blog_create_service.py @@ -0,0 +1,83 @@ +import unittest +from unittest.mock import patch, MagicMock + +from app.service.blog.blog_create_service import BlogContentService +from app.model.schemas import RequestBlogCreate + + +class TestBlogContentGeneration(unittest.TestCase): + """블로그 콘텐츠 생성 핵심 로직 테스트""" + + @patch.dict("os.environ", {"OPENAI_API_KEY": "test-key"}) + @patch("app.service.blog.blog_create_service.OpenAI") + def setUp(self, mock_openai_class): + """테스트 설정 - OpenAI Mock 적용""" + # Mock OpenAI 클라이언트 설정 + self.mock_client = MagicMock() + mock_openai_class.return_value = self.mock_client + + # 서비스 인스턴스 생성 + self.service = BlogContentService() + + def test_generate_blog_content_success(self): + """정상적인 콘텐츠 생성 테스트""" + # Mock 응답 설정 + mock_choice = MagicMock() + mock_choice.message.content = """이 케이스는 뛰어난 보호 성능을 제공합니다.
""" + + mock_response = MagicMock() + mock_response.choices = [mock_choice] + + self.mock_client.chat.completions.create.return_value = mock_response + + # 테스트 요청 + request = RequestBlogCreate( + keyword="아이폰 케이스", + product_info={"title": "아이폰 15 투명 케이스", "price": 25000}, + ) + + # 실행 + result = self.service.generate_blog_content(request) + + # 검증 + self.assertIn("title", result) + self.assertIn("content", result) + self.assertIn("tags", result) + # 실제 파싱 로직에 따른 제목 검증 (키워드가 제목에 포함되지 않아 기본 제목 생성됨) + self.assertEqual( + result["title"], "아이폰 15 투명 케이스 - 아이폰 케이스 완벽 가이드" + ) + self.assertIn("