diff --git a/backend/app/services/scene_reference_service.py b/backend/app/services/scene_reference_service.py index d6f9e545..a7be130d 100644 --- a/backend/app/services/scene_reference_service.py +++ b/backend/app/services/scene_reference_service.py @@ -1552,12 +1553,43 @@ class SceneReferenceService: ) return (label, best_prev_bytes, loc_id_resolved, ref_role, ref_role_metadata) + def load_identity_family_by_sid(self, episode_id: str) -> Dict[str, Set[str]]: + """entity_relation 체크포인트 → character identity-variant family 맵 (인스턴스 캐시). + + detect_state_variant_sids 의 identity-variant aware 매칭 입력 (2026-07-02, + S12 시신: staging=base 표기 vs VE=variant 표기 어긋남 수정). cp 부재/파손 시 + 빈 dict — 소비자는 기존 exact 매칭으로 환원. + """ + cache: Dict[str, Dict[str, Set[str]]] = getattr( + self, "_identity_family_cache", {}) + if episode_id in cache: + return cache[episode_id] + fam: Dict[str, Set[str]] = {} + try: + from app.core.config import settings + from app.modules.pipeline.visual_continuity_anchor_plan import ( + build_identity_family_by_sid, + ) + from app.services.scene_checkpoint_loaders import _ep_checkpoint_path + p = _ep_checkpoint_path( + settings.projects_dir, self._project_id, episode_id, "entity_relation") + if p.exists(): + relations = ((json.loads(p.read_text(encoding="utf-8")).get("data") or {}) + .get("relations") or []) + fam = build_identity_family_by_sid(relations) + except Exception as exc: # noqa: BLE001 — 비차단(기존 동작 환원) + logger.warning("identity family 로드 실패 (episode=%s): %s", episode_id, exc) + cache[episode_id] = fam + self._identity_family_cache = cache + return fam + def detect_state_variant_sids( self, visible_entities: List[Dict[str, Any]], entity_lookup: Dict[str, Dict[str, Any]], scene_ref_image_map: Dict[str, bytes], staging: Optional[Dict[str, Any]], + identity_family_by_sid: Optional[Dict[str, Set[str]]] = None, ) -> Dict[str, Dict[str, str]]: """Detect state_variant short_ids from staging subject_state. @@ -1578,6 +1610,7 @@ class SceneReferenceService: if not staging: return state_variant_sids + fam_map = identity_family_by_sid or {} sid_to_uuid = { ve.get("short_id", ""): ve.get("id", "") for ve in visible_entities if ve.get("short_id") @@ -1587,18 +1620,44 @@ class SceneReferenceService: for eid, info in entity_lookup.items() if info.get("short_id") }) + # 이름→sid (episode 전체 lookup, exact equality) — staging 이 base 표기, + # VE 가 variant 표기(별도 EntityCanon)일 때 identity family 로 잇는 브리지. + name_to_sids: Dict[str, Set[str]] = {} + for _eid, info in entity_lookup.items(): + nm, s = info.get("name"), info.get("short_id") + if nm and s: + name_to_sids.setdefault(nm, set()).add(s) + for ca in staging.get("character_angles", []): state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if not is_immobilized_state(state): continue ca_name = ca.get("character", "") + char_sid = "" for ve in visible_entities: if ve.get("name") == ca_name and ve.get("short_id"): char_sid = ve["short_id"] - char_uuid = sid_to_uuid.get(char_sid, "") - sv_key = f"state_variant:{char_uuid}:{state}" - if sv_key in scene_ref_image_map: - state_variant_sids[char_sid] = {"key": sv_key, "state": state} + break + if not char_sid and fam_map: + # ★identity-variant aware (2026-07-02): staging 인물이 VE 에 variant + # EntityCanon 으로 존재하면 같은 인물로 매칭 (모호 이름 2+ 는 미적용). + cands = name_to_sids.get(ca_name) or set() + if len(cands) == 1: + fam = fam_map.get(next(iter(cands))) or set() + for ve in visible_entities: + if ve.get("short_id") and ve["short_id"] in fam: + char_sid = ve["short_id"] + break + if not char_sid: + continue + # state variant 자산 소유자는 family 내 다른 멤버(base EntityCanon)일 수 + # 있다 — VE sid 우선, 이후 family 순회로 sv_key 를 찾는다. + for cand_sid in [char_sid] + sorted( + (fam_map.get(char_sid) or set()) - {char_sid}): + cand_uuid = sid_to_uuid.get(cand_sid, "") + sv_key = f"state_variant:{cand_uuid}:{state}" + if cand_uuid and sv_key in scene_ref_image_map: + state_variant_sids[char_sid] = {"key": sv_key, "state": state} break return state_variant_sids diff --git a/backend/tests/services/test_scene_reference_service.py b/backend/tests/services/test_scene_reference_service.py index 7da04273..73f1c888 100644 --- a/backend/tests/services/test_scene_reference_service.py +++ b/backend/tests/services/test_scene_reference_service.py @@ -502,6 +527,49 @@ def test_detect_state_variant_sids_unconscious_severely_injured_both_recognized( assert result["C02"]["state"] == "severely_injured" +def test_detect_state_variant_sids_identity_variant_family_bridge(svc): + """★identity-variant aware (2026-07-02, S12 시신 실측 재현). + + staging 이름=base(C05) 표기 / VE 에는 variant EntityCanon(C16)만 실재 / + state variant 자산 소유자는 base uuid → family 브리지로 매칭·키 해결. + """ + staging = {"character_angles": [ + {"character": "Base-Woman", "gaze_direction_kind": "closed_eyes", + "subject_state": "dead"}]} + visible = [{"id": "c16_uuid", "short_id": "C16", "name": "Base-Woman (변형)"}] + entity_lookup = { + "c05_uuid": {"short_id": "C05", "name": "Base-Woman"}, + "c16_uuid": {"short_id": "C16", "name": "Base-Woman (변형)"}, + } + refs = {"state_variant:c05_uuid:dead": b"DEAD_REF"} + fam = {"C05": {"C05", "C16"}, "C16": {"C05", "C16"}} + + # family 없으면 기존 동작 = 매칭 실패 + assert svc.detect_state_variant_sids(visible, entity_lookup, refs, staging) == {} + + result = svc.detect_state_variant_sids( + visible, entity_lookup, refs, staging, identity_family_by_sid=fam) + assert result == {"C16": {"key": "state_variant:c05_uuid:dead", "state": "dead"}} + + +def test_detect_state_variant_sids_family_ambiguous_name_not_applied(svc): + """동명 2+ 후보면 family 브리지 미적용 (기존 보수 계약 유지).""" + staging = {"character_angles": [ + {"character": "Twin", "gaze_direction_kind": "closed_eyes", + "subject_state": "dead"}]} + visible = [{"id": "c16_uuid", "short_id": "C16", "name": "Twin (변형)"}] + entity_lookup = { + "a_uuid": {"short_id": "C05", "name": "Twin"}, + "b_uuid": {"short_id": "C07", "name": "Twin"}, + "c16_uuid": {"short_id": "C16", "name": "Twin (변형)"}, + } + refs = {"state_variant:a_uuid:dead": b"X"} + fam = {"C05": {"C05", "C16"}, "C16": {"C05", "C16"}} + result = svc.detect_state_variant_sids( + visible, entity_lookup, refs, staging, identity_family_by_sid=fam) + assert result == {} + + def test_detect_state_variant_sids_no_ref_in_map_skipped(svc): """scene_ref_image_map에 state_variant 키 없으면 매핑 스킵.""" # Area #2 W5: v13 3 field shape.