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

130 lines
5.3 KiB
Python

"""일정 응답 해석 — 모델이 준 JSON 에서 **화면에 설 수 있는 코스만** 남긴다.
★ 스키마 검증을 하지 않는다
항목 모양의 단일 출처는 `shared/lib/section-data.ts` 다. 그 모양을 파이썬에 한 벌 더 적으면
프론트가 필드를 하나 늘린 날 서버가 그걸 조용히 떨어뜨린다(`grounding/story.py` 와 같은 판단).
여기서 보는 것은 셋뿐이다 — 이름이 있나 · 정거장이 하나라도 있나 · 앞 코스와 같은 코스인가.
★ 같은 코스를 버린다 (2026-09-11 결정)
정거장 **겹침은 허용**이다. 다만 정거장 집합이 완전히 같으면 순서만 바꾼 것이고, 손님 눈에는
같은 코스 둘이다. 프롬프트 규칙 7 로도 막지만 그건 부탁이지 보장이 아니다 —
실제로 컨셉을 지정하기 전에는 5개 중 4개가 같은 집합이었다(스파이크 실측).
★ duration 은 우리가 덮어쓴다
이 값이 화면 탭을 가른다(`ItinerarySection` 이 `duration` 으로 탭을 세운다).
모델이 "반나절" 이라고 적어 버리면 1박 2일을 요청해 받은 코스가 엉뚱한 탭에 선다.
★ 출처가 없어도 코스는 살린다
지역 이야기는 출처 없는 항목을 버린다 — 그건 '사실' 이라서다. 일정은 '제안' 이고,
출처를 이유로 버리면 화면이 통째로 빈다. 대신 Perplexity 가 실제로 읽은 첫 출처를 붙여 준다.
"""
import json
import re
from common.logger import LOG
# 코드펜스를 두르고 오는 경우가 있다. 규칙 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["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 _stop_names(course: dict) -> list[str]:
"""코스의 정거장 이름 — days 를 펴서 모은다."""
out = []
for day in course.get("days") or []:
if not isinstance(day, dict):
continue
for stop in day.get("stops") or []:
if isinstance(stop, dict):
name = (stop.get("name") or "").strip() if isinstance(stop.get("name"), str) else ""
if name:
out.append(name)
return out
def parse_courses(payload: dict, duration: str) -> tuple[list[dict], list[str]]:
"""(쓸 수 있는 코스, 버린 이유) — 버린 이유는 로그와 잡 결과에 남긴다.
한 코스가 잘못돼도 나머지를 살린다. 기간당 5개인데 한 줄 때문에 전부 버리면
그 업장은 다음 재생성까지 빈 채로 남는다.
"""
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"[itinerary] {duration} 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] = []
seen_stop_sets: list[frozenset[str]] = []
for raw in raw_items:
if not isinstance(raw, dict):
dropped.append("코스가 객체가 아니다")
continue
name = (raw.get("name") or "").strip() if isinstance(raw.get("name"), str) else ""
if not name:
dropped.append("name 이 없다")
continue
stops = _stop_names(raw)
if not stops:
dropped.append(f"{name}: 정거장이 없다")
continue
stop_set = frozenset(stops)
if stop_set in seen_stop_sets:
dropped.append(f"{name}: 앞 코스와 정거장 집합이 같다")
continue
course = {k: v for k, v in raw.items() if v not in (None, "", [], {})}
course["name"] = name
course["duration"] = duration
source = _clean_source(raw.get("source")) or fallback
if source is not None:
course["source"] = source
else:
course.pop("source", None)
seen_stop_sets.append(stop_set)
out.append(course)
return out, dropped