"""T2I 검수 v2 — entity_t2i + scene_detail 배치 검증 (Gemini Flash) - entity_t2i: 한국어 설명과 함께 T2I 검증 - scene_detail: shot 원본과 함께 T2I 검증 """ import json import re import sys sys.path.insert(0, "/Users/manta/Documents/Projects/TheRoad-I1/backend") from app.modules.llm.llm_client import call_structured CPDIR = "/Users/manta/Documents/Projects/TheRoad-I1/projects/b0a366c5-f5bb-4b46-985b-b66d6bc4bf05/checkpoints/episodes/77714701-5e11-4bc9-b8ea-d3e04a81d90f" # ── 데이터 로드 ── vwr = json.load(open(f"{CPDIR}/visual_world_rules/manifest.json"))["data"] t2i_context = vwr.get("t2i_context", "") entity_merge = json.load(open(f"{CPDIR}/entity_merge/manifest.json"))["data"] char_names = [c["name"] for c in entity_merge.get("characters", [])] char_names_str = ", ".join(char_names) entity_t2i_cp = json.load(open(f"{CPDIR}/entity_t2i/manifest.json"))["data"] scene_detail_cp = json.load(open(f"{CPDIR}/scene_detail/manifest.json"))["data"] shot_extract_cp = json.load(open(f"{CPDIR}/shot_extract/manifest.json"))["data"] # entity_detail에서 한국어 description 로드 entity_detail_cp = json.load(open(f"{CPDIR}/entity_detail/manifest.json"))["data"] entity_desc_map = {} # short_id → 한국어 description for key, val in entity_detail_cp.get("entity_details", {}).items(): parts = key.split(":") name = parts[0] # entity_merge에서 short_id 찾기 for cat in ["characters", "locations", "props"]: for e in entity_merge.get(cat, []): if e.get("name") == name: entity_desc_map[e.get("short_id", "")] = { "name": name, "type": parts[1] if len(parts) > 1 else cat[:-1], "description": val.get("description", ""), } # shot description 맵 (scene_index → shots) shot_map = {} for s in shot_extract_cp.get("scenes", []): si = s.get("scene_index") shots = [] for sh in s.get("shots", []): shots.append({ "shot_index": sh.get("shot_index", 0), "description": sh.get("description", ""), }) shot_map[si] = shots # ══════════════════════════════════════════════════════════ # Part 1: entity_t2i 배치 검증 # ══════════════════════════════════════════════════════════ ENTITY_DETECT_SYSTEM = """당신은 T2I(text-to-image) 프롬프트 품질 검수관입니다. 프롬프트를 수정하지 마세요. 문제를 감지만 하세요. ## 시각 컨텍스트 {t2i_context} ## 검수 방법 각 엔티티에 대해 [한국어 설명]과 [T2I 프롬프트]를 비교합니다. 한국어 설명은 시나리오 원문 기반 정보이고, T2I는 이미지 생성용 영어 프롬프트입니다. ## 감지 대상 1. **국적/인종 누락**: 인간형 인물이나 사진 속 인물에 국적/인종이 빠진 경우 2. **고유명사 번역**: 지명·상호명 등이 영어로 번역된 경우 (예: "Incheon" → "인천"이어야 함) 3. **원어 어색**: 원어가 자연스러운데 영어로 번역되어 어색한 표현 ## 감지하지 않을 것 - 동물, 차량, 물체 등 인간이 아닌 대상의 국적/인종 - 엔티티 ID (C01, P03 등) - 프롬프트 구조, 문체, 길이 - 문제가 없으면 해당 엔티티의 issues를 빈 배열로 ## target/suggestion 규칙 - target: T2I 프롬프트 원문에서 정확히 찾을 수 있는 문자열 - suggestion: target을 대체할 문자열 """.format(t2i_context=t2i_context) # entity 배치 구성 entity_items = [] for cat in ["characters", "props"]: for c in entity_t2i_cp.get(cat, []): sid = c.get("short_id", "") t2i = c.get("t2i_prompt", "") if not t2i: continue desc_info = entity_desc_map.get(sid, {}) entity_items.append({ "short_id": sid, "name": c.get("name", ""), "category": cat, "korean_desc": desc_info.get("description", ""), "t2i_prompt": t2i, }) # 배치 user prompt 구성 entity_user_lines = [] for ei in entity_items: entity_user_lines.append( f"[{ei['short_id']}] {ei['name']} ({ei['category']})\n" f" 한국어 설명: {ei['korean_desc']}\n" f" T2I: {ei['t2i_prompt']}" ) entity_user_prompt = "아래 엔티티들의 T2I 프롬프트를 검수하세요:\n\n" + "\n\n".join(entity_user_lines) ENTITY_DETECT_SCHEMA = { "type": "object", "properties": { "results": { "type": "array", "items": { "type": "object", "properties": { "short_id": {"type": "string"}, "has_issues": {"type": "boolean"}, "issues": { "type": "array", "items": { "type": "object", "properties": { "type": {"type": "string", "enum": ["missing_ethnicity", "proper_noun_translated", "awkward_translation"]}, "target": {"type": "string"}, "suggestion": {"type": "string"} }, "required": ["type", "target", "suggestion"], "additionalProperties": False } } }, "required": ["short_id", "has_issues", "issues"], "additionalProperties": False } } }, "required": ["results"], "additionalProperties": False } print(f"═══ Part 1: entity_t2i 배치 검증 ({len(entity_items)}개) ═══\n") entity_result = call_structured( step="scene_segmentation", # gemini-flash system_prompt=ENTITY_DETECT_SYSTEM, user_prompt=entity_user_prompt, response_schema=ENTITY_DETECT_SCHEMA, ) entity_fixes = [] for r in entity_result.get("results", []): sid = r.get("short_id", "") if r.get("has_issues") and r.get("issues"): print(f" {sid}: {len(r['issues'])}건 감지") for iss in r["issues"]: print(f" [{iss['type']}] \"{iss['target']}\" → {iss['suggestion']}") entity_fixes.append({"short_id": sid, **iss}) else: print(f" {sid}: OK") print(f"\n entity 감지 합계: {len(entity_fixes)}건\n") # ══════════════════════════════════════════════════════════ # Part 2: scene_detail T2I 배치 검증 (10개씩 배치) # ══════════════════════════════════════════════════════════ SCENE_DETECT_SYSTEM = """당신은 T2I(text-to-image) 프롬프트 품질 검수관입니다. 프롬프트를 수정하지 마세요. 문제를 감지만 하세요. ## 시각 컨텍스트 {t2i_context} ## 등록된 인물 목록 (이들에 대한 국적/인종은 이미 관리됨 — 검수 대상 아님) {char_names} ## 검수 방법 각 씬에 대해 [shot 원본 설명]과 [T2I 프롬프트]를 비교합니다. shot 원본은 한국어 장면 묘사이고, T2I는 이미지 생성용 영어 프롬프트입니다. ## 감지 대상 1. **국적/인종 누락**: 등록 인물이 아닌 보통명사 인물(직원, 경찰, 행인 등)에 국적/인종이 빠진 경우 2. **고유명사 번역**: 지명·상호명 등이 영어로 번역된 경우 (예: "Incheon" → "인천") 3. **원어 어색**: 원어가 자연스러운데 영어로 번역되어 어색한 표현 ## 감지하지 않을 것 - 등록된 인물(위 목록)의 국적/인종 - 엔티티 ID (C01O02, P03 등) - 동물, 차량, 물체의 국적/인종 - 프롬프트 구조, 문체, 길이 - 문제 없으면 해당 씬의 issues를 빈 배열로 ## target/suggestion 규칙 - target: T2I 프롬프트 원문에서 정확히 찾을 수 있는 문자열 - suggestion: target을 대체할 문자열 """.format(t2i_context=t2i_context, char_names=char_names_str) SCENE_DETECT_SCHEMA = { "type": "object", "properties": { "results": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "var_index": {"type": "integer"}, "has_issues": {"type": "boolean"}, "issues": { "type": "array", "items": { "type": "object", "properties": { "type": {"type": "string", "enum": ["missing_ethnicity", "proper_noun_translated", "awkward_translation"]}, "target": {"type": "string"}, "suggestion": {"type": "string"} }, "required": ["type", "target", "suggestion"], "additionalProperties": False } } }, "required": ["scene_index", "var_index", "has_issues", "issues"], "additionalProperties": False } } }, "required": ["results"], "additionalProperties": False } # scene_detail 배치 구성 scene_items = [] for s in scene_detail_cp["scenes"]: si = s.get("scene_index") shots = shot_map.get(si, []) shot_text = "\n".join(f" shot{sh['shot_index']}: {sh['description']}" for sh in shots) if shots else "(없음)" for vi, v in enumerate(s.get("t2i_variations", [])): t2i = v.get("t2i_prompt", "") if not t2i: continue scene_items.append({ "scene_index": si, "var_index": vi, "shot_text": shot_text, "t2i_prompt": t2i, }) print(f"═══ Part 2: scene_detail 배치 검증 ({len(scene_items)}개, 10개씩) ═══\n") BATCH_SIZE = 10 scene_fixes = [] for batch_start in range(0, len(scene_items), BATCH_SIZE): batch = scene_items[batch_start:batch_start + BATCH_SIZE] batch_label = f"배치 {batch_start//BATCH_SIZE + 1}/{(len(scene_items)-1)//BATCH_SIZE + 1}" user_lines = [] for si in batch: user_lines.append( f"[씬{si['scene_index']} var{si['var_index']}]\n" f" shot 원본:\n{si['shot_text']}\n" f" T2I: {si['t2i_prompt']}" ) user_prompt = "아래 씬들의 T2I 프롬프트를 검수하세요:\n\n" + "\n\n".join(user_lines) try: result = call_structured( step="scene_segmentation", # gemini-flash system_prompt=SCENE_DETECT_SYSTEM, user_prompt=user_prompt, response_schema=SCENE_DETECT_SCHEMA, ) issues_in_batch = 0 for r in result.get("results", []): si = r.get("scene_index") vi = r.get("var_index", 0) if r.get("has_issues") and r.get("issues"): issues_in_batch += len(r["issues"]) for iss in r["issues"]: print(f" 씬{si} var{vi}: [{iss['type']}] \"{iss['target']}\" → {iss['suggestion']}") scene_fixes.append({"scene_index": si, "var_index": vi, **iss}) print(f" {batch_label}: {len(batch)}개 검수, {issues_in_batch}건 감지") except Exception as e: print(f" {batch_label}: ERROR — {e}") print(f"\n scene 감지 합계: {len(scene_fixes)}건\n") # ══════════════════════════════════════════════════════════ # 치환 적용 # ══════════════════════════════════════════════════════════ print("=" * 60) print("치환 적용") print("=" * 60) # entity_t2i 치환 entity_t2i_full = json.load(open(f"{CPDIR}/entity_t2i/manifest.json")) e_fix_count = 0 for fix in entity_fixes: sid = fix["short_id"] target = fix["target"] suggestion = fix["suggestion"] for cat in ["characters", "props"]: for c in entity_t2i_full["data"].get(cat, []): if c.get("short_id") == sid: old = c.get("t2i_prompt", "") if target in old: c["t2i_prompt"] = old.replace(target, suggestion, 1) e_fix_count += 1 print(f" entity {sid}: \"{target}\" → \"{suggestion}\"") # completed도 for k, v in entity_t2i_full["data"].get("completed", {}).items(): if isinstance(v, dict) and v.get("short_id") == sid: old = v.get("t2i_prompt", "") if target in old: v["t2i_prompt"] = old.replace(target, suggestion, 1) # scene_detail 치환 scene_detail_full = json.load(open(f"{CPDIR}/scene_detail/manifest.json")) s_fix_count = 0 for fix in scene_fixes: si = fix["scene_index"] vi = fix["var_index"] target = fix["target"] suggestion = fix["suggestion"] for s in scene_detail_full["data"]["scenes"]: if s.get("scene_index") == si: vars = s.get("t2i_variations", []) if vi < len(vars): old = vars[vi].get("t2i_prompt", "") if target in old: vars[vi]["t2i_prompt"] = old.replace(target, suggestion, 1) s_fix_count += 1 print(f" 씬{si} var{vi}: \"{target}\" → \"{suggestion}\"") print(f"\n치환 결과: entity={e_fix_count}, scene={s_fix_count}") # 저장 import os TEST_OUT = "/tmp/t2i_review_v2" os.makedirs(TEST_OUT, exist_ok=True) json.dump(entity_t2i_full, open(f"{TEST_OUT}/entity_t2i.json", "w"), ensure_ascii=False, indent=2) json.dump(scene_detail_full, open(f"{TEST_OUT}/scene_detail.json", "w"), ensure_ascii=False, indent=2) print(f"\n저장: {TEST_OUT}/") # 검증 print(f"\n{'='*60}") print("검증") print(f"{'='*60}") incheon_remaining = 0 for s in scene_detail_full["data"]["scenes"]: for v in s.get("t2i_variations", []): if "Incheon" in v.get("t2i_prompt", ""): incheon_remaining += 1 print(f"scene_detail 'Incheon' 잔여: {incheon_remaining}건")