diff --git a/blog-lab/app/content_generator.py b/blog-lab/app/content_generator.py index 255d80d..ea5de49 100644 --- a/blog-lab/app/content_generator.py +++ b/blog-lab/app/content_generator.py @@ -58,7 +58,36 @@ def generate_trend_brief(analysis: Dict[str, Any]) -> str: return _call_claude(prompt) -def generate_blog_post(analysis: Dict[str, Any], trend_brief: str) -> Dict[str, str]: +def _parse_blog_json(raw: str, keyword: str) -> Dict[str, str]: + """Claude 응답에서 블로그 JSON을 파싱.""" + try: + text = raw.strip() + if text.startswith("```"): + lines = text.split("\n") + lines = [l for l in lines if not l.strip().startswith("```")] + text = "\n".join(lines) + result = json.loads(text) + return { + "title": result.get("title", ""), + "body": result.get("body", ""), + "excerpt": result.get("excerpt", ""), + "tags": result.get("tags", []), + } + except (json.JSONDecodeError, KeyError): + logger.warning("Blog post JSON parse failed, using raw text") + return { + "title": f"{keyword} 추천 리뷰", + "body": raw, + "excerpt": raw[:200], + "tags": [keyword], + } + + +def generate_blog_post( + analysis: Dict[str, Any], + trend_brief: str, + brand_links: Optional[list] = None, +) -> Dict[str, str]: """트렌드 브리프를 바탕으로 블로그 글 작성. Returns: @@ -73,10 +102,34 @@ def generate_blog_post(analysis: Dict[str, Any], trend_brief: str) -> Dict[str, for p in analysis.get("top_products", []) ) or "없음" + # 크롤링된 블로그 본문 참고 자료 + reference_blogs_text = "" + for blog in analysis.get("top_blogs", []): + content = blog.get("content", "") + if content: + reference_blogs_text += f"\n### {blog.get('title', '제목 없음')}\n{content}\n" + if not reference_blogs_text: + reference_blogs_text = "없음" + + # 브랜드커넥트 링크 정보 + brand_products_text = "" + if brand_links: + for link in brand_links: + brand_products_text += ( + f"- 상품명: {link.get('product_name', '')}\n" + f" 설명: {link.get('description', '')}\n" + f" 링크: {link.get('url', '')}\n" + f" 배치 힌트: {link.get('placement_hint', '자연스럽게')}\n" + ) + if not brand_products_text: + brand_products_text = "없음 (제휴 링크 없이 일반 리뷰로 작성)" + prompt = template.format( keyword=analysis.get("keyword", ""), trend_brief=trend_brief, top_products=top_products_text, + reference_blogs=reference_blogs_text, + brand_products=brand_products_text, ) # 구조화된 응답을 위한 추가 지시 @@ -88,31 +141,7 @@ def generate_blog_post(analysis: Dict[str, Any], trend_brief: str) -> Dict[str, ) raw = _call_claude(prompt, max_tokens=8192) - - # JSON 파싱 시도 - try: - # ```json ... ``` 블록 제거 - text = raw.strip() - if text.startswith("```"): - lines = text.split("\n") - lines = [l for l in lines if not l.strip().startswith("```")] - text = "\n".join(lines) - result = json.loads(text) - return { - "title": result.get("title", ""), - "body": result.get("body", ""), - "excerpt": result.get("excerpt", ""), - "tags": result.get("tags", []), - } - except (json.JSONDecodeError, KeyError): - # JSON 파싱 실패 시 원본 텍스트를 body로 - logger.warning("Blog post JSON parse failed, using raw text") - return { - "title": f"{analysis.get('keyword', '')} 추천 리뷰", - "body": raw, - "excerpt": raw[:200], - "tags": [analysis.get("keyword", "")], - } + return _parse_blog_json(raw, analysis.get("keyword", "")) def regenerate_blog_post( @@ -128,7 +157,7 @@ def regenerate_blog_post( f"이전에 작성한 글:\n{previous_body[:3000]}\n\n" f"리뷰어 피드백:\n{feedback}\n\n" "위 피드백을 반영하여 글을 개선해주세요.\n" - "작성 규칙: 1인칭 체험기, 1,500자 이상, 자연스러운 구어체, " + "작성 규칙: 1인칭 체험기, 2,000자 이상, 자연스러운 구어체, " "제품 비교표 포함, 광고 고지 문구 포함.\n" "HTML 형식으로 작성하되, 네이버 블로그에서 바로 붙여넣기 가능한 형태로.\n\n" "---\n" @@ -136,27 +165,5 @@ def regenerate_blog_post( '{"title": "블로그 제목", "body": "HTML 본문", "excerpt": "2줄 요약", ' '"tags": ["태그1", "태그2", ...]}' ) - raw = _call_claude(prompt, max_tokens=8192) - - try: - text = raw.strip() - if text.startswith("```"): - lines = text.split("\n") - lines = [l for l in lines if not l.strip().startswith("```")] - text = "\n".join(lines) - result = json.loads(text) - return { - "title": result.get("title", ""), - "body": result.get("body", ""), - "excerpt": result.get("excerpt", ""), - "tags": result.get("tags", []), - } - except (json.JSONDecodeError, KeyError): - logger.warning("Regenerate JSON parse failed, using raw text") - return { - "title": f"{analysis.get('keyword', '')} 추천 리뷰 (개선)", - "body": raw, - "excerpt": raw[:200], - "tags": [analysis.get("keyword", "")], - } + return _parse_blog_json(raw, analysis.get("keyword", "")) diff --git a/blog-lab/app/db.py b/blog-lab/app/db.py index 0d45559..a9f5e2c 100644 --- a/blog-lab/app/db.py +++ b/blog-lab/app/db.py @@ -120,6 +120,7 @@ def init_db() -> None: # 기본 프롬프트 템플릿 시딩 (존재하지 않을 때만) _seed_templates(conn) + _migrate_templates(conn) def _seed_templates(conn: sqlite3.Connection) -> None: @@ -206,6 +207,38 @@ def _seed_templates(conn: sqlite3.Connection) -> None: ) +def _migrate_templates(conn: sqlite3.Connection) -> None: + """기존 템플릿을 최신 버전으로 업데이트.""" + new_blog_write = ( + "당신은 네이버 블로그에서 월 100만 이상 수익을 올리는 전문 블로거입니다.\n" + "아래 브리프와 참고 자료를 바탕으로 블로그 글을 작성하세요.\n\n" + "키워드: {keyword}\n" + "트렌드 브리프: {trend_brief}\n\n" + "=== 상위 블로그 참고 자료 ===\n" + "{reference_blogs}\n\n" + "=== 상위 상품 정보 ===\n" + "{top_products}\n\n" + "=== 제휴 상품 (브랜드커넥트 링크) ===\n" + "{brand_products}\n\n" + "작성 규칙:\n" + "- 1인칭 체험기 형식 (\"제가 직접 써봤는데요\")\n" + "- 2,000자 이상\n" + "- 자연스러운 구어체 (네이버 블로그 톤)\n" + "- 상위 블로그 참고하되 표절 금지 (자신만의 시각으로 재구성)\n" + "- 제품 비교표 포함 (HTML 테이블)\n" + "- 장단점 솔직하게 작성\n" + "- 제휴 상품이 있으면 자연스럽게 체험 맥락에 녹여서 작성\n" + "- 제휴 링크는 태그로 자연스럽게 삽입\n" + "- 추천 매트릭스 (가성비/품질/디자인 기준)\n" + "- 자연스러운 CTA (구매 링크 유도)\n\n" + "HTML 형식으로 작성하되, 네이버 블로그에서 바로 붙여넣기 가능한 형태로 만들어주세요." + ) + conn.execute( + "UPDATE prompt_templates SET template = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE name = 'blog_write'", + (new_blog_write,), + ) + + # ── keyword_analyses CRUD ──────────────────────────────────────────────────── def _ka_row_to_dict(r) -> Dict[str, Any]: diff --git a/blog-lab/app/main.py b/blog-lab/app/main.py index 318fada..b84e506 100644 --- a/blog-lab/app/main.py +++ b/blog-lab/app/main.py @@ -16,6 +16,7 @@ from .db import ( get_dashboard_stats, get_task, create_task, update_task, add_brand_link, get_brand_links, update_brand_link, delete_brand_link, + link_brand_links_to_post, ) from .naver_search import analyze_keyword_with_crawling from .content_generator import generate_trend_brief, generate_blog_post, regenerate_blog_post @@ -144,11 +145,14 @@ def _run_generate(task_id: str, keyword_id: int): update_task(task_id, "failed", 0, "", error="키워드 분석 결과를 찾을 수 없습니다") return + # 연결된 브랜드커넥트 링크 조회 + brand_links = get_brand_links(keyword_id=keyword_id) + update_task(task_id, "processing", 20, "트렌드 브리프 생성 중...") trend_brief = generate_trend_brief(analysis) update_task(task_id, "processing", 60, "블로그 글 작성 중...") - post_data = generate_blog_post(analysis, trend_brief) + post_data = generate_blog_post(analysis, trend_brief, brand_links=brand_links) update_task(task_id, "processing", 90, "저장 중...") saved = add_post({ @@ -161,6 +165,9 @@ def _run_generate(task_id: str, keyword_id: int): "trend_brief": trend_brief, }) + # keyword_id에 연결된 링크를 post_id에도 연결 + link_brand_links_to_post(keyword_id=keyword_id, post_id=saved["id"]) + update_task(task_id, "succeeded", 100, "글 생성 완료", result_id=saved["id"]) except Exception as e: logger.exception("Generate failed for keyword_id=%s", keyword_id) diff --git a/blog-lab/tests/test_writer.py b/blog-lab/tests/test_writer.py new file mode 100644 index 0000000..5638b81 --- /dev/null +++ b/blog-lab/tests/test_writer.py @@ -0,0 +1,86 @@ +"""작가 단계 테스트 -- 크롤링 본문 + 링크 참조 글 생성.""" +import json +import pytest +from unittest.mock import patch + + +def test_generate_blog_post_includes_crawled_content(): + """크롤링 본문이 프롬프트에 포함되는지 확인.""" + from app.content_generator import generate_blog_post + + analysis = { + "keyword": "무선 이어폰", + "top_products": [{"title": "에어팟", "lprice": 200000, "mallName": "애플"}], + "top_blogs": [ + {"title": "에어팟 리뷰", "content": "에어팟을 한 달간 써봤는데 음질이 정말 좋았습니다."}, + ], + } + + mock_response = json.dumps({ + "title": "무선 이어폰 추천", + "body": "

본문

", + "excerpt": "요약", + "tags": ["이어폰"], + }) + + with patch("app.content_generator._call_claude", return_value=mock_response) as mock_call, \ + patch("app.content_generator.get_template", return_value=( + "키워드: {keyword}\n참고 블로그:\n{reference_blogs}\n상품: {top_products}\n링크 상품: {brand_products}" + )): + result = generate_blog_post(analysis, "트렌드 브리프", brand_links=[]) + + prompt_used = mock_call.call_args[0][0] + assert "에어팟을 한 달간 써봤는데" in prompt_used + assert result["title"] == "무선 이어폰 추천" + + +def test_generate_blog_post_includes_brand_links(): + """브랜드커넥트 링크 정보가 프롬프트에 포함되는지 확인.""" + from app.content_generator import generate_blog_post + + analysis = {"keyword": "무선 이어폰", "top_products": [], "top_blogs": []} + brand_links = [ + {"url": "https://link.coupang.com/abc", "product_name": "삼성 버즈3", + "description": "노이즈캔슬링 지원", "placement_hint": "본문 중간"}, + ] + + mock_response = json.dumps({ + "title": "제목", "body": "

본문

", "excerpt": "요약", "tags": ["태그"], + }) + + with patch("app.content_generator._call_claude", return_value=mock_response) as mock_call, \ + patch("app.content_generator.get_template", return_value=( + "키워드: {keyword}\n참고 블로그:\n{reference_blogs}\n상품: {top_products}\n링크 상품: {brand_products}" + )): + result = generate_blog_post(analysis, "트렌드 브리프", brand_links=brand_links) + + prompt_used = mock_call.call_args[0][0] + assert "삼성 버즈3" in prompt_used + assert "노이즈캔슬링 지원" in prompt_used + + +def test_generate_blog_post_works_without_links(): + """링크 없이도 정상 동작.""" + from app.content_generator import generate_blog_post + + analysis = {"keyword": "테스트", "top_products": [], "top_blogs": []} + mock_response = json.dumps({ + "title": "제목", "body": "

본문

", "excerpt": "요약", "tags": ["태그"], + }) + + with patch("app.content_generator._call_claude", return_value=mock_response), \ + patch("app.content_generator.get_template", return_value=( + "키워드: {keyword}\n참고 블로그:\n{reference_blogs}\n상품: {top_products}\n링크 상품: {brand_products}" + )): + result = generate_blog_post(analysis, "브리프") + + assert result["title"] == "제목" + + +def test_parse_blog_json_fallback(): + """JSON 파싱 실패 시 원본 텍스트를 body로 사용.""" + from app.content_generator import _parse_blog_json + + result = _parse_blog_json("잘못된 JSON", "테스트 키워드") + assert result["title"] == "테스트 키워드 추천 리뷰" + assert result["body"] == "잘못된 JSON"