o2o-site-AEO/solution/backend/services/grounding/story.py

"""지역 이야기 응답 해석 — 모델이 준 JSON 에서 **쓸 수 있는 항목만** 남긴다.

★ 왜 스키마 검증을 하지 않나
  항목 모양의 단일 출처는 `shared/lib/section-data.ts` 다. 그 모양을 파이썬에 한 벌 더 적으면,
  프론트가 필드를 하나 늘린 날 서버가 그걸 조용히 떨어뜨린다 — `site_payload._sections` 가
  붙여넣기 아이템을 파싱하지 않는 것과 같은 이유다.
  그래서 여기서는 **그 항목이 화면에 설 수 있는가**만 본다: 종류마다 하나씩 있는 '이름 칸'.

★ 출처는 두 곳에서 온다
  모델이 항목에 단 `source` 가 1순위다. 그게 없으면 Perplexity 가 실제로 읽은
  `search_results` 의 첫 줄을 붙인다 — 모델 답변은 환각이 섞이지만 search_results 는
  실제로 검색된 주소다(`grounding/channels.py` 와 같은 판단).
  둘 다 없으면 항목을 버린다. 출처 없는 사실은 이 레포의 규칙 위반이다.
"""
import json
import re

from common.logger import LOG

# 종류별 '이름 칸' — 이게 비면 화면에 세울 수 없다(제목 없는 카드가 된다).
_TITLE_KEY = {
    "songs": "title",
    "people": "name",
    "chronicle": "title",
    "postcard": "line",
    "quiz": "question",
}

# 코드펜스를 두르고 오는 경우가 있다. 규칙 1 로 금지했지만 모델은 종종 어긴다.
_FENCE_RE = re.compile(r"^\s*```(?:json)?\s*|\s*```\s*$", re.MULTILINE)


def _payload_text(payload: dict) -> str:
    choices = payload.get("choices") or []
    if not choices or not isinstance(choices[0], dict):
        return ""
    return ((choices[0].get("message") or {}).get("content")) or ""


def _first_source(payload: dict) -> dict | None:
    """Perplexity 가 실제로 읽은 첫 출처. 항목에 source 가 없을 때의 대체값."""
    for row in payload.get("search_results") or []:
        if isinstance(row, dict) and (row.get("url") or "").startswith("http"):
            return {"name": row.get("title") or row.get("url"), "url": row["url"]}
    return None


def _clean_source(value) -> dict | None:
    """모델이 준 source. url 이 http 로 시작하지 않으면 없는 것으로 친다 —
    "검색결과 참조" 같은 문자열이 그대로 링크가 되면 눌러도 아무 데도 안 간다."""
    if not isinstance(value, dict):
        return None
    url = (value.get("url") or "").strip()
    if not url.startswith("http"):
        return None
    return {"name": (value.get("name") or url).strip(), "url": url}


def parse_items(payload: dict, kind: str, limit: int) -> tuple[list[dict], list[str]]:
    """(쓸 수 있는 항목, 버린 이유) — 버린 이유는 로그와 잡 결과에 남긴다.

    한 항목이 잘못돼도 나머지를 살린다. 지역 하나에 8~14건인데 한 줄 때문에 전부 버리면
    그 지역은 다음 재생성까지 빈 채로 남는다.
    """
    title_key = _TITLE_KEY.get(kind)
    if title_key is None:
        raise ValueError(f"모르는 지역 이야기 종류: {kind}")

    text = _FENCE_RE.sub("", _payload_text(payload)).strip()
    if not text:
        return [], ["응답이 비었다"]

    try:
        envelope = json.loads(text)
    except (json.JSONDecodeError, ValueError) as ex:
        LOG.w(f"[story] {kind} JSON 파싱 실패: {ex}")
        return [], [f"JSON 이 아니다: {ex}"]

    if not isinstance(envelope, dict):
        return [], ["최상위가 객체가 아니다"]
    raw_items = envelope.get("items")
    if not isinstance(raw_items, list):
        return [], ["items 가 배열이 아니다"]

    fallback = _first_source(payload)
    out: list[dict] = []
    dropped: list[str] = []

    for raw in raw_items:
        if len(out) >= limit:
            break
        if not isinstance(raw, dict):
            dropped.append("항목이 객체가 아니다")
            continue
        title = (raw.get(title_key) or "").strip() if isinstance(raw.get(title_key), str) else ""
        if not title:
            dropped.append(f"{title_key} 가 없다")
            continue

        item = {k: v for k, v in raw.items() if v not in (None, "", [], {})}
        item[title_key] = title

        source = _clean_source(raw.get("source")) or fallback
        if source is None:
            dropped.append(f"{title}: 출처가 없다")
            continue
        item["source"] = source

        # ★ 모델이 "확인" 이라고 우겨도, 대체 출처로 때운 항목은 확인필요다 —
        #   그 URL 은 이 항목이 아니라 이번 검색 전체의 출처다.
        if item.get("verified") not in ("확인", "확인필요"):
            item["verified"] = "확인필요"
        elif _clean_source(raw.get("source")) is None:
            item["verified"] = "확인필요"

        out.append(item)

    return out, dropped