diff --git a/blog-lab/app/db.py b/blog-lab/app/db.py
index a9f5e2c..13092ef 100644
--- a/blog-lab/app/db.py
+++ b/blog-lab/app/db.py
@@ -195,6 +195,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"
+ "- 제휴 링크를 상품명 형태로 본문 흐름에 맞게 2~3곳 삽입\n"
+ "- 결론에 CTA(Call-to-Action) 블록 추가 (\"지금 확인하기\" 등)\n"
+ "- 글 맨 아래에 광고 고지 문구 자동 삽입: \"이 포스팅은 브랜드로부터 소정의 수수료를 받을 수 있습니다\"\n"
+ "- 작가의 1인칭 톤과 구어체를 유지\n"
+ "- 과도한 광고 느낌 없이 자연스러운 추천 흐름 유지\n"
+ "- 구매 심리를 자극하는 표현 강화 (한정 수량, 가격 비교, 실사용 만족도 등)\n"
+ "- 배치 힌트가 있으면 참고하되, 문맥이 더 자연스러운 위치 우선\n"
+ "- 기존 본문의 구조와 길이를 크게 변경하지 않음"
+ ),
+ },
]
for t in templates:
existing = conn.execute(
@@ -238,6 +258,25 @@ def _migrate_templates(conn: sqlite3.Connection) -> None:
(new_blog_write,),
)
+ # 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"
+ "- 제휴 링크를 상품명 형태로 본문 흐름에 맞게 2~3곳 삽입\n"
+ "- 결론에 CTA(Call-to-Action) 블록 추가\n"
+ "- 글 맨 아래에 광고 고지 문구 자동 삽입\n"
+ "- 작가의 1인칭 톤과 구어체를 유지\n"
+ "- 과도한 광고 느낌 없이 자연스러운 추천 흐름 유지"),
+ )
+
# ── keyword_analyses CRUD ────────────────────────────────────────────────────
diff --git a/blog-lab/app/main.py b/blog-lab/app/main.py
index b84e506..3d9fc5c 100644
--- a/blog-lab/app/main.py
+++ b/blog-lab/app/main.py
@@ -21,6 +21,7 @@ from .db import (
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__)
@@ -348,6 +349,63 @@ def remove_link(link_id: int):
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")
diff --git a/blog-lab/app/marketer.py b/blog-lab/app/marketer.py
new file mode 100644
index 0000000..8919657
--- /dev/null
+++ b/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],
+ }
diff --git a/blog-lab/tests/test_marketer.py b/blog-lab/tests/test_marketer.py
new file mode 100644
index 0000000..96e3551
--- /dev/null
+++ b/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": '본문 갤럭시 버즈3
',
+ "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="초안 본문
",
+ 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="본문
",
+ 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="원본
",
+ post_title="원본 제목",
+ brand_links=brand_links,
+ keyword="테스트",
+ )
+
+ assert result["title"] == "원본 제목"
+ assert result["body"] == "잘못된 JSON"