"""미니 블로그 — 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 date, datetime, timedelta, timezone from common.enums import LocalContentType, PlaceCategory, 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 # 숙소(LODGING) 기본 갈래 규칙 — 업종별 규칙이 없을 때의 폴백이기도 하다. TOPIC_RULES: dict[int, str] = { # ★ 게시일의 실제 날씨는 모른다(글은 며칠·몇 주 앞서 만든다) — "오늘은 비가 옵니다"라고 단정하게 두지 않는다. PostTopicKind.WEATHER.value: "소재로 주어진 날씨인 날, 이 숙소에서 하기 좋은 일을 한 장면으로 적는다. 게시일의 날씨를 단정하지 않는다.", PostTopicKind.FESTIVAL.value: "주어진 축제 하나를 게시일 기준으로(곧 열리는지, 열리는 중인지) 언급하고, 숙소에서 그곳까지 어떻게 가는지를 걸음 단위로 적는다.", PostTopicKind.SEASON.value: "게시일 무렵 절기에 이 지역과 숙소가 어떻게 달라지는지를 적는다.", PostTopicKind.NEARBY.value: "주어진 주변 장소 하나를 손님 시선에서 적는다. 영업시간과 가격은 쓰지 않는다.", PostTopicKind.GUIDE.value: "확인된 이용 안내 하나를 손님이 알아두면 좋은 말투로 풀어 적는다.", } # 업종별 분기 — 지금은 숙소만 채워져 있다. 새 업종을 넣으려면 여기 두 딕셔너리에만 항목을 더한다. _BUSINESS_NOUN_BY_CATEGORY: dict[int, str] = { PlaceCategory.LODGING.value: "숙소", } _TOPIC_RULES_BY_CATEGORY: dict[int, dict[int, str]] = { PlaceCategory.LODGING.value: TOPIC_RULES, } def _business_noun(place_category: int) -> str: return _BUSINESS_NOUN_BY_CATEGORY.get(place_category, "숙소") def _topic_rules(place_category: int) -> dict[int, str]: return _TOPIC_RULES_BY_CATEGORY.get(place_category, TOPIC_RULES) _RULES = ( "규칙\n" f"- {MIN_LEN}~{MAX_LEN}자 사이 한 문단. 제목·해시태그·이모지를 쓰지 않는다.\n" "- 숫자로 된 요금·시간·인원·전화번호를 쓰지 않는다. 확인되지 않은 주장을 하지 않는다.\n" "- '최고' '유일' 같은 최상급을 쓰지 않는다.\n" "- 손님에게 말하듯 존댓말로 적는다.\n" "- 아래 '이미 쓴 주제'와 겹치는 소재를 고르지 않는다.\n" "- 게시일과 맞지 않는 계절·날씨·행사 이야기를 쓰지 않는다.\n" ) _WEEKDAYS = "월화수목금토일" def season_term(on: date) -> str: """게시일 → 절기 이름(materials 의 계절 소재와 같은 말). 달로만 가른다.""" return { 3: "봄", 4: "봄", 5: "봄", 6: "초여름", 7: "한여름", 8: "한여름", 9: "초가을", 10: "늦가을", 11: "늦가을", 12: "초겨울", 1: "한겨울", 2: "한겨울", }[on.month] def _date_line(on: date) -> str: return f"게시일: {on.year}년 {on.month}월 {on.day}일({_WEEKDAYS[on.weekday()]}) · {season_term(on)}\n" def build_prompt(*, place_name: str, region: str, topic_kind: int, material: str, used_topics: list[str], place_category: int = PlaceCategory.LODGING.value, post_date: date | None = None) -> str: """갈래 하나에 대한 프롬프트 한 벌. 프롬프트를 두 곳에 적지 않으려고 여기서만 만든다. post_date 가 있으면 게시일을 알려 준다 — 글이 그 날짜의 계절·행사와 맞게 쓰이도록(2026-09-23).""" used = ", ".join(used_topics[:40]) or "없음" noun = _business_noun(place_category) rules = _topic_rules(place_category) return ( f"{region}에 있는 {noun} '{place_name}'의 짧은 홍보 글을 쓴다.\n" f"{_date_line(post_date) if post_date else ''}" f"갈래: {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], place_category: int = PlaceCategory.LODGING.value, post_date: date | None = None, client=None) -> tuple[str, str] | None: """(문구, 모델명) 한 쌍. LLM 이 없거나 실패하면 None — 생성 실패가 잡을 죽이지 않는다. 모델명은 생성 이력 화면이 "어느 모델썼는지" 보여주는 데 쓴다(2026-09-17, 사장님 지시). ★ 발행 링크는 여기서 붙이지 않는다 — 호출부가 길이 게이트(is_publishable_body/ filter_drafts, MIN_LEN~MAX_LEN)를 이 반환값 그대로에 건다. 링크까지 포함해서 길이를 재면 정상 문구도 게이트에 걸려 버려진다. 링크는 게이트를 통과한 뒤 호출부가 붙인다. ★ 공급자는 LLM_PROVIDER 설정을 따른다(services/llm/provider.py) — Gemini 로 고정하지 않는다. generate_social_post(services/external/gemini_text.py)와 달리 구조화 출력 재시도 루프가 없는 단순 텍스트 생성이라 공급자를 가려도 된다.""" from services.llm import provider from services.llm.errors import LlmError llm = provider.active() if not llm.is_configured(): return None prompt = build_prompt(place_name=place_name, region=region, topic_kind=topic_kind, material=material, used_topics=used_topics, place_category=place_category, post_date=post_date) owns = client is None if owns: import httpx client = httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0)) try: result = await llm.generate(client, llm.DEFAULT_MODEL, prompt=prompt, temperature=0.9) text = result.text.strip() return (text, llm.DEFAULT_MODEL) if text else None except LlmError as error: LOG.w(f"[blog] 생성 실패: {error}") return None finally: if owns: await client.aclose() # 축제 글을 시작일 며칠 전부터 낼 수 있나. 끝난 축제는 내지 않는다. FESTIVAL_LEAD_DAYS = 14 # 그 달에 말이 되는 날씨만 소재로 쓴다 — 여름에 "눈인 날" 글이 나가지 않게. _SKIES = ("맑음", "흐림", "비", "안개") _SKIES_BY_MONTH = {12: ("눈",), 1: ("눈",), 2: ("눈",), 6: ("소나기",), 7: ("소나기",), 8: ("소나기",)} def _ymd(value) -> date | None: digits = "".join(ch for ch in str(value or "") if ch.isdigit()) if len(digits) != 8: return None try: return date(int(digits[:4]), int(digits[4:6]), int(digits[6:])) except ValueError: return None def materials(snapshot: dict, on: date) -> list[tuple[int, str, str]]: """게시일 on 에 맞는 (갈래, topic_key, 소재). 앞에 있을수록 먼저 고른다 — 축제 → 계절 → 주변 → 날씨. 소재가 없는 갈래는 아예 만들지 않는다 — 지어내지 않는다. ★ 날짜에 맞춘다(2026-09-23). 예전에는 날짜와 무관한 한 줄 목록이라, 9월 날짜에 '한겨울' 글이나 이미 끝난 축제 글이 붙을 수 있었다. - 축제: 시작 FESTIVAL_LEAD_DAYS 일 전 ~ 끝나는 날 사이에만. 기간을 모르는 축제는 쓰지 않는다. - 계절: 게시일의 절기 하나. 키에 연도를 넣어 해마다 한 번씩 다시 쓸 수 있다. - 날씨: 그 달에 있을 법한 것만. 키에 연·월을 넣어 달마다 다시 쓸 수 있다. - 주변 장소: 날짜와 무관해 늘 후보다. ★ 스냅샷의 지역 정보는 원문 행 목록(snapshot["local"]["contents"])이다. 예전 코드는 site_payload 모양(local.festivals·attractions)을 읽어 축제·주변 소재가 늘 비어 있었다.""" contents = (snapshot.get("local") or {}).get("contents") or [] by_type: dict[int, list[dict]] = {} for row in contents: if isinstance(row, dict): by_type.setdefault(row.get("content_type"), []).append(row) out: list[tuple[int, str, str]] = [] for row in by_type.get(LocalContentType.FESTIVAL.value, []): body = row.get("body") or {} name = str(body.get("name") or row.get("title") or "").strip() start = _ymd(body.get("eventstartdate")) end = _ymd(body.get("eventenddate")) or start if not name or start is None or not (start - timedelta(days=FESTIVAL_LEAD_DAYS) <= on <= end): continue period = f"{start:%Y.%m.%d}" + (f" ~ {end:%Y.%m.%d}" if end != start else "") detail = f"{body.get('location') or ''} {str(body.get('overview') or '')[:300]}".strip() out.append((PostTopicKind.FESTIVAL.value, f"festival:{start.year}:{name}"[:120], f"{name} (기간 {period}) — {detail}".strip(" —"))) term = season_term(on) out.append((PostTopicKind.SEASON.value, f"season:{on.year}:{term}", term)) spots = (by_type.get(LocalContentType.ATTRACTION.value, []) + by_type.get(LocalContentType.RESTAURANT.value, [])) for row in spots[:20]: body = row.get("body") or {} name = str(body.get("name") or row.get("title") or "").strip() if name: detail = body.get("description") or body.get("overview") or body.get("location") or "" out.append((PostTopicKind.NEARBY.value, f"nearby:{name}"[:120], f"{name} — {detail}".strip(" —"))) for sky in _SKIES + _SKIES_BY_MONTH.get(on.month, ()): out.append((PostTopicKind.WEATHER.value, f"weather:{on:%Y-%m}:{sky}", f"{sky}인 날")) return out