o2o-site-AEO/solution/backend/services/blog_service.py

"""미니 블로그 — AI 자동 포스트. 기획: docs/MINI_BLOG.md

★ 발행 게이트와 부딪히지 않게 만든다. 규칙 1(미검증 fact 는 화면에 내지 않는다)은 홍보 문구에도
  그대로 걸린다 — 가격·시간·인원을 문구가 주장하면 그 주장을 뒷받침할 fact 가 없다.
  프롬프트로 금지하고, 생성 뒤 `is_publishable_body()` 로 한 번 더 거른다.
★ 중복은 프롬프트가 아니라 DB 가 막는다 — (place_id, topic_key) 유니크.
"""
import hashlib
import re
import secrets
from datetime import datetime, timedelta, timezone

from common.enums import PostStatus, PostTopicKind
from common.logger import LOG

# 본문 길이 — 회의 확정값(140~150자)에 여유를 둔다. 벗어나면 버린다.
MIN_LEN = 120
MAX_LEN = 170

_KST = timezone(timedelta(hours=9))

# 문구가 주장하면 안 되는 것. 게이트가 잡기 전에 여기서 버린다.
_FORBIDDEN = (
    re.compile(r"\d{1,3},\d{3}\s*원"),        # 198,000원
    re.compile(r"\d+\s*원"),                   # 50000원 · 3만원 은 아래에서
    re.compile(r"\d+\s*만\s*원"),
    re.compile(r"\d{1,2}\s*:\s*\d{2}"),        # 15:00
    re.compile(r"\d+\s*시\s*(\d+\s*분)?\s*(부터|까지|에)"),
    re.compile(r"\d+\s*(인|명)\s*(까지|기준|이상)"),
    re.compile(r"\d{2,3}-\d{3,4}-\d{4}"),      # 전화번호
    re.compile(r"(무료|공짜)\s*(제공|이용|주차)"),
    re.compile(r"(최고|최저|1위|유일)"),         # 근거를 못 대는 최상급
)


