"""지역 이야기 응답 해석 — 모델이 준 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", "daily": "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