"""Phase T 실제 데이터 시뮬레이션. 기존 run(1774922951020)의 실제 Kei API 응답 + 실제 MDX로 전 Stage를 시뮬레이션하여 설계 오류를 사전에 잡는다. """ import json import sys from pathlib import Path sys.path.insert(0, ".") # ── 실제 데이터 로드 ── RUN_DIR = Path("data/runs/1774922951020") STAGE_0_DIR = Path("data/runs/20260401_151426") # Stage 0 결과 (실제 실행된 것) stage0_ctx = json.loads((STAGE_0_DIR / "stage_0_context.json").read_text(encoding="utf-8")) raw_content = stage0_ctx["raw_content"] normalized = stage0_ctx["normalized"] # Kei 1A 실제 응답 analysis_1a = json.loads((RUN_DIR / "step1_analysis.json").read_text(encoding="utf-8")) # Kei 1B 실제 응답 concepts_1b = json.loads((RUN_DIR / "step1b_concepts.json").read_text(encoding="utf-8")) def test(): passed = 0 failed = 0 def check(name, condition, detail=""): nonlocal passed, failed if condition: print(f" ✅ {name}") passed += 1 else: print(f" ❌ {name}") if detail: print(f" → {detail}") failed += 1 # ══════════════════════════════════════ # Stage 0: 이미 실행됨 — 결과 확인만 # ══════════════════════════════════════ print("── Stage 0: 실제 결과 확인 ──") check("clean_text", len(normalized["clean_text"]) > 200) check("title", normalized["title"] == "건설산업 DX의 올바른 이해") check("images", len(normalized["images"]) == 1) check("popups", len(normalized["popups"]) == 2, f"got {len(normalized['popups'])}") check("sections", len(normalized["sections"]) >= 3) # ══════════════════════════════════════ # Stage 1A: 실제 Kei 응답 → Topic 모델 변환 # ══════════════════════════════════════ print("\n── Stage 1A: 실제 Kei 응답 → Topic 변환 ──") from src.pipeline_context import Topic, FontHierarchy # 실제 Kei 응답의 topic dict 구조 확인 topics_raw = analysis_1a.get("topics", []) print(f" Kei 반환 topic 수: {len(topics_raw)}") print(f" Kei topic 키: {list(topics_raw[0].keys()) if topics_raw else '없음'}") # 실제 변환 시도 (pipeline.py의 코드와 동일) try: topics = [Topic(**{k: v for k, v in t.items() if k in Topic.model_fields}) for t in topics_raw] check("Topic 변환 성공", True) for t in topics: print(f" topic {t.id}: {t.title} / {t.purpose} / role={t.role}") except Exception as e: check("Topic 변환", False, str(e)) return False # Kei가 안 주는 필드 확인 kei_keys = set(topics_raw[0].keys()) if topics_raw else set() topic_keys = set(Topic.model_fields.keys()) missing_from_kei = topic_keys - kei_keys extra_from_kei = kei_keys - topic_keys print(f" Topic 모델에 있고 Kei에 없는 필드: {missing_from_kei}") print(f" Kei에 있고 Topic 모델에 없는 필드: {extra_from_kei}") check("Kei 미제공 필드가 기본값으로 처리됨", all(hasattr(topics[0], f) for f in missing_from_kei)) # 1A 검증 from src.validators import validate_stage_1a errors_1a = validate_stage_1a(analysis_1a, normalized["clean_text"]) check(f"1A 검증 통과", not errors_1a) for e in errors_1a: print(f" {e['severity']}: {e.get('localization', '')}") # ══════════════════════════════════════ # Stage 1B: 실제 Kei 1B 응답 병합 # ══════════════════════════════════════ print("\n── Stage 1B: 실제 Kei 1B 응답 병합 ──") concepts = concepts_1b.get("concepts", []) print(f" Kei 1B 반환 수: {len(concepts)}") # 병합 (pipeline.py의 코드와 동일) updated_topics = [] for t in topics: match = next((c for c in concepts if c.get("id") == t.id), None) if match: updated = t.model_copy(update={ "relation_type": match.get("relation_type", t.relation_type), "expression_hint": match.get("expression_hint", t.expression_hint), "source_data": match.get("source_data", t.source_data), }) updated_topics.append(updated) else: updated_topics.append(t) check("1B 병합 성공", len(updated_topics) == len(topics)) for t in updated_topics: print(f" topic {t.id}: relation={t.relation_type}, hint={t.expression_hint[:30]}...") # 1B 검증 (raw_content 포함 — popups 대조) from src.validators import validate_stage_1b errors_1b = validate_stage_1b( [t.model_dump() for t in updated_topics], normalized["clean_text"], raw_content=raw_content, ) check(f"1B 검증 통과", not errors_1b) for e in errors_1b: print(f" {e['severity']}: {e.get('localization', '')}") if e.get("evidence"): print(f" 증거: {str(e['evidence'])[:100]}") # ══════════════════════════════════════ # Stage 1.5a: 폰트 위계 + 동적 비율 # ══════════════════════════════════════ print("\n── Stage 1.5a: 폰트 위계 + 동적 비율 ──") from src.pipeline_context import PipelineContext, create_context, NormalizedContent, Analysis, PageStructure from src.space_allocator import calculate_font_hierarchy, calculate_dynamic_ratio # context 구성 (실제 데이터) ctx = create_context(raw_content) ctx = ctx.model_copy(update={ "normalized": NormalizedContent(**normalized), "topics": updated_topics, "page_structure": PageStructure(roles=analysis_1a.get("page_structure", {})), "analysis": Analysis( core_message=analysis_1a.get("core_message", ""), title=analysis_1a.get("title", ""), ), }) # 역할별 텍스트 양 role_text_lengths = {} for role in ["배경", "본심", "첨부", "결론"]: role_text = ctx.get_role_content(role) role_text_lengths[role] = len(role_text) print(f" {role}: {len(role_text)}자") fh_dict = calculate_font_hierarchy(role_text_lengths) try: fh = FontHierarchy( key_msg=fh_dict.get("핵심", 14), core=fh_dict.get("본심", 12), bg=fh_dict.get("배경", 11), sidebar=fh_dict.get("첨부", 10), ) check("폰트 위계 생성", True) print(f" 위계: 핵심={fh.key_msg} > 본심={fh.core} >= 배경={fh.bg} > 첨부={fh.sidebar}") except Exception as e: check("폰트 위계", False, str(e)) return False ratio = calculate_dynamic_ratio(role_text_lengths, fh_dict) check("동적 비율 생성", ratio[0] + ratio[1] == 100) print(f" 비율: body:sidebar = {ratio[0]}:{ratio[1]}") # ══════════════════════════════════════ # Stage 1.7: 참고 블록 선택 # ══════════════════════════════════════ print("\n── Stage 1.7: 참고 블록 선택 ──") from src.block_reference import select_and_generate_references from src.space_allocator import calculate_container_specs from src.design_director import LAYOUT_PRESETS, select_preset preset_name = select_preset(analysis_1a) preset = LAYOUT_PRESETS.get(preset_name, {}) print(f" 프리셋: {preset_name}") container_specs = calculate_container_specs( page_structure=analysis_1a.get("page_structure", {}), topics=[t.model_dump() for t in updated_topics], preset=preset, ) print(f" 컨테이너: {', '.join(f'{r}={s.height_px}px' for r, s in container_specs.items())}") refs = select_and_generate_references( [t.model_dump() for t in updated_topics], container_specs, analysis_1a.get("page_structure", {}), ) check("참고 블록 선택", len(refs) >= 3) for role, ref in refs.items(): html_len = len(ref.get("design_reference_html", "")) has_diff = "차별점" in ref.get("design_reference_html", "") print(f" {role}: {ref['block_id']} ({ref['visual_type']}, html={html_len}자, diff={'✅' if has_diff else '—'})") # ══════════════════════════════════════ # Stage 1.5b: 디자인 예산 # ══════════════════════════════════════ print("\n── Stage 1.5b: 디자인 예산 ──") from src.space_allocator import calculate_design_budget for role, ref in refs.items(): schema = ref.get("schema_info", {}) spec = container_specs.get(role) if not spec: continue font_map = {"본심": fh.core, "배경": fh.bg, "첨부": fh.sidebar, "결론": fh.core} budget = calculate_design_budget(spec.height_px, spec.width_px, schema, font_map.get(role, 12)) check(f"{role} 예산 (fits={budget['fits']})", True) print(f" {role}: container={spec.height_px}px, text={budget['text_height_px']}px, avail={budget['available_height_px']}px") # ══════════════════════════════════════ # Stage 2: 프롬프트 supplement 생성 # ══════════════════════════════════════ print("\n── Stage 2: 프롬프트 supplement ──") from src.html_generator import _build_phase_t_supplement phase_t_ctx = { "font_hierarchy": fh.model_dump(), "container_ratio": ratio, "references": refs, "design_budgets": { role: calculate_design_budget( container_specs[role].height_px, container_specs[role].width_px, refs.get(role, {}).get("schema_info", {}), {"본심": fh.core, "배경": fh.bg, "첨부": fh.sidebar, "결론": fh.core}.get(role, 12) ) for role in container_specs }, } analysis_with_t = {**analysis_1a, "phase_t": phase_t_ctx} for role in ["배경", "본심", "첨부", "결론"]: supp = _build_phase_t_supplement(role, analysis_with_t) has_font = "폰트 위계" in supp has_budget = "디자인 예산" in supp has_ref = "디자인 레퍼런스" in supp check(f"{role} supplement ({len(supp)}자)", len(supp) > 50) if not has_font: print(f" ⚠️ 폰트 위계 누락") if not has_budget: print(f" ⚠️ 디자인 예산 누락") # ══════════════════════════════════════ # 결과 # ══════════════════════════════════════ print(f"\n{'═' * 55}") print(f" 실제 데이터 시뮬레이션: {passed} passed, {failed} failed") if failed == 0: print(" 전체 통과 ✅ — 서버에서 실행해도 이 지점까지 동일하게 동작") else: print(f" ❌ {failed}개 실패 — 서버 실행 전에 수정 필요") print(f"{'═' * 55}") return failed == 0 if __name__ == "__main__": success = test() sys.exit(0 if success else 1)