web-page-backend/blog-lab/app/content_generator.py

"""Claude API 기반 콘텐츠 생성 — 트렌드 브리프 + 블로그 글 작성."""

import json
import logging
from datetime import date
from typing import Any, Dict, 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 = 4096) -> str:
    """Claude API 호출. 단일 user 메시지. 현재 날짜 시스템 프롬프트 포함."""
    client = _get_client()
    today = date.today().isoformat()
    resp = client.messages.create(
        model=CLAUDE_MODEL,
        max_tokens=max_tokens,
        system=f"현재 날짜는 {today}입니다. 모든 콘텐츠는 이 날짜 기준으로 작성하세요.",
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.content[0].text


def generate_trend_brief(analysis: Dict[str, Any]) -> str:
    """키워드 분석 데이터를 바탕으로 트렌드 브리프 생성."""
    template = get_template("trend_brief")
    if not template:
        raise RuntimeError("trend_brief 템플릿이 없습니다")

    top_blogs_text = "\n".join(
        f"- {b.get('title', '')}" for b in analysis.get("top_blogs", [])
    ) or "없음"

    top_products_text = "\n".join(
        f"- {p.get('title', '')} ({p.get('lprice', '?')}원, {p.get('mallName', '')})"
        for p in analysis.get("top_products", [])
    ) or "없음"

    prompt = template.format(
        keyword=analysis.get("keyword", ""),
        competition=analysis.get("competition", 0),
        opportunity=analysis.get("opportunity", 0),
        top_blogs=top_blogs_text,
        top_products=top_products_text,
    )

    return _call_claude(prompt)


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:
        {"title": str, "body": str, "excerpt": str, "tags": [...]}
    """
    template = get_template("blog_write")
    if not template:
        raise RuntimeError("blog_write 템플릿이 없습니다")

    top_products_text = "\n".join(
        f"- {p.get('title', '')} ({p.get('lprice', '?')}원, {p.get('mallName', '')})"
        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,
    )

    # 구조화된 응답을 위한 추가 지시
    prompt += (
        "\n\n---\n"
        "응답은 반드시 아래 JSON 형식으로 해주세요 (JSON만 출력, 다른 텍스트 없이):\n"
        '{"title": "블로그 제목", "body": "HTML 본문", "excerpt": "2줄 요약", '
        '"tags": ["태그1", "태그2", ...]}'
    )

    raw = _call_claude(prompt, max_tokens=8192)
    return _parse_blog_json(raw, analysis.get("keyword", ""))


def regenerate_blog_post(
    analysis: Dict[str, Any],
    trend_brief: str,
    previous_body: str,
    feedback: str,
) -> Dict[str, str]:
    """피드백을 반영하여 블로그 글 재생성."""
    prompt = (
        "당신은 네이버 블로그에서 월 100만 이상 수익을 올리는 전문 블로거입니다.\n"
        f"키워드: {analysis.get('keyword', '')}\n\n"
        f"이전에 작성한 글:\n{previous_body[:3000]}\n\n"
        f"리뷰어 피드백:\n{feedback}\n\n"
        "위 피드백을 반영하여 글을 개선해주세요.\n"
        "작성 규칙: 1인칭 체험기, 2,000자 이상, 자연스러운 구어체, "
        "제품 비교표 포함, 광고 고지 문구 포함.\n"
        "HTML 형식으로 작성하되, 네이버 블로그에서 바로 붙여넣기 가능한 형태로.\n\n"
        "---\n"
        "응답은 반드시 아래 JSON 형식으로 해주세요 (JSON만 출력):\n"
        '{"title": "블로그 제목", "body": "HTML 본문", "excerpt": "2줄 요약", '
        '"tags": ["태그1", "태그2", ...]}'
    )
    raw = _call_claude(prompt, max_tokens=8192)
    return _parse_blog_json(raw, analysis.get("keyword", ""))