feat(blog-lab): 작가 단계 — 크롤링 본문 + 브랜드 링크 참조 글 생성

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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)
-
+    return _parse_blog_json(raw, analysis.get("keyword", ""))
-    # 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", "")],
-        }
 
 
 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)
-
+    return _parse_blog_json(raw, analysis.get("keyword", ""))
-    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", "")],
-        }

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"
+        "- 제휴 링크는 <a> 태그로 자연스럽게 삽입\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]:

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)

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": "<p>본문</p>",
+        "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": "<p>본문</p>", "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": "<p>본문</p>", "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"