docs: CLAUDE.md에 blog-lab 파이프라인 변경사항 반영

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

CLAUDE.md

@@ -329,20 +329,24 @@ docker compose up -d
 | POST | `/api/travel/reload` | 메모리 캐시 초기화 |
 
 ### blog-lab (blog-lab/)
-- 블로그 마케팅 수익화 서비스 (키워드 분석 → AI 글 생성 → 품질 리뷰 → 포스팅 → 수익 추적)
+- 블로그 마케팅 수익화 서비스 (키워드 분석 → AI 글 생성 → 마케팅 강화 → 품질 리뷰 → 포스팅 → 수익 추적)
 - AI 엔진: Claude API (Anthropic, `claude-sonnet-4-20250514`)
-- 웹 검색: Naver Search API (블로그 + 쇼핑)
+- 웹 검색: Naver Search API (블로그 + 쇼핑) + 상위 블로그 본문 크롤링
 - DB: `/app/data/blog_marketing.db`
-- 파일 구조: `main.py`, `db.py`, `config.py`, `naver_search.py`, `content_generator.py`, `quality_reviewer.py`
+- 파일 구조: `main.py`, `db.py`, `config.py`, `naver_search.py`, `content_generator.py`, `marketer.py`, `quality_reviewer.py`, `web_crawler.py`
+
+**파이프라인**: 리서치(+크롤링) → 작가(초안) → 마케터(링크 삽입) → 평가자(6기준 60점)
+**상태 흐름**: `draft` → `marketed` → `reviewed` → `published`
 
 **blog_marketing.db 테이블**
 
 | 테이블 | 설명 |
 |--------|------|
-| `keyword_analyses` | 키워드 분석 결과 (네이버 검색 데이터 + 경쟁도/기회 점수) |
+| `keyword_analyses` | 키워드 분석 결과 (네이버 검색 데이터 + 경쟁도/기회 점수 + 크롤링 본문) |
-| `blog_posts` | 블로그 글 (draft → reviewed → published) |
+| `blog_posts` | 블로그 글 (draft → marketed → reviewed → published) |
+| `brand_links` | 브랜드커넥트 제휴 링크 (post_id/keyword_id FK) |
 | `commissions` | 포스트별 월간 클릭/구매/수익 |
-| `generation_tasks` | 비동기 작업 상태 (research/generate/review) |
+| `generation_tasks` | 비동기 작업 상태 (research/generate/market/review) |
 | `prompt_templates` | AI 프롬프트 템플릿 (DB 저장, 코드 배포 없이 수정 가능) |
 
 **blog-lab API 목록**
@@ -350,14 +354,19 @@ docker compose up -d
 | 메서드 | 경로 | 설명 |
 |--------|------|------|
 | GET | `/api/blog-marketing/status` | 서비스 상태 (API 키 설정 현황) |
-| POST | `/api/blog-marketing/research` | 키워드 분석 시작 (BackgroundTask) |
+| POST | `/api/blog-marketing/research` | 키워드 분석 시작 (+ 상위 블로그 크롤링) |
 | GET | `/api/blog-marketing/research/history` | 분석 이력 조회 |
 | GET | `/api/blog-marketing/research/{id}` | 분석 상세 조회 |
 | DELETE | `/api/blog-marketing/research/{id}` | 분석 삭제 |
 | GET | `/api/blog-marketing/task/{task_id}` | 작업 상태 폴링 |
-| POST | `/api/blog-marketing/generate` | AI 글 생성 (트렌드 브리프 + 본문) |
+| POST | `/api/blog-marketing/generate` | 작가 단계: AI 글 생성 (크롤링 참고 + 링크 반영) |
-| POST | `/api/blog-marketing/review/{post_id}` | 품질 리뷰 (5기준 × 10점) |
+| POST | `/api/blog-marketing/market/{post_id}` | 마케터 단계: 전환율 강화 + 링크 삽입 |
+| POST | `/api/blog-marketing/review/{post_id}` | 평가자 단계: 품질 리뷰 (6기준 × 10점, 42/60 통과) |
 | POST | `/api/blog-marketing/regenerate/{post_id}` | 피드백 기반 재생성 |
+| POST | `/api/blog-marketing/links` | 브랜드커넥트 링크 등록 |
+| GET | `/api/blog-marketing/links` | 링크 조회 (post_id, keyword_id 필터) |
+| PUT | `/api/blog-marketing/links/{id}` | 링크 수정 |
+| DELETE | `/api/blog-marketing/links/{id}` | 링크 삭제 |
 | GET | `/api/blog-marketing/posts` | 포스트 목록 (status 필터) |
 | GET | `/api/blog-marketing/posts/{id}` | 포스트 상세 |
 | PUT | `/api/blog-marketing/posts/{id}` | 포스트 수정 |

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

@@ -102,8 +102,25 @@ def init_db() -> None:
             )
         """)
 
+        # 브랜드커넥트 제휴 링크
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS brand_links (
+                id              INTEGER PRIMARY KEY AUTOINCREMENT,
+                post_id         INTEGER REFERENCES blog_posts(id),
+                keyword_id      INTEGER REFERENCES keyword_analyses(id),
+                url             TEXT NOT NULL,
+                product_name    TEXT NOT NULL DEFAULT '',
+                description     TEXT NOT NULL DEFAULT '',
+                placement_hint  TEXT NOT NULL DEFAULT '',
+                created_at      TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
+            )
+        """)
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_bl_post ON brand_links(post_id)")
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_bl_keyword ON brand_links(keyword_id)")
+
         # 기본 프롬프트 템플릿 시딩 (존재하지 않을 때만)
         _seed_templates(conn)
+        _migrate_templates(conn)
 
 
 def _seed_templates(conn: sqlite3.Connection) -> None:
@@ -151,18 +168,19 @@ def _seed_templates(conn: sqlite3.Connection) -> None:
         },
         {
             "name": "quality_review",
-            "description": "블로그 글 품질 리뷰 (5기준 × 10점)",
+            "description": "블로그 글 품질 리뷰 (6기준 × 10점)",
             "template": (
                 "당신은 블로그 콘텐츠 품질 평가 전문가입니다.\n"
-                "아래 블로그 글을 5가지 기준으로 평가해주세요.\n\n"
+                "아래 블로그 글을 6가지 기준으로 평가해주세요.\n\n"
                 "제목: {title}\n"
                 "본문: {body}\n\n"
                 "평가 기준 (각 1-10점):\n"
-                "1. 독자 공감도: 1인칭 체험기가 자연스럽고 공감되는가?\n"
+                "1. 독자 공감도 (empathy): 1인칭 체험기가 자연스럽고 공감되는가?\n"
-                "2. 제목 클릭 유도력: 검색 결과에서 클릭하고 싶은 제목인가?\n"
+                "2. 제목 클릭 유도력 (click_appeal): 검색 결과에서 클릭하고 싶은 제목인가?\n"
-                "3. 구매 전환력: 읽고 나서 제품을 사고 싶어지는가?\n"
+                "3. 구매 전환력 (conversion): 읽고 나서 제품을 사고 싶어지는가?\n"
-                "4. SEO 최적화: 키워드 배치, 소제목, 길이가 적절한가?\n"
+                "4. SEO 최적화 (seo): 키워드 배치, 소제목, 길이가 적절한가?\n"
-                "5. 형식 완성도: 비교표, 이미지 설명, 단락 구성이 잘 되어있는가?\n\n"
+                "5. 형식 완성도 (format): 비교표, 이미지 설명, 단락 구성이 잘 되어있는가?\n"
+                "6. 링크 자연스러움 (link_natural): 제휴 링크가 광고처럼 느껴지지 않고 자연스럽게 녹아있는가? (링크가 없으면 5점 기본)\n\n"
                 "JSON 형식으로 응답:\n"
                 "{{\n"
                 "  \"scores\": {{\n"
@@ -170,7 +188,8 @@ def _seed_templates(conn: sqlite3.Connection) -> None:
                 "    \"click_appeal\": N,\n"
                 "    \"conversion\": N,\n"
                 "    \"seo\": N,\n"
-                "    \"format\": N\n"
+                "    \"format\": N,\n"
+                "    \"link_natural\": N\n"
                 "  }},\n"
                 "  \"total\": N,\n"
                 "  \"pass\": true/false,\n"
@@ -178,6 +197,26 @@ def _seed_templates(conn: sqlite3.Connection) -> None:
                 "}}"
             ),
         },
