[feat] solution/backend: 일정 응답 해석 — 같은 정거장 집합인 코스는 버린다

solution/backend/services/grounding/itinerary.py

@@ -0,0 +1,129 @@
+"""일정 응답 해석 — 모델이 준 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

solution/backend/tests/test_itinerary_grounding.py

@@ -0,0 +1,148 @@
+"""일정 응답 해석 — 모델이 준 JSON 에서 화면에 설 수 있는 코스만 남긴다.
+
+DB 도 네트워크도 쓰지 않는 순수 단위 테스트다.
+
+★ 스키마 전체를 검증하지 않는다. 항목 모양의 단일 출처는 `shared/lib/section-data.ts` 이고,
+  그 모양을 파이썬에 한 벌 더 적으면 프론트가 필드를 늘린 날 서버가 조용히 떨어뜨린다
+  (`grounding/story.py` 머리주석과 같은 판단). 여기서는 "화면에 설 수 있는가" 만 본다.
+"""
+import json
+
+from services.grounding import itinerary as grounding
+
+
+def _payload(items: list, *, fence: bool = False, search: list | None = None) -> dict:
+    envelope = {"kind": "itinerary", "version": 1, "title": "추천 일정", "items": items}
+    text = json.dumps(envelope, ensure_ascii=False)
+    if fence:
+        text = f"```json\n{text}\n```"
+    return {
+        "choices": [{"message": {"content": text}}],
+        "search_results": search if search is not None else [
+            {"title": "군산문화관광", "url": "https://www.gunsan.go.kr/tour/"}
+        ],
+    }
+
+
+def _course(name: str, stops: list[str], *, days: int = 2) -> dict:
+    per_day = max(1, len(stops) // days)
+    chunks = [stops[i:i + per_day] for i in range(0, len(stops), per_day)][:days]
+    return {
+        "name": name,
+        "duration": "1박 2일",
+        "audience": "누구에게나",
+        "why": "이유 한 문장.",
+        "days": [
+            {
+                "label": f"{i + 1}일차",
+                "startTime": "14:00",
+                "stops": [
+                    {"name": s, "minutes": 60, "moveMinutes": 10, "searchQuery": s,
+                     "latitude": 35.99, "longitude": 126.71}
+                    for s in chunk
+                ],
+            }
+            for i, chunk in enumerate(chunks)
+        ],
+    }
+
+
+def test_accepts_five_distinct_courses():
+    """검증: 정상 응답 5개.
+    기대결과: 전부 채택되고 버린 것이 없다."""
+    items = [_course(f"코스{i}", [f"장소{i}-{j}" for j in range(4)]) for i in range(5)]
+    courses, dropped = grounding.parse_courses(_payload(items), "1박 2일")
+    assert len(courses) == 5
+    assert dropped == []
+    assert [c["name"] for c in courses] == [f"코스{i}" for i in range(5)]
+
+
+def test_strips_code_fence():
+    """검증: 코드펜스를 두르고 온 응답.
+    기대결과: 규칙 1 로 금지했지만 모델은 종종 어긴다 — 벗겨서 읽는다."""
+    items = [_course("코스", ["가", "나"])]
+    courses, dropped = grounding.parse_courses(_payload(items, fence=True), "1박 2일")
+    assert len(courses) == 1
+
+
+def test_forces_duration_even_if_model_wrote_something_else():
+    """검증: 모델이 duration 을 다르게 적은 코스.
+    기대결과: 우리가 요청한 기간으로 덮어쓴다 — 이 값이 화면 탭을 가르므로 틀리면 탭이 갈린다."""
+    item = _course("코스", ["가", "나"])
+    item["duration"] = "반나절"
+    courses, _ = grounding.parse_courses(_payload([item]), "1박 2일")
+    assert courses[0]["duration"] == "1박 2일"
+
+
+def test_drops_course_without_name():
+    """검증: 이름 없는 코스.
+    기대결과: 버린다 — 제목 없는 카드가 된다."""
+    item = _course("", ["가", "나"])
+    courses, dropped = grounding.parse_courses(_payload([item]), "1박 2일")
+    assert courses == []
+    assert dropped and "name" in dropped[0]
+
+
+def test_drops_course_without_stops():
+    """검증: days 가 있어도 정거장이 하나도 없는 코스.
+    기대결과: 버린다 — 발행본 ItinerarySection 이 어차피 걸러 빈 탭을 만든다."""
+    item = {"name": "빈 코스", "duration": "1박 2일", "days": [{"label": "1일차", "stops": []}]}
+    courses, dropped = grounding.parse_courses(_payload([item]), "1박 2일")
+    assert courses == []
+    assert dropped
+
+
+def test_drops_a_course_whose_stop_set_duplicates_an_earlier_one():
+    """검증: 앞 코스와 정거장 집합이 완전히 같은 코스(순서만 다름).
+    기대결과: 뒤에 온 것을 버린다 — 손님 눈에는 같은 코스 둘이다.
+    ★ 정거장 '겹침' 자체는 허용이다(2026-09-11 결정). 집합이 **같을 때**만 버린다."""
+    first = _course("원도심 코스", ["가", "나", "다", "라"])
+    same = _course("미식 코스", ["라", "다", "나", "가"])   # 순서만 다르다
+    overlap = _course("자연 코스", ["가", "나", "마", "바"])  # 절반 겹치지만 다른 코스다
+    courses, dropped = grounding.parse_courses(_payload([first, same, overlap]), "1박 2일")
+    assert [c["name"] for c in courses] == ["원도심 코스", "자연 코스"]
+    assert dropped and "미식 코스" in dropped[0]
+
+
+def test_attaches_search_result_as_source_when_model_gave_none():
+    """검증: 코스에 source 가 없을 때.
+    기대결과: Perplexity 가 실제로 읽은 첫 출처를 붙인다 — 모델 답변보다 이게 실재한다.
+    ★ source 가 아예 없으면 발행본 카드 하단에 빈 점선 띠가 남는다(ItinerarySection 껍데기)."""
+    courses, _ = grounding.parse_courses(_payload([_course("코스", ["가", "나"])]), "1박 2일")
+    assert courses[0]["source"]["url"] == "https://www.gunsan.go.kr/tour/"
+
+
+def test_keeps_model_source_when_it_is_a_real_url():
+    """검증: 모델이 준 source 가 http 로 시작할 때.
+    기대결과: 그걸 쓴다."""
+    item = _course("코스", ["가", "나"])
+    item["source"] = {"name": "군산시", "url": "https://www.gunsan.go.kr/index.gunsan"}
+    courses, _ = grounding.parse_courses(_payload([item]), "1박 2일")
+    assert courses[0]["source"]["url"] == "https://www.gunsan.go.kr/index.gunsan"
+
+
+def test_course_survives_without_any_source():
+    """검증: 모델 source 도 없고 search_results 도 빈 응답.
+    기대결과: 코스는 살린다 — 출처가 없다고 일정을 버리면 화면이 통째로 빈다.
+    (지역 이야기와 다른 판단이다. 저쪽은 '사실' 이고 이쪽은 '제안' 이다.)"""
+    items = [_course("코스", ["가", "나"])]
+    courses, _ = grounding.parse_courses(_payload(items, search=[]), "1박 2일")
+    assert len(courses) == 1
+    assert "source" not in courses[0]
+
+
+def test_returns_empty_on_non_json():
+    """검증: JSON 이 아닌 응답.
+    기대결과: 빈 목록 + 이유. 예외를 던지지 않는다 — 잡이 죽으면 다른 기간도 못 받는다."""
+    payload = {"choices": [{"message": {"content": "죄송합니다, 일정을 만들 수 없습니다."}}]}
+    courses, dropped = grounding.parse_courses(payload, "1박 2일")
+    assert courses == []
+    assert dropped
+
+
+def test_returns_empty_when_items_is_not_a_list():
+    """검증: items 가 배열이 아닌 응답."""
+    payload = {"choices": [{"message": {"content": json.dumps({"items": "없음"})}}]}
+    courses, dropped = grounding.parse_courses(payload, "1박 2일")
+    assert courses == []
+    assert dropped