"""아웃룩 추출 v2 — 3단계: 목록 추출 -> 씬별 매핑 -> 의상 정리.""" import logging from typing import Dict, List, Optional from app.modules.llm.llm_client import call_structured from app.modules.pipeline import segment_key as segkey from app.modules.prompt_loader import load_prompt, load_schema from app.core.entity_identity import NULL_OUTLOOK_SHORT_ID logger = logging.getLogger(__name__) _MODULE = "outlook_extractor" def extract_outlooks_phase1( segments: List[Dict], characters: List[Dict] = None, scene_character_map: Optional[Dict[int, List[str]]] = None, visual_rules: str = "", fulltext: str = "", project_config: Optional[Dict] = None, opik_metadata: Optional[Dict] = None, prior_roster: Optional[tuple] = None, carry_out: Optional[Dict] = None, planning_block: str = "", ) -> Dict: """1단계: 전체 씬에서 아웃룩 목록 추출 (이름 + 설명). ★``planning_block`` — 기획서 인물 절 (2026-09-18). 기획서에 적힌 의상· 소지품(예: 방진 마스크·오버롤·공구 벨트·인이어 무전기)은 시나리오 본문에 안 적히는 일이 잦아 **아웃룩에 아예 안 실렸다**. 이 단계가 기획서를 못 보던 것이 그 원인이라 여기서 실어 준다. 비면 조립이 한 바이트도 안 바뀐다. ★``prior_roster`` — 앞 화들에서 확정된 아웃룩 명부 `(블록, 허용 ID)`. 발급기만 프로젝트 범위로 바꾸면 충돌은 막지만 **같은 옷의 재사용**은 못 한다 — 화마다 새 번호를 받는다 (Codex 2026-09-04). 명부에는 그 옷을 **입는 인물**이 앵커로 실려 같은 이름의 다른 옷이 갈린다. ★비면(첫 화) 프롬프트도 스키마도 한 바이트 안 바뀐다. """ system = load_prompt(_MODULE, "phase1") schema = load_schema(_MODULE, "phase1_schema") # Build character block with appearance counts appearance_counts: Dict[str, int] = {} if scene_character_map: for _si, char_ids in scene_character_map.items(): for cid in char_ids: appearance_counts[cid] = appearance_counts.get(cid, 0) + 1 char_lines = [] for c in characters: csid = c.get("short_id", c["name"]) count = appearance_counts.get(csid, 0) if count > 0: char_lines.append(f"- {csid}: {c['name']} (등장 {count}씬)") else: char_lines.append(f"- {csid}: {c['name']}") char_block = "\n".join(char_lines) scene_block = _build_scene_block(segments, fulltext) rules_block = f"세계관:\n{visual_rules}\n\n" if visual_rules else "" roster_block, _prior_allowed = (prior_roster or ("", [])) if _prior_allowed: from app.modules.pipeline.episode_carry import patch_schema_with_prior_ids schema = patch_schema_with_prior_ids(schema, _prior_allowed) user_prompt = (f"{rules_block}캐릭터:\n{char_block}\n\n{scene_block}" f"{roster_block}{planning_block}") result = call_structured( step="outlook_extraction", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="outlook_phase1", opik_metadata=opik_metadata, ) if _prior_allowed: # ★모델이 고른 앞 화 신원을 **대조한다** — 명부 밖은 안 받고, 둘이 # 같은 것을 주장하면 합치지 않는다(fail-closed). from app.modules.pipeline.episode_carry import apply_prior_ids _c = apply_prior_ids(result.get("outlooks") or [], _prior_allowed) logger.info("outlook_phase1 앞 화 이관: %s", _c["counts"]) if carry_out is not None: carry_out.update(_c) return result def extract_outlooks_phase2( segments: List[Dict], outlooks: List[Dict] = None, characters: List[Dict] = None, scene_character_map: Optional[Dict[int, List[str]]] = None, fulltext: str = "", project_config: Optional[Dict] = None, opik_metadata: Optional[Dict] = None, ) -> Dict: """2단계: 의상 카탈로그(ID) + 씬별 인물 → 씬별 매핑. enum 기반.""" import copy as _copy system = load_prompt(_MODULE, "phase2") schema = _copy.deepcopy(load_schema(_MODULE, "phase2_schema")) # 의상 카탈로그 (O01, O02, ...) catalog_lines = [] outlook_ids = [] for ol in outlooks: osid = ol.get("short_id", "") outlook_ids.append(osid) cid = ol.get("character_id", "") cid_tag = f" ({cid})" if cid else "" catalog_lines.append(f"- {osid}: {ol['name']}{cid_tag} — {ol.get('description', '')}") # schema에 enum 주입 — ★`required` 에도 넣는다. # # 2026-08-07 실측: enum 만 주입하고 required 에 안 넣었더니, 필수 필드는 # 자유 텍스트 `outlook_name` 뿐이라 **모델이 선택 필드인 outlook_id 를 # 채울지 말지가 실행마다 갈렸다**. 같은 코드·같은 스키마로 한 실행은 # 309/309 를 채웠고 다른 실행은 13/309 만 채웠다. 후자에서 phase3 의 # 미배정 검출(outlook_id 로만 센다)이 118건을 미배정으로 오인해 아웃룩 # 8개를 삭제했고, 하류 scene_detail 이 전량 거부됐다. # # 어느 아웃룩인지 **판단은 LLM 이** 한다. 스키마는 그 답을 카탈로그 id 라는 # 한 가지 형식으로 받을 뿐이다 — 코드가 이름을 보고 대신 맞추지 않는다. if outlook_ids: _item = (schema["properties"]["scene_assignments"]["items"] ["properties"]["assignments"]["items"]) _item["properties"]["outlook_id"] = { "type": "string", "enum": outlook_ids, "description": "카탈로그에서 고른 의상의 id — 반드시 채운다", } _req = _item.get("required") if isinstance(_req, list) and "outlook_id" not in _req: _item["required"] = _req + ["outlook_id"] # 씬별 인물 목록 scene_chars_block = "" if scene_character_map: lines = [] for si in sorted(scene_character_map.keys()): chars = scene_character_map[si] if chars: lines.append(f"씬 {si}: {', '.join(chars)}") scene_chars_block = ( "씬별 인물 (카탈로그에 그 인물 소유 아웃룩이 있을 때만 배정):\n" + "\n".join(lines) + "\n\n" ) # 씬 키를 불투명 토큰으로 잠근다 — 본문의 번호 헤딩과 충돌하지 않게. # 2026-08-07 실측: 이 잠금이 없어 한 실행이 씬 인덱스를 대본 번호(1~116, # 114행)로 되돌려줬고 아웃룩이 한 칸 밀려 배정됐다. scene_director 와 같은 # 결함이라 같은 헬퍼로 잠근다. seg_keys = segkey.segment_keys(len(segments)) key_to_index = segkey.key_index_map(seg_keys, segments) segkey.pin_key_field(schema["properties"]["scene_assignments"]["items"], seg_keys) # 회신 의무는 **인물이 배정될 씬**에만 있다. 블록은 씬 전체를 보내되 # (앞뒤 맥락이 배정 판단의 재료다), 인물 0명 씬은 배정할 대상이 없어 # 행이 없는 것이 정상이라 '누락' 판정에서 뺀다 — 회신은 허용한다. # 인물 맵이 아예 비어 오면 좁힐 근거가 없으니 기존대로 전 키를 요구한다. _char_map = scene_character_map or {} required_keys = ( [k for k in seg_keys if _char_map.get(key_to_index[k])] if _char_map else list(seg_keys) ) # 좁히기가 '전부 빗나가' 빈 목록이 되면 파리티가 아무것도 안 지킨다 — # 빈 회신도, 전 키에 빈 배정을 채운 회신도 통과한다. 그 상태는 인물 맵의 # 씬 번호 공간이 세그먼트 번호 공간과 어긋난 것뿐이고(서수 대체 경로), # 그때 받아 낸 배정은 근거가 없다. 부르기 전에 멈춘다. # ★조건이 좁은 것이 중요하다: 값이 전부 빈 맵(배정할 인물이 아예 없음) # 이나, 번호가 조금이라도 겹치는 경우(08-07 형태의 한 칸 밀림)에는 안 # 터진다 — 지금 도는 실행을 새로 깨뜨리지 않는다. if any(_char_map.values()) and not required_keys: from app.core.errors import AppError raise AppError( code="outlook_phase2.scene_key_space_mismatch", message=( f"인물 맵의 씬 번호 {sorted(map(str, _char_map))[:5]} 가 " f"세그먼트 번호 {sorted(set(key_to_index.values()))[:5]} 와 " f"하나도 겹치지 않는다 — 배정 기준이 없어 중단한다"), status_code=502, ) scene_block = segkey.build_blocks(seg_keys, segments, fulltext) user_prompt = ( f"의상 카탈로그:\n" f"{chr(10).join(catalog_lines)}\n\n" f"{scene_chars_block}" f"{scene_block}\n\n" f"각 씬의 인물에 카탈로그에서 의상을 배정하세요. " f"★카탈로그에 그 인물 소유의 아웃룩이 없으면 배정하지 말고 비워 두세요 — " f"다른 인물의 옷을 빌려 오면 안 됩니다." ) allowed = set(outlook_ids) # ★**소유자까지 본다.** id 존재만 보면 C02 에게 C01 소유 O01 을 준 답도 통과한다 — # 그것이 정확히 이 판에서 막으려던 「남의 옷 빌려오기」다. 프롬프트만 고치고 # 게이트를 그대로 두면 모델이 어기든 말든 아무도 모른다. owner_of = { (ol.get("short_id") or ""): (ol.get("character_id") or "") for ol in (outlooks or []) if ol.get("short_id") } def _owner_violations(rows) -> list: bad = [] for row in rows: for a in row.get("assignments") or []: oid = a.get("outlook_id") or "" cid = a.get("character_id") or "" # ★★O00 은 **옷이 아니라 「덧입은 것 없음」 표시**다 # (2026-09-18). 주인이 없는 게 당연하므로 소유 검사에서 뺀다. # 몸이 곧 신원인 인물(로봇 등)은 맨몸 씬이 정상이라 이 표시를 # 받는다 — 막으면 그 인물이 나오는 씬 전부가 멎는다(실측: # 찰리 C06 에서 단계가 죽었다). if oid == NULL_OUTLOOK_SHORT_ID: continue if oid not in owner_of: continue # 카탈로그 밖 — `allowed` 검사가 본다 owner = owner_of.get(oid) or "" if not owner: # ★★★**주인 없는 옷은 아무에게도 안 준다** (Codex). # 전에는 `if owner and cid` 로 걸러 이 갈래가 통째로 # 빠졌다 — 주인이 없으면 「남의 것인지」를 물을 수가 없어 # **아무에게나** 배정됐다. 그것도 빌려오기다. # ★저장 manifest 58개에 주인 없는 행이 **0건**이라 # 기존 데이터가 여기 기대는 근거도 없다. bad.append(f"씬{row.get('scene_index')}:{cid}→{oid}" f"(소유자 없음)") continue if cid and owner != cid: bad.append(f"씬{row.get('scene_index')}:{cid}→{oid}(소유 {owner})") return bad def _valid(payload: Dict) -> bool: """씬 키 파리티 + 배정마다 카탈로그 id + **소유자 일치** — 하나라도 깨지면 재시도.""" rows = (payload or {}).get("scene_assignments") or [] if not segkey.parity_ok(rows, required_keys, step="outlook_phase2", allowed=seg_keys): return False if allowed: for row in rows: for a in row.get("assignments") or []: if (a.get("outlook_id") or "") not in allowed: return False return not _owner_violations(rows) result = call_structured( step="outlook_extraction", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="outlook_phase2", opik_metadata=opik_metadata, validate_response=_valid, ) # 게이트 — validate_response 는 Tier 3 에 적용되지 않으므로 여기서 한 번 더. # 조용히 통과시키면 아웃룩이 한 칸 밀려 배정되고(씬 키), phase3 가 미배정으로 # 오인해 아웃룩을 지운다(outlook_id). 둘 다 실측된 경로다. segkey.assert_parity( (result or {}).get("scene_assignments") or [], required_keys, step="outlook_phase2", allowed=seg_keys) # ★validate_response 는 Tier 3 에 안 걸리므로 여기서 다시 본다 (기존 관례). _owner_bad = _owner_violations((result or {}).get("scene_assignments") or []) if _owner_bad: from app.core.errors import AppError raise AppError( code="outlook_phase2.owner_mismatch", message=("다른 인물 소유 아웃룩을 배정했다: " + ", ".join(_owner_bad[:5])), status_code=502, ) if allowed: bad = [ f"씬{row.get('scene_index')}:{a.get('character_id')}" f"→{a.get('outlook_id') or a.get('outlook_name') or '(없음)'}" for row in (result or {}).get("scene_assignments") or [] for a in row.get("assignments") or [] if (a.get("outlook_id") or "") not in allowed ] if bad: from app.core.errors import AppError raise AppError( code="outlook_phase2.assignment_id_missing", message=( f"아웃룩 배정 {len(bad)}건이 카탈로그 id 를 갖지 않는다 " f"(outlook_id enum 밖이거나 비었다) — 예: {bad[:5]}" ), status_code=502, ) # 키를 벗기고 세그먼트 인덱스를 달아 준다 — 하류 계약은 그대로. result["scene_assignments"] = segkey.map_back( (result or {}).get("scene_assignments") or [], key_to_index) return result def extract_outlooks_phase3( outlooks: List[Dict], scene_assignments: List[Dict], scene_summaries: Dict[int, str], character_routes: Dict[str, List[int]], # C01 -> [1,3,5,7] project_config: Optional[Dict] = None, opik_metadata: Optional[Dict] = None, ) -> Dict: """3단계: 아웃룩 병합 판별 (LLM) + 코드에서 정리. LLM은 removed(병합 대상)만 반환. 코드에서: 1. 병합 적용 (merge_into로 배정 교체) 2. 배정 0건 아웃룩 제거 3. 인물 유일 아웃룩 보호 """ import copy as _copy system = load_prompt(_MODULE, "phase3") schema = _copy.deepcopy(load_schema(_MODULE, "phase3_schema")) # enum 주입 — removed에서 사용 가능한 ID outlook_ids = [o.get("short_id", "") for o in outlooks if o.get("short_id")] if outlook_ids: schema["properties"]["removed"]["items"]["properties"]["outlook_id"] = { "type": "string", "enum": outlook_ids, } schema["properties"]["removed"]["items"]["properties"]["merge_into"] = { "type": "string", "enum": outlook_ids, } outlook_block = "\n".join( f"- {o.get('short_id','?')} ({o.get('character_id','?')}): {o.get('name', '')} — {o.get('description', '')}" for o in outlooks ) user_prompt = ( f"아웃룩 카탈로그 ({len(outlooks)}개):\n{outlook_block}\n\n" "내용이 매우 동일한 아웃룩이 있으면 병합 대상을 알려주세요." ) llm_result = call_structured( step="outlook_extraction", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="outlook_phase3", opik_metadata=opik_metadata, ) # ── 코드에서 정리 ── removed_list = llm_result.get("removed", []) merge_map = {} # old_id → new_id remove_ids = set() for r in removed_list: old_id = r.get("outlook_id", "") new_id = r.get("merge_into", "") if not (old_id and new_id and old_id != new_id): continue # 인물 유일 아웃룩 보호: 제거 후 남는 아웃룩이 0개면 스킵 cid_for_old = next((o.get("character_id", "") for o in outlooks if o.get("short_id") == old_id), "") surviving = sum(1 for o in outlooks if o.get("character_id") == cid_for_old and o.get("short_id") not in remove_ids and o.get("short_id") != old_id) if surviving == 0: logger.warning("Phase3: skipping merge %s→%s, would leave char %s with 0 outlooks", old_id, new_id, cid_for_old) continue merge_map[old_id] = new_id remove_ids.add(old_id) # 체인 병합 해소: O03→O02→O01 → O03→O01, O02→O01 def _resolve_chain(mm, oid): seen = set() while oid in mm and oid not in seen: seen.add(oid) oid = mm[oid] return oid merge_map = {k: _resolve_chain(merge_map, v) for k, v in merge_map.items()} # 1) 배정에서 merge 적용 for sa in scene_assignments: for a in sa.get("assignments", []): oid = a.get("outlook_id", "") if oid in merge_map: a["outlook_id"] = merge_map[oid] # 2) 배정 0건 아웃룩 검출 # # ★여기는 `outlook_id` 만 읽는다 — 그게 맞다. 배정이 어느 아웃룩을 가리키는지 # **판단은 phase2 의 LLM 이** 하고, 그 답은 outlook_id enum 으로 잠겨 있다. # 코드가 이름을 보고 대신 맞춰 주면 그 판단을 코드가 뺏는 것이고, 잠금이 # 풀려 있어도 드러나지 않는다. 2026-08-07 실측: enum 이 없던 시절 한 실행이 # 308건 중 13건만 id 를 채웠고, 여기서 미배정으로 오인돼 **118건이 참조 중인 # 아웃룩 8개가 삭제**됐다(하류 scene_detail 전량 거부). 잠금은 phase2 스키마에 # 있고, 여기 도달했는데 id 가 비어 있으면 그건 상류 결함이므로 아래에서 # 조용히 넘기지 않고 드러낸다. assigned_oids = set() missing_ref = 0 for sa in scene_assignments: for a in sa.get("assignments", []): oid = a.get("outlook_id", "") if not oid: missing_ref += 1 assigned_oids.add(oid) if missing_ref: logger.error( "Phase3: 배정 %d건에 outlook_id 가 없다 — phase2 의 enum 잠금이 " "풀렸거나 상류 데이터가 낡았다. 이대로면 미배정 오인 삭제가 난다.", missing_ref, ) # 3) 인물별 아웃룩 수 계산 (보호용) char_outlook_count = {} for ol in outlooks: cid = ol.get("character_id", "") oid = ol.get("short_id", "") if oid not in remove_ids: char_outlook_count[cid] = char_outlook_count.get(cid, 0) + 1 # 4) 최종 정리 — 배정 0건 + merge 대상 제거 (인물 유일 아웃룩 보호) cleaned_outlooks = [] final_removed = [] for ol in outlooks: oid = ol.get("short_id", "") cid = ol.get("character_id", "") if oid in remove_ids: final_removed.append({"outlook_id": oid, "merge_into": merge_map[oid]}) continue if oid not in assigned_oids and char_outlook_count.get(cid, 0) > 1: final_removed.append({"outlook_id": oid, "reason": "unassigned"}) char_outlook_count[cid] -= 1 continue cleaned_outlooks.append(ol) logger.info("Phase3 code cleanup: %d→%d outlooks (removed %d, merged %d)", len(outlooks), len(cleaned_outlooks), len(final_removed), len(merge_map)) return { "cleaned_outlooks": cleaned_outlooks, "cleaned_assignments": scene_assignments, "removed": final_removed, } def _build_scene_block(segments: List[Dict], fulltext: str = "") -> str: lines = [] for seg in segments: si = seg.get("scene_index", 0) heading = seg.get("heading", "") text = seg.get("text") or fulltext[seg.get("start_char", 0):seg.get("end_char", 0)] lines.append(f"## 씬 {si}: {heading}\n{text}") return "\n\n".join(lines)