"""SceneStillNormalizer — CheckpointBundle → list[PlannedStill]. Pure 로직. DB 접근 없음. 3경로: 1) Legacy — selected_shots 없을 때 scene당 1 still (scene_extractor_v3) 2) V4 per-shot — 각 selected shot당 1 still (scene_extractor_v4) 3) Unselected — shot-more에서 미선택 shot (shot_more_unselected) visible_entities 해석을 위해 EntityMaps 주입. """ from __future__ import annotations import json import re from typing import Any, Dict, List from app.core.steps._evidence_helpers import _normalize_scene_detail_result from app.services.checkpoint_sync._scene_still_contracts import ( CheckpointBundle, EntityMaps, PlannedStill, ) _BARE_ID_RE = re.compile(r"[CLP]\d{2,3}") def _resolve_ve(ve_list: List[str], maps: EntityMaps) -> List[Dict[str, str]]: return [{ "short_id": sid, "id": maps.short_to_id.get(sid, ""), "entity_name": maps.short_to_name.get(sid, sid), } for sid in ve_list] def _backfill_ve_dict(ve_raw: List[Dict[str, Any]], maps: EntityMaps) -> None: """Codex P3-1 Medium: baseline parity — name→id 매핑 실패 시 빈 문자열을 강제로 주입하지 않음 (baseline은 `if matched_id: ve_item["id"] = ...`). short_id는 baseline과 동일하게 `.get(ename, "")` 폴백 유지. """ for ve_item in ve_raw: if not isinstance(ve_item, dict): continue ename = ve_item.get("entity_name", "") sid = ve_item.get("short_id", "") if sid and not ve_item.get("id"): ve_item["id"] = maps.short_to_id.get(sid, "") if sid and not ename: ve_item["entity_name"] = maps.short_to_name.get(sid, sid) ename = ve_item["entity_name"] if not ve_item.get("id") and ename: matched_id = maps.name_to_id.get(ename) if matched_id: ve_item["id"] = matched_id if not ve_item.get("short_id") and ename: ve_item["short_id"] = maps.name_to_short.get(ename, "") def _shot_ve( bundle: CheckpointBundle, director_ve: List[str], scene_idx: int, shot_idx: int, shot_vars: List[Dict[str, Any]], ) -> List[str]: k = (scene_idx, shot_idx) if k in bundle.shot_director_ve_map: return bundle.shot_director_ve_map[k] if not shot_vars: return director_ve text = " ".join(v.get("t2i_prompt", "") for v in shot_vars) used = set(_BARE_ID_RE.findall(text)) for v in shot_vars: for a in v.get("outfit_assignments", []): cid = a.get("character_id", "") if cid: used.add(cid) return [sid for sid in director_ve if sid in used] class SceneStillNormalizer: def __init__(self, entity_maps: EntityMaps): self.maps = entity_maps def normalize(self, bundle: CheckpointBundle) -> List[PlannedStill]: if not bundle.sd_completed: return [] planned: List[PlannedStill] = [] idx = 0 for s in bundle.scenes: idx = self._plan_scene(s, bundle, idx, planned) # unselected — scene loop 이후 for si, shots in bundle.shot_info_by_scene.items(): sel = bundle.selected_flag_by_scene.get(si, set()) for shot in shots: if shot["shot_index"] in sel: continue idx += 1 planned.append(self._unselected(idx, si, shot)) return planned def _plan_scene(self, s, bundle, idx, out) -> int: # G3.1: 옛 cp 4-field 누락 lazy backfill (마킹=legacy). _normalize_scene_detail_result(s, where="scene_still_normalizer._plan_scene") si = s.get("scene_index", 0) t2i_vars = s.get("t2i_variations", []) all_shots = bundle.shot_info_by_scene.get(si, []) selected_ids = bundle.selected_flag_by_scene.get(si, set()) selected_shots = [sh for sh in all_shots if sh["shot_index"] in selected_ids] director_ve = bundle.scene_director_ve.get(si, []) result_idx = s.get("_shot_index") if result_idx is not None and selected_shots: shot = next((sh for sh in selected_shots if sh["shot_index"] == result_idx), None) if not shot: return idx selected_shots = [shot] if not selected_shots: idx += 1 out.append(self._legacy(si, s, t2i_vars, bundle, idx)) return idx is_partial = result_idx is not None vars_per = len(t2i_vars) if is_partial else max(1, len(t2i_vars) // max(len(selected_shots), 1)) for i, shot in enumerate(selected_shots): idx += 1 start = i * vars_per end = len(t2i_vars) if i == len(selected_shots) - 1 else start + vars_per out.append(self._shot(si, s, shot, t2i_vars[start:end], bundle, director_ve, idx)) return idx def _legacy(self, si, s, t2i_vars, bundle, idx) -> PlannedStill: ve_raw = s.get("visible_entities") or s.get("present_entity_ids") or [] ve_raw = [{"short_id": v} if isinstance(v, str) else v for v in ve_raw] _backfill_ve_dict(ve_raw, self.maps) return PlannedStill( key=(si, None), still_index=idx, scene_index=si, t2i_composer_version="scene_extractor_v3", columns={ "screenplay_scene_heading": s.get("heading", ""), "beat_title": s.get("beat_title", ""), "still_frame_prompt": s.get("representative_moment", ""), "visible_entities_json": json.dumps(ve_raw, ensure_ascii=False), "t2i_prompt_cinematic": s.get("t2i_prompt", ""), "t2i_variations_json": json.dumps(t2i_vars, ensure_ascii=False) if t2i_vars else None, "scene_type": s.get("scene_type", "normal"), "audio_entity_ids": json.dumps(bundle.scene_director_audio.get(si, []), ensure_ascii=False), "hallucination_entity_ids": json.dumps(bundle.scene_director_hall.get(si, []), ensure_ascii=False), }, ) def _shot(self, si, s, shot, shot_vars, bundle, director_ve, idx) -> PlannedStill: shot_idx = shot["shot_index"] ve_ids = _shot_ve(bundle, director_ve, si, shot_idx, shot_vars) ve_raw = _resolve_ve(ve_ids, self.maps) sfp = shot.get("description", "") if s.get("_user_edited") and s.get("representative_moment"): sfp = s["representative_moment"] return PlannedStill( key=(si, shot_idx), still_index=idx, scene_index=si, t2i_composer_version="scene_extractor_v4", columns={ "screenplay_scene_heading": s.get("heading", ""), "beat_title": s.get("beat_title", ""), "still_frame_prompt": sfp, "visible_entities_json": json.dumps(ve_raw, ensure_ascii=False), "t2i_prompt_cinematic": s.get("t2i_prompt", ""), "t2i_variations_json": json.dumps(shot_vars, ensure_ascii=False) if shot_vars else None, "scene_type": s.get("scene_type", "normal"), "shot_index": shot_idx, "shot_description": shot.get("description", ""), "based_on_beat": shot.get("based_on_beat"), "audio_entity_ids": json.dumps(bundle.scene_director_audio.get(si, []), ensure_ascii=False), "hallucination_entity_ids": json.dumps(bundle.scene_director_hall.get(si, []), ensure_ascii=False), "is_selected": True, }, ) def _unselected(self, idx, si, shot) -> PlannedStill: shot_idx = shot["shot_index"] return PlannedStill( key=(si, shot_idx), still_index=idx, scene_index=si, t2i_composer_version="shot_more_unselected", columns={ "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": shot.get("description", ""), "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None, "scene_type": "normal", "shot_index": shot_idx, "shot_description": shot.get("description", ""), "based_on_beat": shot.get("based_on_beat"), "audio_entity_ids": "[]", "hallucination_entity_ids": "[]", "is_selected": False, }, )