"""콘텐츠 로더 — content/ YAML 메모리 로드 (MVP 구현). 상위 계층(services)은 이 인터페이스만 사용한다. 2단계에서 GraphRAG(kb_query 활성) 구현으로 교체해도 services 코드가 바뀌지 않아야 한다 (ablation 요건). 스포일러 스코프: 카드에는 전역 순서(order)가 있다. 모든 조회는 `order(card) <= order(현재 카드)` 필터를 강제한다 (beat_id ≤ 현재). """ from __future__ import annotations import hashlib from pathlib import Path from engine.schemas.beatcard import BeatCard, RCard from engine.schemas.character import CharacterSheet from engine.schemas.chapter import Chapter from engine.schemas.validate import check_links, load_content class ContentError(RuntimeError): pass class ContentRepository: def __init__(self, root: str | Path = "content"): self.root = Path(root) episodes, r_cards, chapter, characters, errors = load_content(self.root) link_errors, _ = check_links(episodes, chapter, r_cards) errors += link_errors if errors: raise ContentError("검증 실패 콘텐츠는 적재하지 않는다:\n" + "\n".join(errors)) if chapter is None: raise ContentError("챕터 메타 없음") self.chapter: Chapter = chapter self.r_cards: dict[str, RCard] = r_cards self.characters: dict[str, CharacterSheet] = { c.id: c for c in (characters.characters if characters else []) } self._cards: dict[str, BeatCard] = {} self._order: dict[str, int] = {} self.episode_entry: dict[str, str] = {} i = 0 for ep in episodes: self.episode_entry[ep.episode.id] = ep.cards[0].id for card in ep.cards: self._cards[card.id] = card self._order[card.id] = i i += 1 # 콘텐츠 버전 스탬프 — 세션 복원 시 불일치 검사용 (카드 ID/순서 기준. # 본문만 바뀐 갱신은 감지 못하지만 복원이 깨지지는 않는다) ordered_ids = sorted(self._order, key=self._order.get) self.content_version: str = hashlib.sha256( "|".join([self.chapter.id, *ordered_ids]).encode()).hexdigest()[:16] # ── 조회 (스포일러 필터 강제) ──────────────────────────── def get_card(self, card_id: str, current_id: str | None = None) -> BeatCard: """카드 조회. current_id를 주면 beat_id ≤ 현재 필터를 강제한다.""" card = self._cards.get(card_id) if card is None: raise ContentError(f"카드 없음: {card_id}") if current_id is not None and self.order(card_id) > self.order(current_id): raise ContentError(f"스포일러 필터: {card_id}는 현재({current_id}) 이후 카드") return card def order(self, card_id: str) -> int: return self._order[card_id] @property def total_cards(self) -> int: return len(self._cards) def has_card(self, card_id: str) -> bool: return card_id in self._cards def all_cards(self) -> list[BeatCard]: """전역 순서의 전체 카드 (페이지 파티션 등 정적 계산용 — 스포일러 필터 대상 아님).""" return [self._cards[cid] for cid in sorted(self._order, key=self._order.get)] def cards_upto(self, current_id: str) -> list[BeatCard]: """현재 시점까지의 카드 전부 (컨텍스트 주입용).""" cut = self.order(current_id) return [c for cid, c in self._cards.items() if self._order[cid] <= cut] def resolve_ref(self, target: str) -> str | None: """카드 ID 또는 에피소드 참조(E02)를 실제 카드 ID로. 챕터 밖이면 None(출구).""" if target in self._cards: return target if target in self.episode_entry: return self.episode_entry[target] return None # E07 등 — 챕터 출구