"""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 os import re import shutil import sys import time from dataclasses import asdict, 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, load_frame_contracts, 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 # trace-only runtime 연결 v0 — B1 → B4 chain. # final.html / mapper / render path 미영향. debug_zones[i].placement_trace 만 기록. from phase_z2_content_extractor import extract_content_objects from phase_z2_placement_planner import plan_placement # ─── 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") # ─── trace-only runtime 연결 v0 — B1 → B4 chain (final.html 영향 X) ─── # B1~B4 의 dormant chain 을 *real MDX runtime data* 로 처음 호출. # 결과 (PlacementPlan) = debug_zones[i].placement_trace 로 *기록만*. # render path / mapper output / final.html 모두 미변경 — B5 baseline SHA 유지. # B4 frame selection = catalog declaration order (V4 evidence 미사용 — 별 axis). content_objects = extract_content_objects(synth_section) placement_plan = plan_placement( content_objects=content_objects, frame_contracts=list(load_frame_contracts().values()), section_id=synth_section.section_id, ) mapper_frame_template_id = unit.frame_template_id matches_mapper = ( placement_plan.selected_template_id == mapper_frame_template_id ) match_note: Optional[str] = None if not matches_mapper: if placement_plan.selected_template_id is None: match_note = "no_frame_covers_content_types" else: match_note = ( f"B4 selected '{placement_plan.selected_template_id}' but " f"mapper uses '{mapper_frame_template_id}' (composition V4 rank-1)" ) placement_trace = { **asdict(placement_plan), "mapper_frame_template_id": mapper_frame_template_id, "frame_selection_matches_mapper": matches_mapper, "frame_selection_match_note": match_note, } # ─── end trace-only runtime 연결 v0 ─── # ─── B4 gatekeeper (Q-V4B4 / PHASE_Z_B4_GATEKEEPER, default OFF) ─── if ( os.environ.get("PHASE_Z_B4_GATEKEEPER", "").strip().lower() in {"1", "true", "yes"} and not matches_mapper ): adapter_record = { "position": position, "source_section_ids": unit.source_section_ids, "merge_type": unit.merge_type, "template_id": unit.frame_template_id, "reason": "v4_b4_mismatch", "mismatch_detail": { "v4_template_id": mapper_frame_template_id, "b4_selected_template_id": placement_plan.selected_template_id, "match_note": match_note, }, } adapter_needed_units.append(adapter_record) print(f" adapter : zone--{position} {unit.source_section_ids} → " f"{unit.frame_template_id} v4_b4_mismatch → adapter_needed (skip render)") continue # ─── end B4 gatekeeper ─── # 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, "reason": "fit_error", "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, # trace-only runtime 연결 v0 — B1 → B2 → B4 chain 결과 (render 미영향). "placement_trace": placement_trace, }) # 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)