"""Phase Z-2 MVP-1.5b — single slide + Type B + frame-derived adapted blocks. 원래 Phase Z 설계 복귀 (멀티-슬라이드 / native-fit 모두 폐기) : - MDX 1 = slide 1 - slide-base → slide-body → layout preset (Type B) → zones[] → frame-derived block (zone-compatible adapt) - frame은 시각 언어 / slot 구성 / 패턴의 source. native geometry 통째 삽입 X. - AI 는 layout / zone / frame / variant 선택에 관여 X — code / catalog 가 결정. MVP-1.5b spec : - 대상 : MDX 03 (회귀) - 출력 : data/runs/{run_id}/phase_z2/final.html (single slide) - AI : 미사용 — MDX → slot_payload 결정론적 매핑 - status : matched_zone only — non-matched 발생 시 abort + error.json - layout : 2 sections → Type B (top + bottom zones) - Frame partials : templates/phase_z2/families/{template_id}.html (Figma 시각 언어 promote, geometry adapt) - Assets : render time copy → data/runs/{run_id}/phase_z2/assets/{template_id}/ 상세 설계 : - docs/architecture/PHASE-Z-CATALOG-RUNTIME-DESIGN.md § 17 (frame-derived partial promotion + zone-compatible adapt) 이전 실험 실패 기록 : - mvp1_test5 : scaffold 임의 — frame 느낌 부재 - mvp1.5_test3 : frame native 통째 — slide 대체 - mvp1.5a_test1 : 멀티-슬라이드 — MDX 1=slide 1 위반 - mvp1.5b_test* : 본 모듈, 원래 설계 라인 합류 """ import json import re import shutil import sys import time from dataclasses import dataclass from pathlib import Path from typing import Optional import yaml from jinja2 import Environment, FileSystemLoader, select_autoescape from phase_z2_composition import ( LAYOUT_PRESETS, CompositionUnit, plan_composition, ) from phase_z2_mapper import ( FitError, compute_capacity_fit, get_contract, map_with_contract, ) from phase_z2_classifier import classify_visual_runtime_check from phase_z2_router import route_fit_classification from phase_z2_retry import ( DEFAULT_SAFETY_MARGIN_PX, apply_retry_to_layout_css, plan_zone_ratio_retry, ) from phase_z2_failure_router import enrich_retry_trace_with_failure_classification # ─── Constants ────────────────────────────────────────────────── PROJECT_ROOT = Path(__file__).parent.parent TEMPLATE_DIR = PROJECT_ROOT / "templates" / "phase_z2" ASSETS_SOURCE_BASE = PROJECT_ROOT / "figma_to_html_agent" / "blocks" V4_RESULT_PATH = PROJECT_ROOT / "tests" / "matching" / "v4_full32_result.yaml" RUNS_DIR = PROJECT_ROOT / "data" / "runs" # V4 label → Phase Z status (§ 7.4 매트릭스) V4_LABEL_TO_PHASE_Z_STATUS = { "use_as_is": "matched_zone", "light_edit": "adapt_matched_zone", "restructure": "extract_matched_zone", "reject": "fallback_candidate", } MVP1_ALLOWED_STATUSES = {"matched_zone", "adapt_matched_zone"} # adapt_matched_zone (V4 light_edit) = frame 구조 동일, 텍스트만 minor edit 필요. # minor edit 정책 (mapper 의무) : # 1. MDX item 수 < frame slot 수 → 빈 slot 그대로 (Jinja2 {% if %} 로 스킵) # 2. MDX item 수 > frame slot 수 → 추가 item 누락 (truncate) — debug.json 에 기록 # 3. 텍스트 길이 mismatch → 그대로 통과 (overflow 는 zone-fit + Selenium check 가 처리) # 4. slot ↔ MDX item 의미 매핑 → 순서 기반 (간단). V4 anchor_match 정교화는 future # AI 호출 X — MVP-1.5b 의 "MDX 1:1 결정론적 매핑" 룰 그대로. # Slide-body geometry (legacy contract) SLIDE_BODY_HEIGHT = 590 GRID_GAP = 12 # zone min-height fallback — contract 에 visual_hints.min_height_px 없을 때 사용. # token-based font (var(--font-body) 11px 등) 기준 최소 가독 높이. DEFAULT_ZONE_MIN_HEIGHT_PX = 100 # content_weight 계산 가중치 CONTENT_WEIGHT_COEFFS = { "text_per_chars": 800, # text_len / 800 = score "top_bullet": 0.4, "nested_bullet": 0.15, "table_bonus": 1.5, "subsection": 0.6, } # ─── Data classes ─────────────────────────────────────────────── @dataclass class MdxSection: section_id: str section_num: int title: str raw_content: str @dataclass class V4Match: section_id: str frame_id: str frame_number: int template_id: str confidence: float label: str def to_phase_z_status(match: V4Match) -> str: return V4_LABEL_TO_PHASE_Z_STATUS.get(match.label, "unknown") # ─── MDX parsing ──────────────────────────────────────────────── def parse_mdx(mdx_path: Path) -> tuple[str, list[MdxSection], Optional[str]]: """basic MDX parser — ## level sections only. V4 무관 (matching artifact 모름). section.raw_content 에 ### sub-section 그대로 포함. V4 granularity 와 align 은 align_sections_to_v4_granularity() 가 처리. """ text = mdx_path.read_text(encoding="utf-8") fm_match = re.match(r"^---\n(.*?)\n---\n", text, re.DOTALL) slide_title = "" if fm_match: fm = yaml.safe_load(fm_match.group(1)) slide_title = fm.get("title", "") text = text[fm_match.end():] footer_match = re.search(r":::note\[[^\]]*\]\n(.*?)\n:::", text, re.DOTALL) footer_text = None if footer_match: body = footer_match.group(1) bullet_match = re.search(r"\*\s*\*\*([^*]+)\*\*", body) footer_text = (bullet_match.group(1).strip() if bullet_match else body.strip()) text = text[:footer_match.start()] + text[footer_match.end():] sections = [] section_pattern = re.compile(r"^##\s+(\d+)\.\s+(.+?)$", re.MULTILINE) matches = list(section_pattern.finditer(text)) mdx_num_match = re.match(r"(\d+)", mdx_path.stem) mdx_id = mdx_num_match.group(1).zfill(2) if mdx_num_match else "00" for i, m in enumerate(matches): section_num = int(m.group(1)) title_text = m.group(2).strip() start = m.end() end = matches[i + 1].start() if i + 1 < len(matches) else len(text) raw_content = text[start:end].strip() sections.append(MdxSection( section_id=f"{mdx_id}-{section_num}", section_num=section_num, title=f"{section_num}. {title_text}", raw_content=raw_content, )) return slide_title, sections, footer_text # ─── V4 lookup ────────────────────────────────────────────────── def load_v4_result() -> dict: return yaml.safe_load(V4_RESULT_PATH.read_text(encoding="utf-8")) def align_sections_to_v4_granularity(sections: list[MdxSection], v4: dict) -> list[MdxSection]: """V4 section granularity 에 맞춰 sections 조정. 각 section 에 대해 : - V4 에 section.section_id 키 있음 → 그대로 유지 (## level 매칭) - V4 에 키 없고 raw_content 에 ### sub-section 존재 → ### 로 drill - V4 에 키 없고 ### 도 없음 → 원본 그대로 (V4 lookup 단계에서 자연스럽게 abort) 설계 원칙 : - parser (parse_mdx) = MDX 만 앎 (V4 무관) - aligner (이 함수) = V4 키 기준 granularity 결정 - runtime parser 가 matching artifact 의 granularity 를 *따라가는* 구조 """ v4_keys = set(v4.get("mdx_sections", {}).keys()) aligned: list[MdxSection] = [] for section in sections: if section.section_id in v4_keys: aligned.append(section) continue # ### drill 시도 sub_pattern = re.compile(r"^###\s+(\d+\.\d+)\s+(.+?)$", re.MULTILINE) sub_matches = list(sub_pattern.finditer(section.raw_content)) if not sub_matches: aligned.append(section) # drill 불가, V4 lookup 에서 abort 됨 continue # ### sub-section 추출 mdx_id = section.section_id.split("-")[0] # e.g., "04" for i, m in enumerate(sub_matches): subnum = m.group(1) # e.g., "2.1" sub_title = m.group(2).strip() start = m.end() end = sub_matches[i + 1].start() if i + 1 < len(sub_matches) else len(section.raw_content) raw = section.raw_content[start:end].strip() aligned.append(MdxSection( section_id=f"{mdx_id}-{subnum}", # e.g., "04-2.1" section_num=section.section_num, title=f"{subnum} {sub_title}", raw_content=raw, )) return aligned def lookup_v4_match(v4: dict, section_id: str) -> Optional[V4Match]: sec = v4.get("mdx_sections", {}).get(section_id) if not sec: return None judgments = sec.get("judgments_full32", []) if not judgments: return None top = judgments[0] return V4Match( section_id=section_id, frame_id=str(top["frame_id"]), frame_number=int(top["frame_number"]), template_id=top["template_id"], confidence=float(top["confidence"]), label=top["label"], ) # ─── Content weight + zone layout 계산 ───────────────────────── # layout preset 선택은 phase_z2_composition.select_layout_preset (composition v0) 가 담당. # 본 모듈의 select_layout_preset 은 이전 단순 count-based 구현이었고 dead code 로 제거 (2026-04-29). def compute_content_weight(section: MdxSection) -> dict: """Section 의 콘텐츠 부피 측정 — text/bullet/table/subsection 합성 score.""" text = section.raw_content lines = text.splitlines() text_len = len(text) top_bullets = sum(1 for l in lines if re.match(r"^[\*\-]\s", l)) nested_bullets = sum(1 for l in lines if re.match(r"^\s+[\*\-]\s", l)) has_table = bool(re.search(r"\|[^\n]+\|\n[ \t]*\|[\s\-:|]+\|", text)) subsections = len(re.findall(r"^###\s", text, re.MULTILINE)) c = CONTENT_WEIGHT_COEFFS score = ( text_len / c["text_per_chars"] + top_bullets * c["top_bullet"] + nested_bullets * c["nested_bullet"] + (c["table_bonus"] if has_table else 0) + subsections * c["subsection"] ) return { "score": round(score, 3), "text_length": text_len, "top_bullets": top_bullets, "nested_bullets": nested_bullets, "has_table": has_table, "subsection_count": subsections, } def compute_zone_layout(zones_data: list[dict], total_height: int = SLIDE_BODY_HEIGHT, gap: int = GRID_GAP) -> dict: """zone height 계산 — frame_min_height_px 우선 + 남은 공간 content_weight 비율 분배. Returns dict with per-zone heights + reasoning trace. """ n = len(zones_data) if n == 0: return {"heights_px": [], "ratios": [], "zones": []} available = total_height - gap * (n - 1) # Step 1: 각 zone 의 min_height 할당 — pipeline 가 zones_data 에 frame contract 의 # visual_hints.min_height_px 를 미리 주입했음. 없으면 DEFAULT_ZONE_MIN_HEIGHT_PX. min_heights = [ z.get("min_height_px", DEFAULT_ZONE_MIN_HEIGHT_PX) for z in zones_data ] total_min = sum(min_heights) min_scaled = False if total_min > available: scale = available / total_min min_heights = [int(m * scale) for m in min_heights] total_min = sum(min_heights) min_scaled = True remaining = available - total_min # Step 2: 남은 공간을 content_weight 비율로 분배 weights = [z["content_weight"]["score"] for z in zones_data] total_w = sum(weights) if sum(weights) > 0 else n extras = [int(round(remaining * (w / total_w))) for w in weights] # Step 3: rounding 보정 (마지막 zone 잔여 흡수) heights = [m + e for m, e in zip(min_heights, extras)] diff = available - sum(heights) if diff != 0 and heights: heights[-1] += diff ratios = [round(h / total_height, 3) for h in heights] return { "computation": "min_height_first + content_weight_distribution", "slide_body_height": total_height, "gap": gap, "available_after_gap": available, "min_heights_px": min_heights, "min_scaled": min_scaled, "total_min_height": total_min, "remaining_after_min": remaining, "content_weights": [{"position": z["position"], "template_id": z["template_id"], "score": w} for z, w in zip(zones_data, weights)], "weight_shares": [round(w / total_w, 3) for w in weights], "extras_px": extras, "heights_px": heights, "ratios": ratios, } # Layout preset → zone position 순서 = LAYOUT_PRESETS[preset]["positions"] 직접 사용. # 이전 ZONE_POSITIONS_BY_PRESET (type-b 등 legacy 명) 는 dead code 로 제거 (2026-04-29). def build_layout_css(layout_preset: str, zones_data: list[dict], gap: int = GRID_GAP) -> dict: """Composition v0 layout preset → CSS grid 문자열. horizontal-2 (= old type-b, 2-zone vertical stack) 만 dynamic heights 유지 (MDX 03 회귀 보존 — content_weight 기반). 다른 preset 은 fr default. 향후 cardinality_fit / density_score axis 가 score_candidate 에 들어가면 cols/rows 도 dynamic 으로 확장 가능. """ preset = LAYOUT_PRESETS[layout_preset] if layout_preset == "horizontal-2": zl = compute_zone_layout(zones_data, gap=gap) rows = " ".join(f"{h}px" for h in zl["heights_px"]) return { "areas": preset["css_areas"], "cols": preset["css_cols"], "rows": rows, "heights_px": zl["heights_px"], "ratios": zl["ratios"], "computation": zl["computation"], "dynamic_rows": True, "raw_zone_layout": zl, } return { "areas": preset["css_areas"], "cols": preset["css_cols"], "rows": preset["css_rows"], "heights_px": [], "ratios": [], "computation": "fr_default_from_preset", "dynamic_rows": False, "raw_zone_layout": None, } # ─── Abort ────────────────────────────────────────────────────── def abort_with_error(run_dir: Path, section: MdxSection, match: Optional[V4Match], stage: str, reason: str): error_data = { "section": {"id": section.section_id, "title": section.title}, "frame": { "id": match.frame_id if match else None, "number": match.frame_number if match else None, "template_id": match.template_id if match else None, }, "v4_label": match.label if match else None, "phase_z_status": to_phase_z_status(match) if match else None, "confidence": match.confidence if match else None, "stage": stage, "reason": reason, } run_dir.mkdir(parents=True, exist_ok=True) err_path = run_dir / "error.json" err_path.write_text(json.dumps(error_data, ensure_ascii=False, indent=2), encoding="utf-8") print(f"\n[Phase Z-2 MVP-1.5b] ABORT @ {stage}", file=sys.stderr) print(f" section : {section.section_id} — {section.title}", file=sys.stderr) if match: print(f" frame : {match.frame_id} ({match.template_id})", file=sys.stderr) print(f" status : V4 label '{match.label}' → Phase Z '{to_phase_z_status(match)}'", file=sys.stderr) print(f" reason : {reason}", file=sys.stderr) print(f" error : {err_path}", file=sys.stderr) sys.exit(1) # ─── Slot mapping (catalog-only dispatch) ────────────────────── def _known_contract_ids() -> list[str]: from phase_z2_mapper import load_frame_contracts return list(load_frame_contracts().keys()) def map_mdx_to_slots(section: MdxSection, template_id: str) -> dict: """template_id → slot_payload via catalog contract only. F13/F29/F16 등 모든 frame 의 slot 구조 / cardinality / role / payload builder 는 `templates/phase_z2/catalog/frame_contracts.yaml` 에 선언. legacy hand-coded mapper / MAPPER_BY_TEMPLATE / COLOR_CLASS_BY_KEYWORD / 관련 helper 는 F13/F29/F16 transition (2026-04-29) 후 모두 제거. template_id 가 catalog 에 없으면 ValueError — fallback 없음. 새 frame 추가 = catalog yaml 에 entry 추가 + (필요시) 새 builder/parser 등록. """ contract = get_contract(template_id) if contract is None: raise ValueError( f"No frame_contracts entry for template_id='{template_id}'. " f"Add an entry in templates/phase_z2/catalog/frame_contracts.yaml. " f"Known contracts: {sorted(_known_contract_ids())}." ) return map_with_contract(section, contract) # ─── Asset copy ───────────────────────────────────────────────── def copy_assets(template_id: str, run_dir: Path) -> Optional[Path]: """Frame asset (Figma) 폴더 복사 — frame_id 는 catalog contract 에서 도출. contract 에 `frame_id` 없으면 (asset 없는 frame) None 반환. 이전엔 pipeline.py 에 TEMPLATE_TO_FRAME_ID Python dict 가 있었지만 catalog 로 이전 (2026-04-29). """ contract = get_contract(template_id) frame_id = (contract or {}).get("frame_id") if not frame_id: return None src = ASSETS_SOURCE_BASE / str(frame_id) / "assets" if not src.exists(): return None dst = run_dir / "assets" / template_id dst.parent.mkdir(parents=True, exist_ok=True) if dst.exists(): shutil.rmtree(dst) shutil.copytree(src, dst) return dst # ─── Render (single slide + Type B) ──────────────────────────── def _read_token_css() -> str: token_dir = PROJECT_ROOT / "templates" / "styles" / "tokens" files = ["typography.css", "spacing.css", "colors.css"] parts = [] for f in files: path = token_dir / f if path.exists(): parts.append(f"/* === {f} === */\n{path.read_text(encoding='utf-8')}") return "\n\n".join(parts) def _attempt_zone_ratio_retry( *, run_dir: Path, out_path: Path, slide_title: str, slide_footer: Optional[str], zones_data: list[dict], debug_zones: list[dict], layout_preset: str, layout_css: dict, overflow: dict, fit_classification: dict, router_decision: dict, gap_px: int, ) -> dict: """A3 zone_ratio_retry orchestration. locked rules : - retry budget = 1 - slide-base / spacing / gap 고정 - target zone height 만 증가, sibling donor 에서 같은 양 차감 - donor 룰 strict (visual ok / capacity ok / slack > 0 / min_height 보존) - (b) revert : redistribution fail 또는 rerender 후 visual fail 시 original final.html 그대로 Returns: retry_trace dict (always returned, even when no retry attempted) with : retry_attempted : bool retry_action : 'zone_ratio_retry' or None plan : phase_z2_retry.plan_zone_ratio_retry() 결과 (있을 때만) rerender_attempted : bool retry_passed : bool retry_failure_reason : str or None retried_candidate_path : str or None (rerender 한 경우 진단 artifact 경로) post_retry_overflow : dict (retry_passed=True 일 때만) post_retry_debug_zones : list (retry_passed=True 일 때만 — height_px 갱신본) post_retry_layout_css : dict (retry_passed=True 일 때만) """ base_trace = { "retry_attempted": False, "retry_action": None, "plan": None, "rerender_attempted": False, "retry_passed": False, "retry_failure_reason": None, "retried_candidate_path": None, "safety_margin_px": DEFAULT_SAFETY_MARGIN_PX, "policy": ( "A3 locked rules : retry budget=1, slide-base/spacing/gap fixed, " "donor strict (sibling/visual ok/capacity ok/slack>0/min_height 보존), " "(b) revert on redistribution fail or rerender visual fail." ), } # 1. retry attempt 자체가 적절한지 판단 if not router_decision.get("router_active"): base_trace["retry_skipped_reason"] = "router_active=False (visual check passed — no overflow)" return base_trace proposed = router_decision.get("proposed_actions_summary") or [] if "zone_ratio_retry" not in proposed: base_trace["retry_skipped_reason"] = ( f"zone_ratio_retry not in proposed_actions {proposed} (다른 action category)" ) return base_trace # 2. plan base_trace["retry_attempted"] = True base_trace["retry_action"] = "zone_ratio_retry" plan = plan_zone_ratio_retry( debug_zones=debug_zones, overflow=overflow, fit_classification=fit_classification, router_decision=router_decision, safety_margin_px=DEFAULT_SAFETY_MARGIN_PX, ) base_trace["plan"] = plan if plan is None: base_trace["retry_failure_reason"] = "plan_zone_ratio_retry returned None — no target classification matched zone_ratio_retry" return base_trace if not plan.get("feasible"): # redistribution check 실패 → rerender 안 함, original final.html 그대로 base_trace["retry_failure_reason"] = plan.get("failure_reason") print( f" retry : zone_ratio_retry redistribution INFEASIBLE — " f"target {plan['target_zone_position']} needs {plan['target_added_px']}px, " f"{plan.get('failure_reason')}" ) return base_trace # 3. feasible — apply plan to layout_css, rerender to candidate path (NOT final.html yet) new_layout_css = apply_retry_to_layout_css( layout_css, plan, zones_data, total_height=SLIDE_BODY_HEIGHT, gap_px=gap_px, ) candidate_path = run_dir / "retried_candidate.html" candidate_html = render_slide( slide_title, slide_footer, zones_data, layout_preset, new_layout_css, gap_px=gap_px, ) candidate_path.write_text(candidate_html, encoding="utf-8") base_trace["rerender_attempted"] = True base_trace["retried_candidate_path"] = str(candidate_path.relative_to(PROJECT_ROOT)) print( f" retry : zone_ratio_retry attempted — target {plan['target_zone_position']} " f"+{plan['target_added_px']}px (donor {plan['donor_zone_position']} -{plan['donor_reduced_px']}px) " f"→ rerender to retried_candidate.html → visual check" ) # 4. 후 visual check on candidate candidate_overflow = run_overflow_check(candidate_path) if candidate_overflow.get("passed", False): # 성공 — final.html 을 candidate 로 promote out_path.write_text(candidate_html, encoding="utf-8") # debug_zones height_px / ratio 갱신 (post-retry 상태) new_heights = new_layout_css["heights_px"] new_ratios = new_layout_css["ratios"] post_retry_debug_zones = [] for i, dz in enumerate(debug_zones): new_dz = dict(dz) new_dz["height_px"] = new_heights[i] if i < len(new_heights) else dz.get("height_px") new_dz["ratio"] = new_ratios[i] if i < len(new_ratios) else dz.get("ratio") new_dz["zone_height_post_retry"] = True post_retry_debug_zones.append(new_dz) base_trace["retry_passed"] = True base_trace["post_retry_overflow"] = candidate_overflow base_trace["post_retry_debug_zones"] = post_retry_debug_zones base_trace["post_retry_layout_css"] = new_layout_css print(f" retry : PASSED — final.html promoted to retried version") return base_trace # 5. rerender 후에도 visual fail → (b) revert : final.html 은 original 그대로 (이미 written) base_trace["retry_passed"] = False base_trace["retry_failure_reason"] = ( f"rerender visual_check failed: {candidate_overflow.get('fail_reasons')}. " f"reverting to original final.html (retried_candidate.html stays as diagnostic only)." ) base_trace["candidate_overflow_summary"] = { "passed": False, "fail_reasons": candidate_overflow.get("fail_reasons", []), } print(f" retry : FAILED — candidate visual_check 도 실패. revert to original. ({candidate_path.name} 은 diagnostic 으로 보존)") return base_trace def render_slide(slide_title: str, slide_footer: Optional[str], zones_data: list[dict], layout_preset: str, layout_css: dict, gap_px: int = GRID_GAP) -> str: """Single slide HTML — slide_base.html + 8-preset layout vocabulary. layout_css = build_layout_css() 결과 — areas/cols/rows 문자열 + 동적 heights flag. Template 은 layout_css.{areas,cols,rows} 를 grid CSS 에 직접 주입. """ env = Environment( loader=FileSystemLoader(str(TEMPLATE_DIR)), autoescape=select_autoescape(["html"]), ) for zone in zones_data: partial = env.get_template(f"families/{zone['template_id']}.html") zone["partial_html"] = partial.render(slot_payload=zone["slot_payload"]) base = env.get_template("slide_base.html") return base.render( slide_title=slide_title, slide_footer=slide_footer, zones=zones_data, layout_preset=layout_preset, layout_css=layout_css, gap_px=gap_px, token_css=_read_token_css(), ) # ─── Selenium check (single slide + per-zone) ────────────────── def run_overflow_check(html_path: Path) -> dict: """Single slide + per-zone overflow + clipping check.""" from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service options = Options() options.add_argument("--headless=new") options.add_argument("--no-sandbox") options.add_argument("--disable-dev-shm-usage") options.add_argument("--window-size=1400,900") chromedriver_candidates = [ PROJECT_ROOT / "chromedriver", PROJECT_ROOT / "chromedriver.exe", ] driver = None last_err = None for path in chromedriver_candidates: if path.is_file(): try: driver = webdriver.Chrome(service=Service(str(path)), options=options) break except Exception as e: last_err = e if driver is None: try: driver = webdriver.Chrome(options=options) except Exception as e: return {"passed": False, "error": f"selenium init failed: {last_err or e}"} try: driver.get(html_path.resolve().as_uri()) driver.set_window_size(1400, 900) driver.implicitly_wait(1) result = driver.execute_script(r""" const measure = (el) => ({ clientWidth: el.clientWidth, clientHeight: el.clientHeight, scrollWidth: el.scrollWidth, scrollHeight: el.scrollHeight, excess_x: Math.max(0, el.scrollWidth - el.clientWidth), excess_y: Math.max(0, el.scrollHeight - el.clientHeight), overflowed: (el.scrollWidth > el.clientWidth + 5) || (el.scrollHeight > el.clientHeight + 5), }); const slide = document.querySelector('.slide'); if (!slide) return { error: '.slide not found' }; const slideM = measure(slide); slideM.size_correct = slide.clientWidth === 1280 && slide.clientHeight === 720; const body = document.querySelector('.slide-body'); const bodyM = body ? measure(body) : null; const zones = []; slide.querySelectorAll('.zone').forEach((z) => { const pos = z.getAttribute('data-zone-position') || 'unknown'; const tid = z.getAttribute('data-template-id') || '?'; const m = measure(z); m.position = pos; m.template_id = tid; // 내부 clipping 검사 — frame-family root/cell 단위. // tolerance / threshold 그대로. inner_content_signals 만 추가 보강 (detection 데이터 늘림). const clipped = []; z.querySelectorAll('[class*="f13b"], [class*="f29b"], [class*="f16b"]').forEach((el) => { const dx = el.scrollWidth - el.clientWidth; const dy = el.scrollHeight - el.clientHeight; if (dx > 5 || dy > 5) { // inner content signals — clipped cell 안에 *어떤 종류의 콘텐츠가 들어있는지* 보고. // classifier 가 frame_internal_cell 만 봐서는 부족하니 inner 까지 본다. const inner_signals = []; if (el.querySelector('.transform-block, .transform-row, .transform-rows')) { inner_signals.push('structural_unit'); } if (el.querySelector('table')) { inner_signals.push('tabular'); } if (el.querySelector('.text-line')) { inner_signals.push('text_flow'); } clipped.push({ class_name: el.className, inner_content_signals: inner_signals, excess_x: Math.max(0, dx), excess_y: Math.max(0, dy), clientWidth: el.clientWidth, clientHeight: el.clientHeight, scrollWidth: el.scrollWidth, scrollHeight: el.scrollHeight, }); } }); m.clipped_inner = clipped; zones.push(m); }); // B5 v0 — frame_slot_metrics (per-cell measurement of [data-frame-slot-id]) // 현재 F29 partial 만 marker 보유 (process_column / product_column × 3 cells = 6 entries 기대). // 다른 frame (F13 / F16) 은 marker 미적용 → entry 0 — 정상. const frame_slot_metrics = []; slide.querySelectorAll('[data-frame-slot-id]').forEach((cell) => { const slotId = cell.getAttribute('data-frame-slot-id'); const m2 = measure(cell); const parentZone = cell.closest('.zone'); const zonePos = parentZone ? (parentZone.getAttribute('data-zone-position') || 'unknown') : 'unknown'; const zoneTid = parentZone ? (parentZone.getAttribute('data-template-id') || '?') : '?'; frame_slot_metrics.push({ zone_position: zonePos, zone_template_id: zoneTid, frame_slot_id: slotId, class_name: cell.className, clientWidth: m2.clientWidth, clientHeight: m2.clientHeight, scrollWidth: m2.scrollWidth, scrollHeight: m2.scrollHeight, excess_x: m2.excess_x, excess_y: m2.excess_y, overflowed: m2.overflowed, }); }); return { slide: slideM, slide_body: bodyM, zones, frame_slot_metrics }; """) screenshot_path = html_path.parent / "preview.png" try: driver.save_screenshot(str(screenshot_path)) result["screenshot"] = str(screenshot_path.relative_to(PROJECT_ROOT)) except Exception as e: result["screenshot_error"] = str(e) finally: driver.quit() if "error" in result: return {"passed": False, **result} fail_reasons = [] if not result["slide"]["size_correct"]: fail_reasons.append( f"slide size != 1280x720 (got {result['slide']['clientWidth']}x{result['slide']['clientHeight']})" ) if result["slide"]["overflowed"]: fail_reasons.append( f"slide overflowed by {result['slide']['excess_y']}px (vert) / {result['slide']['excess_x']}px (horiz)" ) if result.get("slide_body") and result["slide_body"]["overflowed"]: fail_reasons.append( f"slide-body overflowed by {result['slide_body']['excess_y']}px (vert)" ) for z in result["zones"]: if z["overflowed"]: fail_reasons.append( f"zone--{z['position']} ({z['template_id']}) overflowed by {z['excess_y']}px (vert) / {z['excess_x']}px (horiz)" ) for c in z.get("clipped_inner", []): fail_reasons.append( f"zone--{z['position']}: inner clipped .{c['class_name']} — " f"excess {c['excess_y']}px vert / {c['excess_x']}px horiz " f"(content {c['scrollHeight']} vs container {c['clientHeight']})" ) result["passed"] = len(fail_reasons) == 0 result["fail_reasons"] = fail_reasons return result def write_overflow_error(run_dir: Path, overflow: dict) -> Path: error_data = { "stage": "visual_runtime_check", "reason": "Visual runtime contract 위반 — slide / slide-body / zone overflow / clipping.", "fail_reasons": overflow.get("fail_reasons", []), "details": overflow, } err_path = run_dir / "error.json" err_path.write_text(json.dumps(error_data, ensure_ascii=False, indent=2), encoding="utf-8") return err_path # ─── Debug.json (single slide + zones[]) ─────────────────────── def compute_slide_status(sections: list[MdxSection], units: list[CompositionUnit], comp_debug: dict, overflow: dict, adapter_needed_units: Optional[list[dict]] = None, debug_zones: Optional[list[dict]] = None) -> dict: """Slide 산출물의 정확한 상태 계산 — 자동 파이프라인 결과 보고. 축 : - rendered : final.html 이 디스크에 쓰였는가 - visual_check_passed : Selenium per-zone overflow / clipping 통과 여부 - full_mdx_coverage : aligned 된 모든 section_id 가 어떤 selected unit 에 의해 covered - adapter_needed_count : mapper FitError 로 자동 렌더 못 한 unit 수 (별 review 개념 X — 자동 실패 보고) - content_truncated_count : builder 가 truncate 한 zone 수 (informational) overall enum : PASS — visual OK + full coverage + adapter_needed=0 RENDERED_WITH_VISUAL_REGRESSION — full coverage 이지만 visual fail PARTIAL_COVERAGE — 일부 section 필터됨, 렌더된 부분만 visual OK PARTIAL_COVERAGE_WITH_VISUAL_REGRESSION — 둘 다 (adapter_needed > 0 시 status note 추가, overall 은 위 enum 사용) """ aligned_ids = [s.section_id for s in sections] covered = set() for u in units: covered.update(u.source_section_ids) filtered_ids = sorted(set(aligned_ids) - covered) full_coverage = len(filtered_ids) == 0 visual_passed = bool(overflow.get("passed", False)) adapter_needed_units = list(adapter_needed_units or []) content_truncated = [] for z in (debug_zones or []): tc = z.get("content_truncated_count") if tc: content_truncated.append({ "position": z["position"], "source_section_ids": z["source_section_ids"], "template_id": z["v4_template_id"], "truncated_count": tc, }) # 필터된 section 의 사유 (auto pipeline 결정 트레이스 — review 개념 X) filtered_section_reasons = [] for c in comp_debug.get("candidates_summary", []): if c.get("selection_state") == "selected": continue cand_ids = c.get("source_section_ids", []) if any(sid in filtered_ids for sid in cand_ids): filtered_section_reasons.append({ "section_ids": cand_ids, "merge_type": c.get("merge_type"), "template_id": c.get("template_id"), "v4_label": c.get("label"), "phase_z_status": c.get("phase_z_status"), "score": c.get("score"), "selection_state": c.get("selection_state"), # filtered_status / filtered_weak / filtered_lost "filter_reasons": c.get("filter_reasons", []), }) if full_coverage and visual_passed: overall = "PASS" elif full_coverage and not visual_passed: overall = "RENDERED_WITH_VISUAL_REGRESSION" elif not full_coverage and visual_passed: overall = "PARTIAL_COVERAGE" else: overall = "PARTIAL_COVERAGE_WITH_VISUAL_REGRESSION" return { "rendered": True, "visual_check_passed": visual_passed, "full_mdx_coverage": full_coverage, "aligned_section_ids": aligned_ids, "covered_section_ids": sorted(covered), "filtered_section_ids": filtered_ids, "filtered_section_reasons": filtered_section_reasons, "visual_fail_reasons": list(overflow.get("fail_reasons") or []), "adapter_needed_count": len(adapter_needed_units), "adapter_needed_units": adapter_needed_units, "content_truncated_count": len(content_truncated), "content_truncated_units": content_truncated, "overall": overall, "note": ( "자동 파이프라인 결과 보고. review/UI 개념 X. final.html 파일명 != PASS 의미. " "overall == PASS 는 visual OK + full coverage + adapter_needed=0 일 때만. " "adapter_needed_count > 0 = mapper 가 contract 와 안 맞아 자동 렌더 못 한 zone 존재. " "content_truncated_count > 0 = builder 가 truncate 한 zone 존재 (rendered 됐지만 일부 콘텐츠 손실)." ), } def write_debug_json(run_dir: Path, layout_preset: str, debug_zones: list[dict], layout_css: dict, visual_runtime_check: Optional[dict] = None, composition_debug: Optional[dict] = None, slide_status: Optional[dict] = None, fit_classification: Optional[dict] = None, router_decision: Optional[dict] = None, retry_trace: Optional[dict] = None) -> Path: debug = { "v4_source": str(V4_RESULT_PATH.relative_to(PROJECT_ROOT)), "v4_label_to_phase_z_status": V4_LABEL_TO_PHASE_Z_STATUS, "mvp1_allowed_statuses": sorted(MVP1_ALLOWED_STATUSES), "mode": "composition_v0_layout_8preset", "mode_note": ( "MVP-1.5b w/ composition planner v0 — sections → candidates (separate / " "parent_merged) → score → greedy select → 8-preset layout vocabulary " "(single / horizontal-2 / vertical-2 / top-1-bottom-2 / top-2-bottom-1 / " "left-1-right-2 / left-2-right-1 / grid-2x2). v0 layout = count-based; " "v1 axes (cardinality_fit / hierarchy_coherence / density_score) 추후." ), "layout_preset": layout_preset, "layout_css": layout_css, "slide_status": slide_status, "fit_classification": fit_classification, "router_decision": router_decision, "retry_trace": retry_trace, "composition_planner_debug": composition_debug, "zones": debug_zones, "visual_runtime_check": visual_runtime_check, } debug_path = run_dir / "debug.json" debug_path.write_text(json.dumps(debug, ensure_ascii=False, indent=2), encoding="utf-8") return debug_path # ─── Main entry ──────────────────────────────────────────────── def run_phase_z2_mvp1(mdx_path: Path, run_id: Optional[str] = None) -> Path: """MVP-1.5b entry — single slide + composition planner v0 + 8 preset vocabulary. Pipeline : parse_mdx → align_sections_to_v4_granularity → plan_composition → mapper per unit → render slide_base + frame partial → Selenium check """ mdx_path = Path(mdx_path) if run_id is None: run_id = time.strftime("%Y%m%d_%H%M%S") + "_phase_z2" run_dir = RUNS_DIR / run_id / "phase_z2" print(f"[Phase Z-2 MVP-1.5b] start — mdx={mdx_path.name}, run_id={run_id}") # 1. Parse MDX (V4 무관) slide_title, sections, slide_footer = parse_mdx(mdx_path) print(f" parsed : title='{slide_title}', sections={len(sections)} " f"({[s.section_id for s in sections]}), footer={'yes' if slide_footer else 'no'}") # 2. Load V4 v4 = load_v4_result() # 3. Align sections to V4 granularity (### drill if needed) sections = align_sections_to_v4_granularity(sections, v4) print(f" aligned : sections={len(sections)} ({[s.section_id for s in sections]})") # 4. Composition planner v0 — replaces per-section + select_layout_preset. # candidate (separate / parent_merged) → score → greedy non-overlapping select → # layout preset (count-based v0). def lookup_fn(sid: str) -> Optional[V4Match]: return lookup_v4_match(v4, sid) units, layout_preset, comp_debug = plan_composition( sections, lookup_fn, V4_LABEL_TO_PHASE_Z_STATUS, MVP1_ALLOWED_STATUSES, capacity_fit_fn=compute_capacity_fit, ) if not units or layout_preset is None: # composition planner 결과 = 0 units. Sections 가 모두 V4 lookup 실패 또는 # status filter 통과 못 함. error.json 기록 후 abort. run_dir.mkdir(parents=True, exist_ok=True) error_data = { "stage": "composition_planner", "reason": ( "Composition planner v0 selected 0 viable units. " f"Either no V4 entries for any section, or all candidates filtered out by " f"allowed_statuses={sorted(MVP1_ALLOWED_STATUSES)}." ), "aligned_section_ids": [s.section_id for s in sections], "composition_debug": comp_debug, } err_path = run_dir / "error.json" err_path.write_text(json.dumps(error_data, ensure_ascii=False, indent=2), encoding="utf-8") print(f"\n[Phase Z-2 MVP-1.5b] ABORT @ composition_planner", file=sys.stderr) print(f" reason : 0 viable units after composition v0", file=sys.stderr) print(f" error : {err_path}", file=sys.stderr) sys.exit(1) print(f" preset : {layout_preset} ({len(units)} units, composition v0 count-based)") for u in units: print(f" unit : {u.source_section_ids} merge={u.merge_type} → " f"frame {u.frame_number} ({u.frame_template_id}) " f"label={u.label} score={u.score:.3f}") # 5. Per-unit: synthesize MdxSection → mapper → assets → zone data # mapper FitError 는 catch — 자동 파이프라인은 다른 zone 계속 진행. abort X. positions = LAYOUT_PRESETS[layout_preset]["positions"] zones_data = [] debug_zones = [] adapter_needed_units: list[dict] = [] for i, unit in enumerate(units): position = positions[i] if i < len(positions) else f"zone_{i}" synth_section = MdxSection( section_id="+".join(unit.source_section_ids), section_num=0, title=unit.title, raw_content=unit.raw_content, ) contract = get_contract(unit.frame_template_id) builder_name = contract["payload"].get("builder") visual_hints = contract.get("visual_hints") or {} min_height_px = visual_hints.get("min_height_px", DEFAULT_ZONE_MIN_HEIGHT_PX) contract_frame_id = contract.get("frame_id") # mapper 시도 — 실패 (FitError) 시 zone 을 adapter_needed 로 표시하고 skip try: slot_payload = map_mdx_to_slots(synth_section, unit.frame_template_id) except FitError as e: adapter_record = { "position": position, "source_section_ids": unit.source_section_ids, "merge_type": unit.merge_type, "template_id": unit.frame_template_id, "fit_error": str(e), } adapter_needed_units.append(adapter_record) print(f" adapter : zone--{position} {unit.source_section_ids} → " f"{unit.frame_template_id} FitError → adapter_needed (skip render)") continue run_dir.mkdir(parents=True, exist_ok=True) assets_dir = copy_assets(unit.frame_template_id, run_dir) content_weight = compute_content_weight(synth_section) truncated_count = slot_payload.get("_truncated_count") # builder 가 truncate 한 경우 zones_data.append({ "position": position, "template_id": unit.frame_template_id, "slot_payload": slot_payload, "content_weight": content_weight, "min_height_px": min_height_px, }) debug_zones.append({ "position": position, "source_section_ids": unit.source_section_ids, "merge_type": unit.merge_type, "title": unit.title, "v4_rank1_frame_id": unit.frame_id, "v4_rank1_frame_number": unit.frame_number, "v4_template_id": unit.frame_template_id, "v4_label": unit.label, "v4_confidence": unit.confidence, "phase_z_status": unit.phase_z_status, "composition_score": unit.score, "composition_rationale": unit.rationale, "composition_notes": list(unit.notes), # parent_merged_inferred 정보 신호 "mapper_type": "contract", "contract_id": unit.frame_template_id, "contract_frame_id": contract_frame_id, "builder": builder_name, "min_height_px": min_height_px, "slot_payload_keys": sorted(slot_payload.keys()), "content_truncated_count": truncated_count, # None / N (builder 가 N 개 자름) "assets_dir": str(assets_dir.relative_to(run_dir)) if assets_dir else None, "content_weight": content_weight, }) # 6. Build layout CSS — horizontal-2 = dynamic heights (regression preserve), 그 외 = fr default layout_css = build_layout_css(layout_preset, zones_data) if layout_css["dynamic_rows"]: for dz, h, r in zip(debug_zones, layout_css["heights_px"], layout_css["ratios"]): dz["height_px"] = h dz["ratio"] = r print(f" zones : heights {layout_css['heights_px']} px, ratios {layout_css['ratios']}") else: print(f" zones : fr default ({layout_css['cols']} / {layout_css['rows']})") # 7. Render single slide html = render_slide(slide_title, slide_footer, zones_data, layout_preset, layout_css) # 8. Write final.html out_path = run_dir / "final.html" out_path.write_text(html, encoding="utf-8") print(f" html : {out_path}") # 9. Selenium check print(" visual : running per-zone overflow check ...") overflow = run_overflow_check(out_path) # 10. fit_classifier v0 (A1) — Selenium 결과 → spec §3 category 분류 layer. # *분류만*. action / router / rerender X. behavior 변경 0. fit_classification = classify_visual_runtime_check(overflow, debug_zones) # 11. overflow_router v0 (A2) — category → proposed_action 매핑 layer. # *매핑까지만*. 실행 / rerender / behavior 변경 X. # classifications 각 entry 에 proposed_action 추가, router_decision summary 반환. router_decision = route_fit_classification(fit_classification) # 11.5 zone_ratio_retry action (A3) — A3 locked rules (사용자 잠금) 그대로. # retry budget = 1, slide-base 고정, donor 룰, (b) revert 정책. retry_trace = _attempt_zone_ratio_retry( run_dir=run_dir, out_path=out_path, slide_title=slide_title, slide_footer=slide_footer, zones_data=zones_data, debug_zones=debug_zones, layout_preset=layout_preset, layout_css=layout_css, overflow=overflow, fit_classification=fit_classification, router_decision=router_decision, gap_px=GRID_GAP, ) # retry 가 *성공* 했으면 overflow / fit_classification / router_decision / debug_zones 를 # post-retry 상태로 갱신 (slide_status 가 새 상태 반영하도록). if retry_trace.get("retry_passed"): overflow = retry_trace["post_retry_overflow"] debug_zones = retry_trace["post_retry_debug_zones"] layout_css = retry_trace["post_retry_layout_css"] # post-retry classifier / router 재실행 — 새 overflow 가 통과면 router_active=False fit_classification = classify_visual_runtime_check(overflow, debug_zones) router_decision = route_fit_classification(fit_classification) # 11.6 retry_failure_classifier + next_action_router (A4 — 분류/매핑만, 실행 X) # retry 실패 시 failure_type 분류 + next_proposed_action 기록 (escalation 후보). enrich_retry_trace_with_failure_classification(retry_trace) # 12. Slide status — 자동 파이프라인 결과 보고 (review/UI 개념 X) slide_status = compute_slide_status( sections, units, comp_debug, overflow, adapter_needed_units=adapter_needed_units, debug_zones=debug_zones, ) # 13. Debug.json debug_path = write_debug_json( run_dir, layout_preset, debug_zones, layout_css, overflow, composition_debug=comp_debug, slide_status=slide_status, fit_classification=fit_classification, router_decision=router_decision, retry_trace=retry_trace, ) print(f" debug : {debug_path}") # 13. Status report overall = slide_status["overall"] print(f" status : {overall}") if slide_status["filtered_section_ids"]: print(f" filtered_sections = {slide_status['filtered_section_ids']}") if slide_status["adapter_needed_count"]: print(f" adapter_needed_count = {slide_status['adapter_needed_count']}") if slide_status["content_truncated_count"]: print(f" content_truncated = " f"{[(c['position'], c['truncated_count']) for c in slide_status['content_truncated_units']]}") if not slide_status["visual_check_passed"]: for r in (overflow.get("fail_reasons") or [])[:3]: print(f" visual_fail = {r}") # fit_classifier + router 결과 요약 if not fit_classification["visual_check_passed"]: cats = fit_classification["categories_seen"] print(f" fit_categories = {cats}") if router_decision["router_active"]: actions = router_decision["proposed_actions_summary"] status = router_decision["implementation_status_summary"] missing = router_decision["missing_actions_pending_impl"] print(f" proposed_actions = {actions}") print(f" impl_status_summary = {status}") if missing: print(f" missing_actions = {missing} (현재 미구현 → abort)") # retry 결과 요약 (A3) + failure classification / next action proposal (A4) if retry_trace.get("retry_attempted"): print(f" retry_action = {retry_trace['retry_action']}") print(f" retry_passed = {retry_trace['retry_passed']}") if not retry_trace["retry_passed"]: print(f" retry_failure = {retry_trace.get('retry_failure_reason')}") fc = retry_trace.get("failure_classification") or {} nap = retry_trace.get("next_action_proposal") or {} if fc.get("failure_type"): print(f" failure_type = {fc['failure_type']}") if nap.get("next_proposed_action"): print(f" next_proposed_action = {nap['next_proposed_action']} " f"(impl_status={nap.get('next_action_implementation_status')})") # 13. Exit 정책 — visual fail 은 abort, partial coverage 는 abort 안 하지만 PASS 도 아님 if not slide_status["visual_check_passed"]: err_path = write_overflow_error(run_dir, overflow) print(f"\n[Phase Z-2 MVP-1.5b] FAIL @ visual_runtime_check ({overall})", file=sys.stderr) for reason in overflow.get("fail_reasons", [overflow.get("error", "unknown")]): print(f" - {reason}", file=sys.stderr) print(f" error : {err_path}", file=sys.stderr) sys.exit(1) if not slide_status["full_mdx_coverage"]: print( f"\n[Phase Z-2 MVP-1.5b] PARTIAL — visual check OK 지만 " f"sections {slide_status['filtered_section_ids']} 가 composition planner 에서 " f"필터됨 (allowed_statuses 미통과). final.html 은 viable units 만 렌더된 " f"partial artifact. full MDX slide 아님." ) return out_path print( f"\n[Phase Z-2 MVP-1.5b] {overall} — visual check OK + full MDX coverage. " f"최종 사용자 브라우저 검증 후 ship 가능." ) return out_path if __name__ == "__main__": if len(sys.argv) < 2: print("Usage: python phase_z2_pipeline.py [run_id]", file=sys.stderr) sys.exit(2) mdx = Path(sys.argv[1]) rid = sys.argv[2] if len(sys.argv) > 2 else None run_phase_z2_mvp1(mdx, rid)