"""T2I 프롬프트 검수 + 수정 테스트 — 감지 후 target→suggestion 치환.""" import json import re import sys import copy 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) # ── 감지 프롬프트 ── DETECT_SYSTEM = """당신은 T2I(text-to-image) 프롬프트 품질 검수관입니다. 프롬프트를 수정하지 마세요. 문제를 감지만 하세요. ## 시각 컨텍스트 {t2i_context} ## 등록된 인물 목록 (이들에 대한 국적/인종은 이미 관리됨 — 검수 대상 아님) {char_names} ## 감지 대상 1. **국적/인종 누락**: 등록 인물이 아닌 보통명사 인물(직원, 경찰, 행인 등)에 국적/인종이 빠진 경우 2. **고유명사 번역**: 지명·상호명 등이 영어로 번역된 경우 (예: "Incheon" → "인천"이어야 함) 3. **원어 어색**: 원어가 자연스러운데 영어로 번역되어 어색한 표현 ## 감지하지 않을 것 - 등록된 인물(위 목록)의 국적/인종 — 별도 관리됨 - 엔티티 ID (C01O02, P03 등) — 시스템 내부 코드 - 동물, 차량, 물체 등 인간이 아닌 대상에는 국적/인종 불필요 - 프롬프트 구조, 문체, 길이 - 문제가 없으면 has_issues=false — 억지로 찾지 마세요 ## target/suggestion 규칙 - target: 원문에서 정확히 찾을 수 있는 문자열 (치환 대상) - suggestion: target을 대체할 문자열 - target은 반드시 원문에 존재하는 그대로의 문자열이어야 합니다 """.format(t2i_context=t2i_context, char_names=char_names_str) DETECT_SCHEMA = { "type": "object", "properties": { "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", "description": "원문에서 정확히 찾을 수 있는 치환 대상 문자열"}, "suggestion": {"type": "string", "description": "target을 대체할 문자열"} }, "required": ["type", "target", "suggestion"], "additionalProperties": False } } }, "required": ["has_issues", "issues"], "additionalProperties": False } # ── 후보 수집 ── def collect_candidates(): candidates = [] # shot_extract d = json.load(open(f"{CPDIR}/shot_extract/manifest.json"))["data"] for s in d.get("scenes", []): si = s.get("scene_index") for sh in s.get("shots", []): desc = sh.get("description", "") for m in re.finditer(r'(남자|여자|직원|경찰|노인|아저씨|아줌마|행인|사람)', desc): ctx = desc[max(0, m.start()-10):m.end()+5] if not any(w in ctx for w in ['한국', 'Korean', '동양', '아시안']): candidates.append({"type": "shot", "scene": si, "shot_index": sh.get("shot_index", 0), "text": desc}) break # entity_t2i d2 = json.load(open(f"{CPDIR}/entity_t2i/manifest.json"))["data"] for cat in ["characters", "props"]: for c in d2.get(cat, []): t2i = c.get("t2i_prompt", "") if not t2i: continue needs = False if not any(w in t2i.lower() for w in ["korean", "east asian", "한국"]): if "사진" in c.get("name", "") or cat == "characters": needs = True if "Incheon" in t2i or "Seoul" in t2i: needs = True if needs: candidates.append({"type": "entity_t2i", "category": cat, "name": c.get("name", ""), "short_id": c.get("short_id", ""), "text": t2i}) # scene_detail — 전수 검사 (Incheon/Seoul + 보통명사 인물) d3 = json.load(open(f"{CPDIR}/scene_detail/manifest.json"))["data"] for s in d3["scenes"]: si = s.get("scene_index") for vi, v in enumerate(s.get("t2i_variations", [])): t2i = v.get("t2i_prompt", "") if not t2i: continue # 고유명사 영어 번역 체크 (대소문자 무시) needs = bool(re.search(r'Incheon|Seoul|Busan|Daegu|Gwangju|Jeju', t2i, re.IGNORECASE)) if not needs: # 보통명사 인물 국적 미표기 체크 for pat in ["police officer", "employee", "staff member", "customer", "worker", "security guard", "detective", "officer"]: idx = t2i.lower().find(pat) if idx > 0: before = t2i[max(0, idx-20):idx].lower() if "korean" not in before: needs = True break if needs: candidates.append({"type": "scene_t2i", "scene": si, "var_index": vi, "text": t2i}) return candidates # ── 감지 실행 ── candidates = collect_candidates() print(f"사전 필터링: {len(candidates)}개 후보\n") detected = [] for i, item in enumerate(candidates): label = f"[{item['type']}] " if "scene" in item: label += f"씬{item['scene']}" if "name" in item: label += item["name"] try: result = call_structured( step="episode_summary", # gpt-mini system_prompt=DETECT_SYSTEM, user_prompt=f"검수 대상:\n{item['text']}", response_schema=DETECT_SCHEMA, ) if result.get("has_issues") and result.get("issues"): detected.append({"item": item, "issues": result["issues"]}) print(f"{i+1}/{len(candidates)} {label} — {len(result['issues'])}건") for iss in result["issues"]: print(f" [{iss['type']}] \"{iss['target']}\" → {iss['suggestion']}") else: print(f"{i+1}/{len(candidates)} {label} — OK") except Exception as e: print(f"{i+1}/{len(candidates)} {label} — ERROR: {e}") print(f"\n감지 완료: {len(detected)}건\n") if not detected: print("문제 없음. 종료.") sys.exit(0) # ── 치환 적용 ── print("=" * 60) print("치환 적용 시작") print("=" * 60) # 체크포인트 로드 shot_cp = json.load(open(f"{CPDIR}/shot_extract/manifest.json")) entity_t2i_cp = json.load(open(f"{CPDIR}/entity_t2i/manifest.json")) scene_detail_cp = json.load(open(f"{CPDIR}/scene_detail/manifest.json")) fix_count = {"shot": 0, "entity_t2i": 0, "scene_t2i": 0} for d in detected: item = d["item"] issues = d["issues"] original = item["text"] fixed = original for iss in issues: target = iss["target"] suggestion = iss["suggestion"] if target in fixed: fixed = fixed.replace(target, suggestion, 1) if fixed == original: continue # 체크포인트에 반영 if item["type"] == "shot": for s in shot_cp["data"]["scenes"]: if s.get("scene_index") == item["scene"]: for sh in s.get("shots", []): if sh.get("shot_index") == item.get("shot_index") and sh.get("description") == original: sh["description"] = fixed fix_count["shot"] += 1 print(f" [shot] 씬{item['scene']} shot{item.get('shot_index')}: 치환 완료") elif item["type"] == "entity_t2i": cat = item.get("category", "props") for c in entity_t2i_cp["data"].get(cat, []): if c.get("t2i_prompt") == original: c["t2i_prompt"] = fixed fix_count["entity_t2i"] += 1 print(f" [entity_t2i] {item['name']}: 치환 완료") # completed에도 반영 for k, v in entity_t2i_cp["data"].get("completed", {}).items(): if isinstance(v, dict) and v.get("t2i_prompt") == original: v["t2i_prompt"] = fixed elif item["type"] == "scene_t2i": for s in scene_detail_cp["data"]["scenes"]: if s.get("scene_index") == item["scene"]: vi = item.get("var_index", 0) vars = s.get("t2i_variations", []) if vi < len(vars) and vars[vi].get("t2i_prompt") == original: vars[vi]["t2i_prompt"] = fixed fix_count["scene_t2i"] += 1 print(f" [scene_t2i] 씬{item['scene']} var{vi}: 치환 완료") print(f"\n{'='*60}") print(f"치환 결과: shot={fix_count['shot']}, entity_t2i={fix_count['entity_t2i']}, scene_t2i={fix_count['scene_t2i']}") print(f"{'='*60}") # 저장 (테스트이므로 별도 파일로) TEST_OUT = "/tmp/t2i_review_fixed" import os os.makedirs(TEST_OUT, exist_ok=True) json.dump(shot_cp, open(f"{TEST_OUT}/shot_extract.json", "w"), ensure_ascii=False, indent=2) json.dump(entity_t2i_cp, open(f"{TEST_OUT}/entity_t2i.json", "w"), ensure_ascii=False, indent=2) json.dump(scene_detail_cp, open(f"{TEST_OUT}/scene_detail.json", "w"), ensure_ascii=False, indent=2) print(f"\n수정된 체크포인트 저장: {TEST_OUT}/") # 검증 — 수정된 데이터에서 재검사 print(f"\n{'='*60}") print("검증: 수정 후 잔여 문제 확인") print("{'='*60}") remaining = 0 for s in shot_cp["data"]["scenes"]: for sh in s.get("shots", []): desc = sh.get("description", "") for m in re.finditer(r'(남자|여자|직원|경찰|노인|아저씨|아줌마|행인|사람)', desc): ctx = desc[max(0, m.start()-10):m.end()+5] if not any(w in ctx for w in ['한국', 'Korean', '동양', '아시안']): remaining += 1 print(f" 잔여: 씬{s['scene_index']} — ...{ctx}...") break incheon_remaining = 0 for s in scene_detail_cp["data"]["scenes"]: for v in s.get("t2i_variations", []): if "Incheon" in v.get("t2i_prompt", ""): incheon_remaining += 1 print(f"\nshot 국적 미표기 잔여: {remaining}건") print(f"scene_detail 'Incheon' 잔여: {incheon_remaining}건")