def is_publishable_body(text: str) -> tuple[bool, str]:
    """(통과 여부, 사유). 사유는 로그·검수 화면에 그대로 쓴다."""
    body = (text or "").strip()
    if not body:
        return False, "빈 글"
    if len(body) < MIN_LEN or len(body) > MAX_LEN:
        return False, f"길이 {len(body)}자 — {MIN_LEN}~{MAX_LEN} 밖"
    for pattern in _FORBIDDEN:
        hit = pattern.search(body)
        if hit:
            return False, f"확인되지 않은 주장: {hit.group(0)}"
    return True, ""


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def issue_token() -> tuple[str, str, object]:
    """(평문, 해시, 만료시각=오늘 자정 KST). 평문은 메일 본문에만 나가고 DB 에는 해시만 둔다.

    ★ 승인·수정 두 링크 다 그날까지만 산다(2026-09-17, 사장님 지시: "승인이랑 수정모두
      자정에 만료"). 그 뒤로는 로그인해서 빌더 앱에서 처리한다 — 메일 링크는 "오늘 온 것을
      오늘 처리하라"는 뜻이지 보관함이 아니다."""
    token = secrets.token_urlsafe(32)
    now_kst = datetime.now(_KST)
    midnight_kst = (now_kst + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
    expires = midnight_kst.astimezone(timezone.utc).replace(tzinfo=None)
    return token, hash_token(token), expires


TOPIC_RULES: dict[int, str] = {
    PostTopicKind.WEATHER.value:
        "오늘의 날씨와 그 날씨에 이 숙소에서 하기 좋은 일을 한 장면으로 적는다.",
    PostTopicKind.FESTIVAL.value:
        "주어진 축제 하나를 언급하고, 숙소에서 그곳까지 어떻게 가는지를 걸음 단위로 적는다.",
    PostTopicKind.SEASON.value:
        "지금 절기에 이 지역과 숙소가 어떻게 달라지는지를 적는다.",
    PostTopicKind.NEARBY.value:
        "주어진 주변 장소 하나를 손님 시선에서 적는다. 영업시간과 가격은 쓰지 않는다.",
    PostTopicKind.GUIDE.value:
        "확인된 이용 안내 하나를 손님이 알아두면 좋은 말투로 풀어 적는다.",
}

_RULES = (
    "규칙\n"
    f"- {MIN_LEN}~{MAX_LEN}자 사이 한 문단. 제목·해시태그·이모지를 쓰지 않는다.\n"
    "- 숫자로 된 요금·시간·인원·전화번호를 쓰지 않는다. 확인되지 않은 주장을 하지 않는다.\n"
    "- '최고' '유일' 같은 최상급을 쓰지 않는다.\n"
    "- 손님에게 말하듯 존댓말로 적는다.\n"
    "- 아래 '이미 쓴 주제'와 겹치는 소재를 고르지 않는다.\n"
)


def build_prompt(*, place_name: str, region: str, topic_kind: int, material: str, used_topics: list[str]) -> str:
    """갈래 하나에 대한 프롬프트 한 벌. 프롬프트를 두 곳에 적지 않으려고 여기서만 만든다."""
    used = ", ".join(used_topics[:40]) or "없음"
    return (
        f"{region}에 있는 숙소 '{place_name}'의 짧은 홍보 글을 쓴다.\n"
        f"갈래: {TOPIC_RULES.get(topic_kind, '')}\n"
        f"소재: {material}\n"
        f"이미 쓴 주제: {used}\n\n"
        f"{_RULES}\n본문만 출력한다."
    )


def filter_drafts(rows: list[dict]) -> tuple[list[dict], list[tuple[str, str]]]:
    """(통과한 것, 버린 것[(본문앞부분, 사유)]). 버린 이유를 세어 프롬프트를 고칠 근거로 남긴다."""
    kept, dropped = [], []
    seen_keys = set()
    for row in rows:
        ok, reason = is_publishable_body(row.get("body", ""))
        key = (row.get("topic_key") or "").strip()
        if not ok:
            dropped.append((row.get("body", "")[:24], reason))
            continue
        if not key:
            dropped.append((row.get("body", "")[:24], "주제 키가 없다"))
            continue
        if key in seen_keys:
            dropped.append((row.get("body", "")[:24], f"같은 회차에서 주제 중복: {key}"))
            continue
        seen_keys.add(key)
        # 팀 사전검수 없음 — 금칙 필터를 통과하면 그대로 발송 대상이다.
        kept.append({**row, "topic_key": key, "status": PostStatus.REVIEWED.value})
    if dropped:
        LOG.i(f"[blog] 생성분 {len(rows)}건 중 {len(dropped)}건 버림")
    return kept, dropped


async def generate_one(*, place_name: str, region: str, topic_kind: int, material: str,
                       used_topics: list[str], client=None) -> tuple[str, str] | None:
    """(문구, 모델명) 한 쌍. LLM 이 없거나 실패하면 None — 생성 실패가 잡을 죽이지 않는다.
    모델명은 생성 이력 화면이 "어느 모델썼는지" 보여주는 데 쓴다(2026-09-17, 사장님 지시)."""
    from services.llm.gemini import DEFAULT_MODEL, GeminiError, call, extract_text, is_configured

    if not is_configured():
        return None

    prompt = build_prompt(place_name=place_name, region=region, topic_kind=topic_kind,
                          material=material, used_topics=used_topics)
    body = {
        "contents": [{"role": "user", "parts": [{"text": prompt}]}],
        "generationConfig": {"temperature": 0.9},
    }
    owns = client is None
    if owns:
        import httpx
        client = httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0))
    try:
        payload = await call(client, DEFAULT_MODEL, body)
        text = extract_text(payload).strip()
        return (text, DEFAULT_MODEL) if text else None
    except GeminiError as error:
        LOG.w(f"[blog] 생성 실패: {error}")
        return None
    finally:
        if owns:
            await client.aclose()


def materials(snapshot: dict) -> list[tuple[int, str, str]]:
    """(갈래, topic_key, 소재). 소재가 없는 갈래는 아예 만들지 않는다 — 지어내지 않는다."""
    local = snapshot.get("local") or {}
    out: list[tuple[int, str, str]] = []

    for festival in (local.get("festivals") or [])[:12]:
        name = (festival.get("name") or "").strip()
        if name:
            out.append((PostTopicKind.FESTIVAL.value, f"festival:{name}",
                        f"{name} — {festival.get('location') or ''} {festival.get('description') or ''}".strip()))

    for spot in ((local.get("attractions") or []) + (local.get("restaurants") or []))[:20]:
        name = (spot.get("name") or "").strip()
        if name:
            out.append((PostTopicKind.NEARBY.value, f"nearby:{name}",
                        f"{name} — {spot.get('description') or spot.get('address') or ''}".strip()))

    for sky in ("맑음", "흐림", "비", "눈", "안개", "소나기"):
        out.append((PostTopicKind.WEATHER.value, f"weather:{sky}", f"{sky}인 날"))

    for term in ("봄", "초여름", "한여름", "초가을", "늦가을", "초겨울", "한겨울"):
        out.append((PostTopicKind.SEASON.value, f"season:{term}", term))

    return out