+        {
+            "name": "marketer_enhance",
+            "description": "마케터 전환율 강화 + 제휴 링크 삽입",
+            "template": (
+                "당신은 네이버 블로그 수익화 전문 마케터입니다.\n"
+                "아래 블로그 초안에 제휴 링크를 자연스럽게 삽입하고 전환율을 강화하세요.\n\n"
+                "=== 블로그 초안 ===\n{draft_body}\n\n"
+                "=== 타겟 키워드 ===\n{keyword}\n\n"
+                "=== 삽입할 제휴 링크 ===\n{brand_links_info}\n\n"
+                "작업 규칙:\n"
+                "- 제휴 링크를 <a href=\"URL\" target=\"_blank\">상품명</a> 형태로 본문 흐름에 맞게 2~3곳 삽입\n"
+                "- 결론에 CTA(Call-to-Action) 블록 추가 (\"지금 확인하기\" 등)\n"
+                "- 글 맨 아래에 광고 고지 문구 자동 삽입: \"이 포스팅은 브랜드로부터 소정의 수수료를 받을 수 있습니다\"\n"
+                "- 작가의 1인칭 톤과 구어체를 유지\n"
+                "- 과도한 광고 느낌 없이 자연스러운 추천 흐름 유지\n"
+                "- 구매 심리를 자극하는 표현 강화 (한정 수량, 가격 비교, 실사용 만족도 등)\n"
+                "- 배치 힌트가 있으면 참고하되, 문맥이 더 자연스러운 위치 우선\n"
+                "- 기존 본문의 구조와 길이를 크게 변경하지 않음"
+            ),
+        },
     ]
     for t in templates:
         existing = conn.execute(
@@ -190,6 +229,89 @@ 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,),
+    )
+
+    new_quality_review = (
+        "당신은 블로그 콘텐츠 품질 평가 전문가입니다.\n"
+        "아래 블로그 글을 6가지 기준으로 평가해주세요.\n\n"
+        "제목: {title}\n"
+        "본문: {body}\n\n"
+        "평가 기준 (각 1-10점):\n"
+        "1. 독자 공감도 (empathy): 1인칭 체험기가 자연스럽고 공감되는가?\n"
+        "2. 제목 클릭 유도력 (click_appeal): 검색 결과에서 클릭하고 싶은 제목인가?\n"
+        "3. 구매 전환력 (conversion): 읽고 나서 제품을 사고 싶어지는가?\n"
+        "4. SEO 최적화 (seo): 키워드 배치, 소제목, 길이가 적절한가?\n"
+        "5. 형식 완성도 (format): 비교표, 이미지 설명, 단락 구성이 잘 되어있는가?\n"
+        "6. 링크 자연스러움 (link_natural): 제휴 링크가 광고처럼 느껴지지 않고 자연스럽게 녹아있는가? (링크가 없으면 5점 기본)\n\n"
+        "JSON 형식으로 응답:\n"
+        "{{\n"
+        "  \"scores\": {{\n"
+        "    \"empathy\": N,\n"
+        "    \"click_appeal\": N,\n"
+        "    \"conversion\": N,\n"
+        "    \"seo\": N,\n"
+        "    \"format\": N,\n"
+        "    \"link_natural\": N\n"
+        "  }},\n"
+        "  \"total\": N,\n"
+        "  \"pass\": true/false,\n"
+        "  \"feedback\": \"개선 사항 설명\"\n"
+        "}}"
+    )
+    conn.execute(
+        "UPDATE prompt_templates SET template = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE name = 'quality_review'",
+        (new_quality_review,),
+    )
+
+    # marketer_enhance가 없으면 추가
+    existing = conn.execute("SELECT id FROM prompt_templates WHERE name = 'marketer_enhance'").fetchone()
+    if not existing:
+        conn.execute(
+            "INSERT INTO prompt_templates (name, description, template) VALUES (?, ?, ?)",
+            ("marketer_enhance", "마케터 전환율 강화 + 제휴 링크 삽입",
+             "당신은 네이버 블로그 수익화 전문 마케터입니다.\n"
+             "아래 블로그 초안에 제휴 링크를 자연스럽게 삽입하고 전환율을 강화하세요.\n\n"
+             "=== 블로그 초안 ===\n{draft_body}\n\n"
+             "=== 타겟 키워드 ===\n{keyword}\n\n"
+             "=== 삽입할 제휴 링크 ===\n{brand_links_info}\n\n"
+             "작업 규칙:\n"
+             "- 제휴 링크를 <a href=\"URL\" target=\"_blank\">상품명</a> 형태로 본문 흐름에 맞게 2~3곳 삽입\n"
+             "- 결론에 CTA(Call-to-Action) 블록 추가\n"
+             "- 글 맨 아래에 광고 고지 문구 자동 삽입\n"
+             "- 작가의 1인칭 톤과 구어체를 유지\n"
+             "- 과도한 광고 느낌 없이 자연스러운 추천 흐름 유지"),
+        )
+
+
 # ── keyword_analyses CRUD ────────────────────────────────────────────────────
 
 def _ka_row_to_dict(r) -> Dict[str, Any]:
@@ -453,6 +575,94 @@ def delete_commission(comm_id: int) -> bool:
     return True
 
 
+# ── brand_links CRUD ────────────────────────────────────────────────────────
+
+def _bl_row_to_dict(r) -> Dict[str, Any]:
+    return {
+        "id":             r["id"],
+        "post_id":        r["post_id"],
+        "keyword_id":     r["keyword_id"],
+        "url":            r["url"],
+        "product_name":   r["product_name"],
+        "description":    r["description"],
+        "placement_hint": r["placement_hint"],
+        "created_at":     r["created_at"],
+    }
+
+
+def add_brand_link(data: Dict[str, Any]) -> Dict[str, Any]:
+    with _conn() as conn:
+        conn.execute(
+            """INSERT INTO brand_links (post_id, keyword_id, url, product_name, description, placement_hint)
+            VALUES (?, ?, ?, ?, ?, ?)""",
+            (
+                data.get("post_id"),
+                data.get("keyword_id"),
+                data.get("url", ""),
+                data.get("product_name", ""),
+                data.get("description", ""),
+                data.get("placement_hint", ""),
+            ),
+        )
+        row = conn.execute(
+            "SELECT * FROM brand_links WHERE rowid = last_insert_rowid()"
+        ).fetchone()
+    return _bl_row_to_dict(row)
+
+
+def get_brand_links(
+    post_id: Optional[int] = None,
+    keyword_id: Optional[int] = None,
+) -> List[Dict[str, Any]]:
+    with _conn() as conn:
+        if post_id is not None:
+            rows = conn.execute(
+                "SELECT * FROM brand_links WHERE post_id = ? ORDER BY id", (post_id,)
+            ).fetchall()
+        elif keyword_id is not None:
+            rows = conn.execute(
+                "SELECT * FROM brand_links WHERE keyword_id = ? ORDER BY id", (keyword_id,)
+            ).fetchall()
+        else:
+            rows = conn.execute("SELECT * FROM brand_links ORDER BY id DESC LIMIT 100").fetchall()
+    return [_bl_row_to_dict(r) for r in rows]
+
+
+def update_brand_link(link_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    with _conn() as conn:
+        fields = []
+        values = []
+        for k in ("post_id", "keyword_id", "url", "product_name", "description", "placement_hint"):
+            if k in data:
+                fields.append(f"{k} = ?")
+                values.append(data[k])
+        if not fields:
+            row = conn.execute("SELECT * FROM brand_links WHERE id = ?", (link_id,)).fetchone()
+            return _bl_row_to_dict(row) if row else None
+        values.append(link_id)
+        conn.execute(f"UPDATE brand_links SET {', '.join(fields)} WHERE id = ?", values)
+        row = conn.execute("SELECT * FROM brand_links WHERE id = ?", (link_id,)).fetchone()
+    return _bl_row_to_dict(row) if row else None
+
+
+def delete_brand_link(link_id: int) -> bool:
+    with _conn() as conn:
+        row = conn.execute("SELECT id FROM brand_links WHERE id = ?", (link_id,)).fetchone()
+        if not row:
+            return False
+        conn.execute("DELETE FROM brand_links WHERE id = ?", (link_id,))
+    return True
+
+
+def link_brand_links_to_post(keyword_id: int, post_id: int) -> None:
+    """keyword_id로 등록된 링크들을 post_id에도 연결."""
+    with _conn() as conn:
+        conn.execute(
+            "UPDATE brand_links SET post_id = ? WHERE keyword_id = ? AND post_id IS NULL",
+            (post_id, keyword_id),
+        )
+
+
 def get_dashboard_stats() -> Dict[str, Any]:
     """대시보드 집계: 총 포스트/클릭/구매/수익 + 월별 추이."""
     with _conn() as conn:

blog-lab/app/main.py

@@ -15,10 +15,13 @@ from .db import (
     get_commissions, add_commission, update_commission, delete_commission,
     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
+from .naver_search import analyze_keyword_with_crawling
 from .content_generator import generate_trend_brief, generate_blog_post, regenerate_blog_post
 from .quality_reviewer import review_post
+from .marketer import enhance_for_conversion
 
 logger = logging.getLogger(__name__)
 
@@ -65,7 +68,7 @@ def _run_research(task_id: str, keyword: str):
     """BackgroundTask: 네이버 검색 → 키워드 분석 → DB 저장."""
     try:
         update_task(task_id, "processing", 30, "네이버 검색 중...")
-        result = analyze_keyword(keyword)
+        result = analyze_keyword_with_crawling(keyword)
 
         update_task(task_id, "processing", 80, "분석 결과 저장 중...")
         saved = add_keyword_analysis(result)
@@ -126,6 +129,15 @@ class GenerateRequest(BaseModel):
     keyword_id: int  # keyword_analyses.id
 
 
+class LinkRequest(BaseModel):
+    url: str
+    product_name: str
+    keyword_id: Optional[int] = None
+    post_id: Optional[int] = None
+    description: str = ""
+    placement_hint: str = ""
+
+
 def _run_generate(task_id: str, keyword_id: int):
     """BackgroundTask: 트렌드 브리프 → 블로그 글 생성 → DB 저장."""
     try:
@@ -134,11 +146,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({
@@ -151,6 +166,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)
@@ -304,6 +322,90 @@ def publish_post(post_id: int, data: dict = None):
     return result
 
 
+# ── 브랜드커넥트 링크 API ──────────────────────────────────────────────────
+
+@app.post("/api/blog-marketing/links", status_code=201)
+def create_link(req: LinkRequest):
+    return add_brand_link(req.model_dump())
+
+
+@app.get("/api/blog-marketing/links")
+def list_links(post_id: int = None, keyword_id: int = None):
+    return {"links": get_brand_links(post_id=post_id, keyword_id=keyword_id)}
+
+
+@app.put("/api/blog-marketing/links/{link_id}")
+def edit_link(link_id: int, data: dict):
+    result = update_brand_link(link_id, data)
+    if not result:
+        raise HTTPException(status_code=404, detail="Link not found")
+    return result
+
+
+@app.delete("/api/blog-marketing/links/{link_id}")
+def remove_link(link_id: int):
+    if not delete_brand_link(link_id):
+        raise HTTPException(status_code=404, detail="Link not found")
+    return {"ok": True}
+
+
+# ── 마케터 API ──────────────────────────────────────────────────────────────
+
+def _run_market(task_id: str, post_id: int):
+    """BackgroundTask: 마케터 전환율 강화."""
+    try:
+        post = get_post(post_id)
+        if not post:
+            update_task(task_id, "failed", 0, "", error="포스트를 찾을 수 없습니다")
+            return
+
+        brand_links = get_brand_links(post_id=post_id)
+        if not brand_links and post.get("keyword_id"):
+            brand_links = get_brand_links(keyword_id=post["keyword_id"])
+
+        if not brand_links:
+            update_task(task_id, "failed", 0, "", error="브랜드커넥트 링크가 없습니다. 먼저 링크를 등록하세요.")
+            return
+
+        analysis = get_keyword_analysis(post["keyword_id"]) if post.get("keyword_id") else {}
+        keyword = (analysis or {}).get("keyword", "")
+
+        update_task(task_id, "processing", 50, "마케터가 전환율 강화 중...")
+        result = enhance_for_conversion(
+            post_body=post["body"],
+            post_title=post["title"],
+            brand_links=brand_links,
+            keyword=keyword,
+        )
+
+        update_post(post_id, {
+            "title": result["title"],
+            "body": result["body"],
+            "excerpt": result["excerpt"],
+            "status": "marketed",
+        })
+
+        update_task(task_id, "succeeded", 100, "마케팅 강화 완료", result_id=post_id)
+    except Exception as e:
+        logger.exception("Market failed for post_id=%s", post_id)
+        update_task(task_id, "failed", 0, "", error=str(e))
+
+
+@app.post("/api/blog-marketing/market/{post_id}")
+def start_market(post_id: int, background_tasks: BackgroundTasks):
+    """마케터 단계 실행. task_id 즉시 반환."""
+    if not ANTHROPIC_API_KEY:
+        raise HTTPException(status_code=400, detail="Claude API 키가 설정되지 않았습니다")
+    post = get_post(post_id)
+    if not post:
+        raise HTTPException(status_code=404, detail="Post not found")
+
+    task_id = str(uuid.uuid4())
+    create_task(task_id, "market", {"post_id": post_id})
+    background_tasks.add_task(_run_market, task_id, post_id)
+    return {"task_id": task_id}
+
+
 # ── 수익 추적 API ────────────────────────────────────────────────────────────
 
 @app.get("/api/blog-marketing/commissions")

blog-lab/app/marketer.py

@@ -0,0 +1,102 @@
+"""마케터 단계 — 전환율 강화 + 브랜드커넥트 링크 삽입."""
+
+import json
+import logging
+from typing import Any, Dict, List, Optional
+
+import anthropic
+
+from .config import ANTHROPIC_API_KEY, CLAUDE_MODEL
+from .db import get_template
+
+logger = logging.getLogger(__name__)
+
+_client: Optional[anthropic.Anthropic] = None
+
+
+def _get_client() -> anthropic.Anthropic:
+    global _client
+    if _client is None:
+        _client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
+    return _client
+
+
+def _call_claude(prompt: str, max_tokens: int = 8192) -> str:
+    client = _get_client()
+    resp = client.messages.create(
+        model=CLAUDE_MODEL,
+        max_tokens=max_tokens,
+        messages=[{"role": "user", "content": prompt}],
+    )
+    return resp.content[0].text
+
+
+def enhance_for_conversion(
+    post_body: str,
+    post_title: str,
+    brand_links: List[Dict[str, Any]],
+    keyword: str,
+) -> Dict[str, str]:
+    """초안에 제휴 링크를 자연스럽게 삽입하고 전환율을 강화.
+
+    Args:
+        post_body: 작가 초안 HTML 본문
+        post_title: 작가 초안 제목
+        brand_links: 브랜드커넥트 링크 리스트
+        keyword: 타겟 키워드
+
+    Returns:
+        {"title": str, "body": str, "excerpt": str}
+
+    Raises:
+        ValueError: 브랜드 링크가 없을 때
+    """
+    if not brand_links:
+        raise ValueError("브랜드커넥트 링크가 필요합니다")
+
+    template = get_template("marketer_enhance")
+    if not template:
+        raise RuntimeError("marketer_enhance 템플릿이 없습니다")
+
+    brand_links_text = ""
+    for i, link in enumerate(brand_links, 1):
+        brand_links_text += (
+            f"{i}. 상품명: {link.get('product_name', '')}\n"
+            f"   설명: {link.get('description', '')}\n"
+            f"   URL: {link.get('url', '')}\n"
+            f"   배치 힌트: {link.get('placement_hint', '자연스럽게')}\n\n"
+        )
+
+    prompt = template.format(
+        draft_body=post_body[:6000],
+        keyword=keyword,
+        brand_links_info=brand_links_text,
+    )
+
+    prompt += (
+        "\n\n---\n"
+        "응답은 반드시 아래 JSON 형식으로 해주세요 (JSON만 출력):\n"
+        '{"title": "개선된 제목", "body": "개선된 HTML 본문", "excerpt": "2줄 요약"}'
+    )
+
+    raw = _call_claude(prompt)
+
+    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", post_title),
+            "body": result.get("body", post_body),
+            "excerpt": result.get("excerpt", ""),
+        }
+    except (json.JSONDecodeError, KeyError):
+        logger.warning("Marketer JSON parse failed, using raw text")
+        return {
+            "title": post_title,
+            "body": raw,
+            "excerpt": raw[:200],
+        }

blog-lab/app/naver_search.py

@@ -1,9 +1,13 @@
 """네이버 검색 API 연동 — 블로그 + 쇼핑 검색."""
 
+import asyncio
+import logging
 import re
 import requests
 from typing import Any, Dict, List, Optional
 
+logger = logging.getLogger(__name__)
+
 from .config import NAVER_CLIENT_ID, NAVER_CLIENT_SECRET
 
 BLOG_URL = "https://openapi.naver.com/v1/search/blog.json"
@@ -172,3 +176,28 @@ def analyze_keyword(keyword: str) -> Dict[str, Any]:
         "top_products": shop["items"][:5],
         "top_blogs": blog["items"][:5],
     }
+
+
+def _run_enrich(top_blogs: list) -> list:
+    """동기 컨텍스트에서 비동기 enrich_top_blogs 실행."""
+    from .web_crawler import enrich_top_blogs
+    try:
+        loop = asyncio.get_event_loop()
+        if loop.is_running():
+            import concurrent.futures
+            with concurrent.futures.ThreadPoolExecutor() as pool:
+                return pool.submit(
+                    asyncio.run, enrich_top_blogs(top_blogs)
+                ).result(timeout=60)
+        else:
+            return asyncio.run(enrich_top_blogs(top_blogs))
+    except Exception as e:
+        logger.warning("블로그 크롤링 실패, 기존 데이터 사용: %s", e)
+        return top_blogs
+
+
+def analyze_keyword_with_crawling(keyword: str) -> Dict[str, Any]:
+    """analyze_keyword + 상위 블로그 본문 크롤링."""
+    result = analyze_keyword(keyword)
+    result["top_blogs"] = _run_enrich(result["top_blogs"])
+    return result

blog-lab/app/quality_reviewer.py

@@ -1,4 +1,4 @@
-"""Claude API 기반 블로그 글 품질 리뷰 — 5기준 × 10점, 35/50 통과."""
+"""Claude API 기반 블로그 글 품질 리뷰 — 6기준 × 10점, 42/60 통과."""
 
 import json
 import logging
@@ -11,7 +11,7 @@ from .db import get_template
 
 logger = logging.getLogger(__name__)
 
-PASS_THRESHOLD = 35  # 50점 만점 중 35점 이상이면 통과
+PASS_THRESHOLD = 42  # 60점 만점 중 42점 이상이면 통과 (70%)
 
 _client: Optional[anthropic.Anthropic] = None
 
@@ -28,7 +28,10 @@ def review_post(title: str, body: str) -> Dict[str, Any]:
 
     Returns:
         {
-            "scores": {"empathy": N, "click_appeal": N, "conversion": N, "seo": N, "format": N},
+            "scores": {
+                "empathy": N, "click_appeal": N, "conversion": N,
+                "seo": N, "format": N, "link_natural": N
+            },
             "total": N,
             "pass": bool,
             "feedback": str
@@ -69,7 +72,10 @@ def review_post(title: str, body: str) -> Dict[str, Any]:
     except (json.JSONDecodeError, KeyError, TypeError) as e:
         logger.warning("Quality review JSON parse failed: %s", e)
         return {
-            "scores": {"empathy": 0, "click_appeal": 0, "conversion": 0, "seo": 0, "format": 0},
+            "scores": {
+                "empathy": 0, "click_appeal": 0, "conversion": 0,
+                "seo": 0, "format": 0, "link_natural": 0,
+            },
             "total": 0,
             "pass": False,
             "feedback": f"리뷰 파싱 실패. 원본 응답:\n{raw[:500]}",

blog-lab/app/web_crawler.py

@@ -0,0 +1,99 @@
+"""네이버 블로그 본문 크롤링 모듈."""
+
+import asyncio
+import logging
+import re
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import urlparse
+
+import httpx
+from bs4 import BeautifulSoup
+
+logger = logging.getLogger(__name__)
+
+_TIMEOUT = 10  # 글당 크롤링 타임아웃 (초)
+_MAX_CONTENT_LENGTH = 2000  # 본문 최대 길이
+
+# 네이버 블로그 URL 패턴: blog.naver.com/{blogId}/{logNo}
+_BLOG_URL_RE = re.compile(r"blog\.naver\.com/([^/]+)/(\d+)")
+
+
+def _parse_naver_blog_url(url: str) -> Optional[Tuple[str, str]]:
+    """네이버 블로그 URL에서 blogId, logNo 추출. 실패 시 None."""
+    match = _BLOG_URL_RE.search(url)
+    if not match:
+        return None
+    return match.group(1), match.group(2)
+
+
+async def _fetch_html(url: str) -> str:
+    """URL에서 HTML을 가져온다."""
+    async with httpx.AsyncClient(timeout=_TIMEOUT, follow_redirects=True) as client:
+        resp = await client.get(url, headers={
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+        })
+        resp.raise_for_status()
+        return resp.text
+
+
+def _extract_text(html: str) -> str:
+    """HTML에서 본문 텍스트를 추출한다."""
+    soup = BeautifulSoup(html, "html.parser")
+
+    # 스마트에디터 3 (SE3)
+    container = soup.select_one("div.se-main-container")
+    if not container:
+        # 구 에디터
+        container = soup.select_one("div#postViewArea")
+    if not container:
+        # 폴백: body 전체
+        container = soup.body
+
+    if not container:
+        return ""
+
+    # 스크립트/스타일 제거
+    for tag in container.find_all(["script", "style"]):
+        tag.decompose()
+
+    text = container.get_text(separator="\n", strip=True)
+    return text[:_MAX_CONTENT_LENGTH]
+
+
+async def crawl_blog_content(url: str) -> str:
+    """네이버 블로그 URL에서 본문 텍스트 추출.
+
+    - 네이버 블로그가 아니면 빈 문자열
+    - 크롤링 실패 시 빈 문자열 (에러 로그만)
+    - 본문 최대 2,000자
+    """
+    parsed = _parse_naver_blog_url(url)
+    if not parsed:
+        return ""
+
+    blog_id, log_no = parsed
+    # iframe 내부 실제 본문 URL
+    post_url = f"https://blog.naver.com/PostView.naver?blogId={blog_id}&logNo={log_no}"
+
+    try:
+        html = await _fetch_html(post_url)
+        return _extract_text(html)
+    except Exception as e:
+        logger.warning("블로그 크롤링 실패 (%s): %s", url, e)
+        return ""
+
+
+async def enrich_top_blogs(top_blogs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """top_blogs 리스트 각 항목에 content 필드를 추가.
+
+    개별 크롤링 실패 시 해당 항목의 content를 빈 문자열로 설정하고 나머지 계속 진행.
+    """
+    result = []
+    for blog in top_blogs:
+        enriched = dict(blog)
+        try:
+            enriched["content"] = await crawl_blog_content(blog.get("link", ""))
+        except Exception:
+            enriched["content"] = ""
+        result.append(enriched)
+    return result

blog-lab/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_mode = auto
+pythonpath = .

blog-lab/requirements.txt

@@ -2,3 +2,5 @@ fastapi==0.115.6
 uvicorn[standard]==0.34.0
 requests==2.32.3
 anthropic==0.52.0
+beautifulsoup4>=4.12
+httpx>=0.27

blog-lab/tests/conftest.py

@@ -0,0 +1,9 @@
+"""공통 테스트 픽스처."""
+import os
+import sys
+
+# app 패키지를 blog_lab_app으로도 import 가능하게
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+if "blog_lab_app" not in sys.modules:
+    import app as blog_lab_app
+    sys.modules["blog_lab_app"] = blog_lab_app

blog-lab/tests/test_api_links.py

@@ -0,0 +1,85 @@
+"""브랜드커넥트 링크 API 테스트."""
+import os
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(autouse=True)
+def setup_db(tmp_path):
+    test_db = str(tmp_path / "test.db")
+    import app.config as config
+    config.DB_PATH = test_db
+    from app import db
+    db.DB_PATH = test_db
+    db.init_db()
+    yield
+
+
+@pytest.fixture
+def client():
+    from app.main import app
+    return TestClient(app)
+
+
+def test_create_link(client):
+    resp = client.post("/api/blog-marketing/links", json={
+        "keyword_id": 1,
+        "url": "https://link.coupang.com/abc",
+        "product_name": "테스트 상품",
+        "description": "상품 설명",
+    })
+    assert resp.status_code == 201
+    data = resp.json()
+    assert data["url"] == "https://link.coupang.com/abc"
+    assert data["product_name"] == "테스트 상품"
+
+
+def test_create_link_requires_url(client):
+    resp = client.post("/api/blog-marketing/links", json={
+        "product_name": "상품",
+    })
+    assert resp.status_code == 422
+
+
+def test_create_link_requires_product_name(client):
+    resp = client.post("/api/blog-marketing/links", json={
+        "url": "https://a.com",
+    })
+    assert resp.status_code == 422
+
+
+def test_list_links_by_keyword_id(client):
+    client.post("/api/blog-marketing/links", json={
+        "keyword_id": 1, "url": "https://a.com", "product_name": "A",
+    })
+    client.post("/api/blog-marketing/links", json={
+        "keyword_id": 2, "url": "https://b.com", "product_name": "B",
+    })
+    resp = client.get("/api/blog-marketing/links?keyword_id=1")
+    assert resp.status_code == 200
+    assert len(resp.json()["links"]) == 1
+
+
+def test_update_link(client):
+    create_resp = client.post("/api/blog-marketing/links", json={
+        "url": "https://a.com", "product_name": "원래",
+    })
+    link_id = create_resp.json()["id"]
+    resp = client.put(f"/api/blog-marketing/links/{link_id}", json={
+        "product_name": "새이름",
+    })
+    assert resp.status_code == 200
+    assert resp.json()["product_name"] == "새이름"
+
+
+def test_delete_link(client):
+    create_resp = client.post("/api/blog-marketing/links", json={
+        "url": "https://a.com", "product_name": "삭제",
+    })
+    link_id = create_resp.json()["id"]
+    resp = client.delete(f"/api/blog-marketing/links/{link_id}")
+    assert resp.status_code == 200
+    assert resp.json()["ok"] is True
+
+    resp = client.delete(f"/api/blog-marketing/links/{link_id}")
+    assert resp.status_code == 404

blog-lab/tests/test_db_brand_links.py

@@ -0,0 +1,67 @@
+"""brand_links DB CRUD 테스트."""
+import os
+import pytest
+from app import db
+from app.config import DB_PATH
+
+
+@pytest.fixture(autouse=True)
+def setup_db(tmp_path):
+    """테스트용 임시 DB 사용."""
+    test_db = str(tmp_path / "test.db")
+    import app.config as config
+    config.DB_PATH = test_db
+    db.DB_PATH = test_db
+    db.init_db()
+    yield
+
+
+def test_add_brand_link():
+    link = db.add_brand_link({
+        "keyword_id": 1,
+        "url": "https://link.coupang.com/abc",
+        "product_name": "테스트 상품",
+        "description": "상품 설명",
+        "placement_hint": "본문 중간",
+    })
+    assert link["id"] is not None
+    assert link["url"] == "https://link.coupang.com/abc"
+    assert link["product_name"] == "테스트 상품"
+    assert link["keyword_id"] == 1
+    assert link["post_id"] is None
+
+
+def test_get_brand_links_by_keyword_id():
+    db.add_brand_link({"keyword_id": 1, "url": "https://a.com", "product_name": "A"})
+    db.add_brand_link({"keyword_id": 1, "url": "https://b.com", "product_name": "B"})
+    db.add_brand_link({"keyword_id": 2, "url": "https://c.com", "product_name": "C"})
+    links = db.get_brand_links(keyword_id=1)
+    assert len(links) == 2
+
+
+def test_get_brand_links_by_post_id():
+    db.add_brand_link({"post_id": 10, "url": "https://a.com", "product_name": "A"})
+    links = db.get_brand_links(post_id=10)
+    assert len(links) == 1
+    assert links[0]["post_id"] == 10
+
+
+def test_update_brand_link():
+    link = db.add_brand_link({"url": "https://a.com", "product_name": "원래 이름"})
+    updated = db.update_brand_link(link["id"], {"product_name": "새 이름", "post_id": 5})
+    assert updated["product_name"] == "새 이름"
+    assert updated["post_id"] == 5
+
+
+def test_delete_brand_link():
+    link = db.add_brand_link({"url": "https://a.com", "product_name": "삭제할 링크"})
+    assert db.delete_brand_link(link["id"]) is True
+    assert db.delete_brand_link(link["id"]) is False
+
+
+def test_link_keyword_to_post():
+    db.add_brand_link({"keyword_id": 1, "url": "https://a.com", "product_name": "A"})
+    db.add_brand_link({"keyword_id": 1, "url": "https://b.com", "product_name": "B"})
+    db.link_brand_links_to_post(keyword_id=1, post_id=10)
+    links = db.get_brand_links(post_id=10)
+    assert len(links) == 2

blog-lab/tests/test_evaluator.py

@@ -0,0 +1,74 @@
+"""평가자 단계 테스트 — 6기준 60점."""
+import json
+import pytest
+from unittest.mock import patch
+
+
+def test_review_post_has_6_criteria():
+    """6개 기준으로 채점하는지 확인."""
+    from app.quality_reviewer import review_post
+
+    mock_response = json.dumps({
+        "scores": {
+            "empathy": 8, "click_appeal": 7, "conversion": 9,
+            "seo": 8, "format": 7, "link_natural": 9,
+        },
+        "total": 48,
+        "pass": True,
+        "feedback": "전체적으로 우수합니다",
+    })
+
+    with patch("app.quality_reviewer._get_client") as mock_client_fn, \
+         patch("app.quality_reviewer.get_template", return_value="제목: {title}\n본문: {body}"):
+        mock_client = mock_client_fn.return_value
+        mock_client.messages.create.return_value.content = [type("C", (), {"text": mock_response})()]
+        result = review_post("테스트 제목", "<p>본문</p>")
+
+    assert "link_natural" in result["scores"]
+    assert len(result["scores"]) == 6
+    assert result["total"] == 48
+    assert result["pass"] is True
+
+
+def test_review_pass_threshold_is_42():
+    """통과 기준이 42점인지 확인."""
+    from app.quality_reviewer import PASS_THRESHOLD
+    assert PASS_THRESHOLD == 42
+
+
+def test_review_fails_below_42():
+    """42점 미만이면 불통과."""
+    from app.quality_reviewer import review_post
+
+    mock_response = json.dumps({
+        "scores": {
+            "empathy": 5, "click_appeal": 5, "conversion": 5,
+            "seo": 5, "format": 5, "link_natural": 5,
+        },
+        "total": 30,
+        "pass": False,
+        "feedback": "개선 필요",
+    })
+
+    with patch("app.quality_reviewer._get_client") as mock_client_fn, \
+         patch("app.quality_reviewer.get_template", return_value="제목: {title}\n본문: {body}"):
+        mock_client = mock_client_fn.return_value
+        mock_client.messages.create.return_value.content = [type("C", (), {"text": mock_response})()]
+        result = review_post("제목", "<p>본문</p>")
+
+    assert result["pass"] is False
+
+
+def test_review_handles_parse_failure():
+    """JSON 파싱 실패 시 기본값 반환 (6개 기준)."""
+    from app.quality_reviewer import review_post
+
+    with patch("app.quality_reviewer._get_client") as mock_client_fn, \
+         patch("app.quality_reviewer.get_template", return_value="제목: {title}\n본문: {body}"):
+        mock_client = mock_client_fn.return_value
+        mock_client.messages.create.return_value.content = [type("C", (), {"text": "잘못된 응답"})()]
+        result = review_post("제목", "<p>본문</p>")
+
+    assert result["pass"] is False
+    assert "link_natural" in result["scores"]
+    assert result["total"] == 0

blog-lab/tests/test_marketer.py

@@ -0,0 +1,66 @@
+"""마케터 단계 테스트."""
+import json
+import pytest
+from unittest.mock import patch
+
+
+def test_enhance_for_conversion_inserts_links():
+    """마케터가 브랜드 링크를 본문에 삽입."""
+    from app.marketer import enhance_for_conversion
+
+    brand_links = [
+        {"url": "https://link.coupang.com/abc", "product_name": "갤럭시 버즈3",
+         "description": "노이즈캔슬링", "placement_hint": "본문 중간"},
+    ]
+
+    mock_response = json.dumps({
+        "title": "마케팅된 제목",
+        "body": '<p>본문 <a href="https://link.coupang.com/abc">갤럭시 버즈3</a></p>',
+        "excerpt": "요약",
+    })
+
+    with patch("app.marketer._call_claude", return_value=mock_response) as mock_call, \
+         patch("app.marketer.get_template", return_value="초안: {draft_body}\n키워드: {keyword}\n링크:\n{brand_links_info}"):
+        result = enhance_for_conversion(
+            post_body="<p>초안 본문</p>",
+            post_title="초안 제목",
+            brand_links=brand_links,
+            keyword="무선 이어폰",
+        )
+
+    prompt_used = mock_call.call_args[0][0]
+    assert "갤럭시 버즈3" in prompt_used
+    assert "노이즈캔슬링" in prompt_used
+    assert result["title"] == "마케팅된 제목"
+
+
+def test_enhance_requires_brand_links():
+    """브랜드 링크가 없으면 ValueError."""
+    from app.marketer import enhance_for_conversion
+
+    with pytest.raises(ValueError, match="브랜드커넥트 링크가 필요합니다"):
+        enhance_for_conversion(
+            post_body="<p>본문</p>",
+            post_title="제목",
+            brand_links=[],
+            keyword="테스트",
+        )
+
+
+def test_enhance_json_parse_fallback():
+    """JSON 파싱 실패 시 원본 제목 유지."""
+    from app.marketer import enhance_for_conversion
+
+    brand_links = [{"url": "https://a.com", "product_name": "상품"}]
+
+    with patch("app.marketer._call_claude", return_value="잘못된 JSON"), \
+         patch("app.marketer.get_template", return_value="초안: {draft_body}\n키워드: {keyword}\n링크:\n{brand_links_info}"):
+        result = enhance_for_conversion(
+            post_body="<p>원본</p>",
+            post_title="원본 제목",
+            brand_links=brand_links,
+            keyword="테스트",
+        )
+
+    assert result["title"] == "원본 제목"
+    assert result["body"] == "잘못된 JSON"

blog-lab/tests/test_pipeline_integration.py

@@ -0,0 +1,146 @@
+"""4단계 파이프라인 통합 테스트."""
+import os
+import pytest
+from unittest.mock import patch
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(autouse=True)
+def setup_db(tmp_path):
+    test_db = str(tmp_path / "test.db")
+    import app.config as config
+    config.DB_PATH = test_db
+    from app import db
+    db.DB_PATH = test_db
+    db.init_db()
+    yield
+
+
+@pytest.fixture
+def client():
+    from app.main import app
+    return TestClient(app)
+
+
+def test_full_pipeline_status_flow(client):
+    """draft → marketed → reviewed → published 상태 흐름."""
+    from app import db
+
+    # 1. 키워드 분석 결과 직접 삽입
+    analysis = db.add_keyword_analysis({
+        "keyword": "무선 이어폰",
+        "blog_total": 1000,
+        "shop_total": 500,
+        "competition": 45,
+        "opportunity": 60,
+        "top_products": [{"title": "에어팟", "lprice": 200000, "mallName": "애플"}],
+        "top_blogs": [{"title": "리뷰", "link": "https://blog.naver.com/user/123", "content": "본문"}],
+    })
+
+    # 2. 브랜드 링크 등록
+    resp = client.post("/api/blog-marketing/links", json={
+        "keyword_id": analysis["id"],
+        "url": "https://link.coupang.com/abc",
+        "product_name": "삼성 버즈3",
+        "description": "노이즈캔슬링",
+    })
+    assert resp.status_code == 201
+
+    # 3. 포스트 직접 생성 (generate는 Claude API 필요)
+    post = db.add_post({
+        "keyword_id": analysis["id"],
+        "title": "무선 이어폰 추천",
+        "body": "<p>초안 본문</p>",
+        "excerpt": "요약",
+        "tags": ["이어폰"],
+        "status": "draft",
+    })
+    db.link_brand_links_to_post(keyword_id=analysis["id"], post_id=post["id"])
+
+    # 4. 상태 확인: draft
+    resp = client.get(f"/api/blog-marketing/posts/{post['id']}")
+    assert resp.json()["status"] == "draft"
+
+    # 5. marketed 상태
+    db.update_post(post["id"], {"status": "marketed", "body": "<p>마케팅된 본문</p>"})
+    resp = client.get(f"/api/blog-marketing/posts/{post['id']}")
+    assert resp.json()["status"] == "marketed"
+
+    # 6. reviewed 상태 (점수 48/60 = 통과)
+    db.update_post(post["id"], {
+        "status": "reviewed",
+        "review_score": 48,
+        "review_detail": {
+            "scores": {"empathy": 8, "click_appeal": 8, "conversion": 8, "seo": 8, "format": 8, "link_natural": 8},
+            "total": 48, "pass": True, "feedback": "우수"
+        },
+    })
+    resp = client.get(f"/api/blog-marketing/posts/{post['id']}")
+    assert resp.json()["status"] == "reviewed"
+    assert resp.json()["review_score"] == 48
+
+    # 7. 발행
+    resp = client.post(f"/api/blog-marketing/posts/{post['id']}/publish", json={
+        "naver_url": "https://blog.naver.com/mypost/123",
+    })
+    assert resp.json()["status"] == "published"
+
+
+def test_links_associated_with_post(client):
+    """keyword_id로 등록한 링크가 post 생성 후 post_id로도 조회 가능."""
+    from app import db
+
+    analysis = db.add_keyword_analysis({"keyword": "테스트", "blog_total": 10, "shop_total": 5})
+    client.post("/api/blog-marketing/links", json={
+        "keyword_id": analysis["id"],
+        "url": "https://link.com/1",
+        "product_name": "상품1",
+    })
+
+    post = db.add_post({"keyword_id": analysis["id"], "title": "제목", "body": "본문", "status": "draft"})
+    db.link_brand_links_to_post(keyword_id=analysis["id"], post_id=post["id"])
+
+    resp = client.get(f"/api/blog-marketing/links?post_id={post['id']}")
+    links = resp.json()["links"]
+    assert len(links) == 1
+    assert links[0]["product_name"] == "상품1"
+
+
+@patch("app.main.ANTHROPIC_API_KEY", "fake-key-for-test")
+def test_market_endpoint_returns_404_for_missing_post(client):
+    """존재하지 않는 post_id로 마케터 호출 시 404."""
+    resp = client.post("/api/blog-marketing/market/9999")
+    assert resp.status_code == 404
+
+
+@patch("app.main.ANTHROPIC_API_KEY", "fake-key-for-test")
+def test_review_endpoint_returns_404_for_missing_post(client):
+    """존재하지 않는 post_id로 리뷰 호출 시 404."""
+    resp = client.post("/api/blog-marketing/review/9999")
+    assert resp.status_code == 404
+
+
+def test_multiple_links_per_keyword(client):
+    """하나의 키워드에 복수 링크 등록 가능."""
+    from app import db
+    analysis = db.add_keyword_analysis({"keyword": "테스트", "blog_total": 10, "shop_total": 5})
+
+    for i in range(3):
+        resp = client.post("/api/blog-marketing/links", json={
+            "keyword_id": analysis["id"],
+            "url": f"https://link.com/{i}",
+            "product_name": f"상품{i}",
+        })
+        assert resp.status_code == 201
+
+    resp = client.get(f"/api/blog-marketing/links?keyword_id={analysis['id']}")
+    assert len(resp.json()["links"]) == 3
+
+
+def test_dashboard_still_works(client):
+    """대시보드 API가 여전히 정상 작동."""
+    resp = client.get("/api/blog-marketing/dashboard")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert "total_posts" in data
+    assert "published_posts" in data

blog-lab/tests/test_research_crawling.py

@@ -0,0 +1,58 @@
+"""리서치 단계 크롤링 통합 테스트."""
+from unittest.mock import patch
+
+
+def test_analyze_keyword_with_crawling_enriches_top_blogs():
+    """analyze_keyword_with_crawling가 top_blogs에 content 필드를 추가."""
+    from app.naver_search import analyze_keyword_with_crawling
+
+    mock_blog_result = {
+        "total": 100,
+        "items": [
+            {"title": "테스트 블로그", "link": "https://blog.naver.com/user1/111",
+             "bloggername": "유저1", "description": "설명", "postdate": "20260401"},
+        ],
+    }
+    mock_shop_result = {
+        "total": 50,
+        "items": [{"title": "상품1", "lprice": 10000, "mallName": "쿠팡"}],
+        "price_stats": {"min": 10000, "max": 10000, "avg": 10000, "count": 1},
+    }
+
+    with patch("app.naver_search.search_blog", return_value=mock_blog_result), \
+         patch("app.naver_search.search_shopping", return_value=mock_shop_result), \
+         patch("app.naver_search._run_enrich", return_value=[
+             {"title": "테스트 블로그", "link": "https://blog.naver.com/user1/111",
+              "bloggername": "유저1", "description": "설명", "postdate": "20260401",
+              "content": "크롤링된 본문 내용"}
+         ]):
+        result = analyze_keyword_with_crawling("테스트 키워드")
+
+    assert "content" in result["top_blogs"][0]
+    assert result["top_blogs"][0]["content"] == "크롤링된 본문 내용"
+
+
+def test_analyze_keyword_with_crawling_fallback_on_enrich_failure():
+    """크롤링 실패 시 기존 데이터 유지."""
+    from app.naver_search import analyze_keyword_with_crawling
+
+    mock_blog_result = {
+        "total": 50,
+        "items": [{"title": "블로그", "link": "https://blog.naver.com/u/1", "bloggername": "유저", "description": "설명"}],
+    }
+    mock_shop_result = {"total": 10, "items": [], "price_stats": None}
+
+    with patch("app.naver_search.search_blog", return_value=mock_blog_result), \
+         patch("app.naver_search.search_shopping", return_value=mock_shop_result), \
+         patch("app.naver_search._run_enrich", side_effect=Exception("크롤링 실패")):
+        # _run_enrich 내부에서 예외를 잡으므로 실제로는 이 테스트에서는
+        # _run_enrich 자체가 예외를 던지는 상황을 시뮬레이션
+        # 하지만 _run_enrich는 내부에서 잡으므로, 직접 fallback 테스트
+        pass
+
+    # _run_enrich 자체 fallback 테스트
+    from app.naver_search import _run_enrich
+    original_blogs = [{"title": "원본", "link": "https://blog.naver.com/u/1"}]
+    with patch("app.web_crawler.enrich_top_blogs", side_effect=Exception("fail")):
+        result = _run_enrich(original_blogs)
+    assert result == original_blogs  # fallback으로 원본 반환

blog-lab/tests/test_web_crawler.py

@@ -0,0 +1,94 @@
+"""web_crawler 모듈 테스트."""
+import pytest
+from unittest.mock import patch, AsyncMock
+from app.web_crawler import crawl_blog_content, enrich_top_blogs, _parse_naver_blog_url, _extract_text
+
+
+def test_parse_naver_blog_url_valid():
+    """blog.naver.com URL에서 blogId와 logNo를 올바르게 파싱."""
+    result = _parse_naver_blog_url("https://blog.naver.com/testuser/123456")
+    assert result == ("testuser", "123456")
+
+
+def test_parse_returns_none_for_invalid_url():
+    """잘못된 URL은 None 반환."""
+    result = _parse_naver_blog_url("https://example.com/post")
+    assert result is None
+
+
+def test_extract_text_prefers_se_main_container():
+    """SE3 에디터 컨테이너를 우선 선택."""
+    html = '<div class="se-main-container"><p>SE3 본문</p></div><div id="postViewArea"><p>구 에디터</p></div>'
+    assert _extract_text(html) == "SE3 본문"
+
+
+def test_extract_text_falls_back_to_post_view_area():
+    """SE3 없으면 구 에디터 컨테이너 사용."""
+    html = '<div id="postViewArea"><p>구 에디터 본문</p></div>'
+    assert _extract_text(html) == "구 에디터 본문"
+
+
+def test_extract_text_removes_script_and_style():
+    """스크립트/스타일 태그 제거."""
+    html = '<div class="se-main-container"><p>본문</p><script>alert(1)</script><style>.x{}</style></div>'
+    result = _extract_text(html)
+    assert "alert" not in result
+    assert ".x" not in result
+    assert "본문" in result
+
+
+def test_extract_text_returns_empty_on_no_container():
+    """컨테이너가 없고 body도 없으면 빈 문자열."""
+    assert _extract_text("") == ""
+
+
+@pytest.mark.asyncio
+async def test_crawl_returns_empty_on_non_naver_url():
+    """네이버 블로그가 아닌 URL은 빈 문자열 반환."""
+    result = await crawl_blog_content("https://example.com/post")
+    assert result == ""
+
+
+@pytest.mark.asyncio
+async def test_crawl_truncates_to_2000_chars():
+    """본문이 2000자를 초과하면 잘라낸다."""
+    long_html = f'<div class="se-main-container"><p>{"가" * 3000}</p></div>'
+    with patch("app.web_crawler._fetch_html", new_callable=AsyncMock, return_value=long_html):
+        result = await crawl_blog_content("https://blog.naver.com/testuser/123")
+    assert len(result) <= 2000
+
+
+@pytest.mark.asyncio
+async def test_crawl_returns_empty_on_fetch_failure():
+    """HTTP 요청 실패 시 빈 문자열 반환."""
+    with patch("app.web_crawler._fetch_html", new_callable=AsyncMock, side_effect=Exception("timeout")):
+        result = await crawl_blog_content("https://blog.naver.com/testuser/123")
+    assert result == ""
+
+
+@pytest.mark.asyncio
+async def test_enrich_top_blogs_adds_content_field():
+    """enrich_top_blogs가 각 블로그에 content 필드를 추가."""
+    blogs = [
+        {"title": "테스트", "link": "https://blog.naver.com/user1/111", "bloggername": "유저1", "description": "설명"},
+        {"title": "테스트2", "link": "https://blog.naver.com/user2/222", "bloggername": "유저2", "description": "설명2"},
+    ]
+    with patch("app.web_crawler.crawl_blog_content", new_callable=AsyncMock, return_value="크롤링된 본문"):
+        result = await enrich_top_blogs(blogs)
+    assert len(result) == 2
+    assert result[0]["content"] == "크롤링된 본문"
+    assert result[1]["content"] == "크롤링된 본문"
+
+
+@pytest.mark.asyncio
+async def test_enrich_top_blogs_handles_partial_failure():
+    """일부 크롤링 실패 시에도 나머지는 정상 처리."""
+    blogs = [
+        {"title": "성공", "link": "https://blog.naver.com/user1/111"},
+        {"title": "실패", "link": "https://blog.naver.com/user2/222"},
+    ]
+    side_effects = ["성공 본문", Exception("fail")]
+    with patch("app.web_crawler.crawl_blog_content", new_callable=AsyncMock, side_effect=side_effects):
+        result = await enrich_top_blogs(blogs)
+    assert result[0]["content"] == "성공 본문"
+    assert result[1]["content"] == ""

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"