"""실험 Phase 1 (GPT 버전) — 옥탑방 관련 샷 클러스터링 + image chain 구조 LLM(GPT-5.4) 설계. Gemini 버전과 동일한 출력(JSON + tree.txt + diagram.mmd), 모델만 GPT-5.4. """ from __future__ import annotations import argparse import base64 import glob import json import logging import os import sys from pathlib import Path BACKEND = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BACKEND)) from dotenv import load_dotenv # noqa: E402 load_dotenv(BACKEND / ".env") from openai import OpenAI # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger("chain_planner_gpt") SYSTEM_PROMPT = """You are a visual continuity planner for a film BACKGROUND image pipeline. INPUT LANGUAGE NOTICE — READ FIRST: The user message will contain raw scenario text in Korean (scene headings, scene body text, shot descriptions) and JSON character lists with Korean proper names. These Korean strings are CONTEXT ONLY for extracting spatial/material/lighting cues. You MUST: - IGNORE every proper name (character names, place names, production titles) in the Korean source — never reproduce them, never paraphrase them. - Output ALL labels, ids, descriptions, rationales in English COMMON NOUNS only. - NEVER include Korean characters, Hanja, or Japanese kana anywhere in your JSON output (ids, labels, descriptions, rationales — all must be ASCII English). - Treat the Korean text purely as a source for visible-space details (room layout, doors, windows, lighting state, time of day, surface condition). Do not narrate plot or characters. CRITICAL CONCEPT — each node = ONE background image, possibly shared by MANY shots: - A node represents ONE photorealistic background image that will be rendered. - A node has a `shot_ids` array listing ALL shots whose background needs are satisfied by this single image. - Two shots share a node when the INTERSECTION of the wall/door/window/furniture/lighting requirements both demand is large enough that one image can serve as the background for both. - Close-ups, prop inserts, hand-scale plates, "background plate" shots, photograph inserts, and similar tight views DO NOT need their own background image — they reuse the parent group's background. List them under the parent group's `shot_ids`. - DIFFERENT STATES of the same physical space ARE different nodes: * different time-of-day (day / night / dusk / dawn) * different room state (clean / lived-in / disturbed / heavily-ransacked) * different lighting setup (window light only / television glow only / dawn spill / etc.) * different active fixed-element zones (e.g. "open living-room window with torn curtain" vs. "closed window") When the lighting OR the visible surface state differs enough that a single image cannot serve both shots, split into separate nodes. - Same state, same room → ONE node, even if many shots use it. You will receive: 1. ONE OR MORE top-down architectural FLOOR PLANS (images attached, each tagged with its base_plan_id in the text immediately before the image). 2. The full SHOT inventory — for each shot: scene_index, shot_index, description, base_plan_id (may be missing for some), visual_domain, optional camera/additions detail when available. 3. The base_photo specs (already-defined wide anchor views per base_plan). Your job is to design a BACKGROUND-IMAGE chain that maximizes cross-shot visual consistency. Pipeline constraints: - Each node renders ONE image; that image is then reused by every shot in the node's `shot_ids`. - The node's image-gen request takes: * the PARENT node's already-rendered image as the reference * a t2i prompt that describes this node's background view + reuses parent's atmosphere/material/lighting - Anchor nodes (kind=anchor_synthetic / base_photo with no parent) have no parent — they are rendered first, with floor-plan as reference only. GOAL: produce a small, sharply-defined set of background nodes — typically 10–20 nodes for a single dwelling location across a full episode — NOT one node per shot. EVERY shot in the input must end up listed in exactly ONE node's shot_ids. Group / cluster heuristics: - Group all shots that show the SAME spatial subset (e.g., "small bedroom interior", "living-kitchen entry corner", "rooftop terrace overlook"). One group = one connected sub-tree. - A group's anchor is the most distinctive WIDE view. If a base_photo already exists for that base_plan, reuse it as the anchor (kind=base_photo). Otherwise propose a synthetic anchor (kind=anchor_synthetic). - Within a group, prefer SHALLOW trees (depth <= 2). Deep chains accumulate drift. - A shot whose camera is INSIDE a sub-room (e.g., reveal through a doorway) should be parented to the corresponding doorway-anchor shot, not the wide overview, because the doorway frame is the strongest shared anchor. - A shot that is a CLOSE-UP of a prop already visible in another shot should be parented to that other shot. - Different base_plans (interior vs. rooftop terrace exterior) form SEPARATE trees — no cross-plan parenting unless one shot truly contains both (rare). Sub-region coverage (IMPORTANT — read carefully): - Inspect the floor plan IMAGE for each base_plan and identify every clearly-bounded interior sub-region (e.g. living-kitchen, small bedroom, main bedroom, bathroom, entry, hallway). Use the room labels printed on the plan plus visible wall divisions. - For EACH sub-region that is NOT already covered by an existing shot, you MUST propose an additional `anchor_synthetic` node so the chain produces a complete spatial inventory: * id: snake_case, e.g. `anchor_main_bedroom_inside`, `anchor_bathroom_inside` * kind: `anchor_synthetic` * source: empty string "" * label: short English label, e.g. "main bedroom interior anchor" * description: 2-3 English sentences for an eye-level photorealistic view standing inside that sub-region * parent_id: the wide-anchor node of the same base_plan (depth=0 root). Depth = 1. * rationale: explain that this fills the spatial gap and reuses the parent's wall finish / floor / lighting via the doorway connection * shared_visual_anchors_with_parent: include the doorway frame, the relevant wall finish, lighting tone — anchors visible in BOTH the parent wide view and this sub-region's view (typically through the doorway). - Likewise for exterior base_plans, if the plan covers multiple distinct outdoor zones (terrace floor / parapet / stair landing / water-tank corner), propose `anchor_synthetic` nodes for any zone that an existing shot doesn't already frame. - These synthetic anchors come BEFORE any shot that should logically be parented to them. If a shot exists that takes place inside that sub-region (e.g. a doorway reveal into a bedroom), reparent the shot to the synthetic anchor when the synthetic anchor is the closer shared-visual-anchor source. For EACH node output: - id: unique snake_case id (e.g., "interior_living_kitchen_day_normal", "interior_living_kitchen_dusk_ransacked", "small_bedroom_inside_night", "rooftop_terrace_night", "courtyard_stairs_approach"). For base_photo-anchored nodes you may use the base_photo id directly. - kind: one of "anchor_synthetic" | "base_photo" (we no longer use "shot" — every node is a reusable background) - source: base_photo id (for kind=base_photo) or empty string "" - source_plan_id: the base_plan id that this node's image visualizes — REQUIRED for every node so the renderer can find the floor plan PNG. Must match one of the base_plan ids listed in the input. For non-root child nodes, use the same source_plan_id as the parent unless the child genuinely visualizes a different base_plan. NEVER leave empty unless this node really has no associated floor plan (extreme rare case — e.g. an aerial site view with no plan). - label: short human-readable English label naming the BACKGROUND view + state - description: 2-3 sentences in English describing the photorealistic background — the wall/door/window/furniture layout, lighting state, time of day, surface condition (clean / disturbed / ransacked etc.). NO narrative events. - shot_ids: array of shot ids (e.g. "S12_Shot4") — every shot whose background can be served by this single image. Ranges from 1 to many shots per node. EVERY shot in the input must end up in exactly ONE node's shot_ids. - parent_id: id of parent node, or empty string for root anchors. Choose the parent that shares the most spatial/material continuity with this node (typically the same physical space in a different state, or the wide anchor of the same base_plan). - depth: 0 for root, 1 for child of root, etc. - rationale: one sentence explaining the parent choice — what visual element will be reused from parent's image - shared_visual_anchors_with_parent: list of short noun phrases that MUST appear in BOTH parent and child images so the chain stays consistent. Empty list for root anchors. Also output: - groups: list of {name (snake_case), shared_canon (one paragraph describing material/lighting/palette canon for this group), node_ids} - execution_order: topological ordering — every parent strictly before its children. - rationale_summary: one paragraph explaining the overall structure choices. - unassigned_shots: array of shot ids that could not be placed (should be empty in a healthy plan). STRICT RULES: - ENGLISH ONLY in all labels, descriptions, rationales. - COMMON NOUNS only — NEVER use proper names of characters, places, or productions. - No narrative events in descriptions — focus on visible space/material/light only. No people, no blood. - Every input shot must appear in exactly ONE node's shot_ids. No shot left unplaced. No shot duplicated. - Group close-ups, prop inserts, hand-scale plates, "background plate" shots, photograph inserts together with the wide background they share — do NOT create dedicated nodes for them. - DO split when state differs (day vs. night vs. dusk; clean vs. disturbed vs. ransacked) such that one image cannot serve both. - Aim for a small total node count: typically 10–20 nodes for a full single-location episode. Do not exceed 25 nodes unless the input genuinely demands it. - execution_order MUST contain every node id exactly once, with parents before children. - Use empty string "" for missing source/parent_id (do NOT use the literal string "null"). Return strict JSON. No markdown fence. """ RESPONSE_SCHEMA = { "type": "object", "properties": { "rationale_summary": {"type": "string"}, "groups": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "shared_canon": {"type": "string"}, "node_ids": {"type": "array", "items": {"type": "string"}}, }, "required": ["name", "shared_canon", "node_ids"], "additionalProperties": False, }, }, "nodes": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "kind": {"type": "string", "enum": ["anchor_synthetic", "base_photo"]}, "source": {"type": "string"}, "source_plan_id": {"type": "string"}, "label": {"type": "string"}, "description": {"type": "string"}, "shot_ids": {"type": "array", "items": {"type": "string"}}, "parent_id": {"type": "string"}, "depth": {"type": "integer"}, "rationale": {"type": "string"}, "shared_visual_anchors_with_parent": { "type": "array", "items": {"type": "string"}, }, }, "required": [ "id", "kind", "source", "source_plan_id", "label", "description", "shot_ids", "parent_id", "depth", "rationale", "shared_visual_anchors_with_parent", ], "additionalProperties": False, }, }, "execution_order": {"type": "array", "items": {"type": "string"}}, "unassigned_shots": {"type": "array", "items": {"type": "string"}}, }, "required": ["rationale_summary", "groups", "nodes", "execution_order", "unassigned_shots"], "additionalProperties": False, } def collect_context(run_dir: Path, base_plan_ids: list[str], extra_keywords: list[str] | None = None, use_scope: bool = True) -> dict: """전수조사 — 다중 source 합집합: 1) `context.scope.scenes` (ground truth) — episode의 명시적 scope에 있는 모든 scene 2) `step2.shot_assignment`에서 base_plan_id가 base_plan_ids에 속하는 shot들의 scene 3) (fallback) heading/text keyword 매칭 — 위 둘이 비었을 때만 각 shot은 step3_shot_*.json detailed가 있으면 첨부, 없으면 description만. in_scope_reason 라벨로 어느 source에서 들어왔는지 기록. """ bp_set = set(base_plan_ids) extra_keywords = extra_keywords or ["옥탑", "옥상", "rooftop"] ctx: dict = {"base_plans": [], "base_photos": [], "shots": [], "scenes_in_scope": [], "locations_in_scope": [], "spatial": None, "plan_pngs": {}} sp2_path = run_dir / "step2_plan_specs.json" sp2 = json.loads(sp2_path.read_text(encoding="utf-8")) if sp2_path.exists() else {} for p in sp2.get("base_plans", []): if p.get("id") in bp_set: ctx["base_plans"].append(p) png_p = run_dir / f"base_plan_{p['id']}.png" if png_p.exists(): ctx["plan_pngs"][p["id"]] = png_p sp7_path = run_dir / "step7_photo_specs.json" if sp7_path.exists(): d = json.loads(sp7_path.read_text(encoding="utf-8")) for ph in d.get("base_photos", []): if ph.get("source_plan_id") in bp_set: ctx["base_photos"].append(ph) # shot_assignment: (scene, shot) → base_plan_id assign_map: dict = {} for a in sp2.get("shot_assignment", []): si = a.get("scene_index"); sx = a.get("shot_index") assign_map[(si, sx)] = (a.get("base_plan_id"), a.get("visual_domain")) # context.json — 전체 shot/scene 인벤토리 ctx_path = run_dir / "context.json" ctx_json = json.loads(ctx_path.read_text(encoding="utf-8")) if ctx_path.exists() else {} # === Scope detection: 우선순위 순서대로 합집합 === scenes_in_scope: set = set() scope_reasons: dict = {} # scene_index → reason # (1) context.scope.scenes — ground truth (있으면 항상 사용) if use_scope: for si in (ctx_json.get("scope", {}).get("scenes") or []): scenes_in_scope.add(si) scope_reasons.setdefault(si, "context.scope.scenes") # (2) shot_assignment에 base_plan_id가 in-scope인 shot들의 scene for (si, sx), (bp, _) in assign_map.items(): if bp in bp_set and si is not None: scenes_in_scope.add(si) scope_reasons.setdefault(si, "shot_assignment.base_plan_id") # (3) keyword fallback — 위 둘이 비었을 때만 if not scenes_in_scope: for sc in ctx_json.get("scenes", []): h = sc.get("heading", "") txt = (sc.get("text") or "") if any(kw in h or kw in txt for kw in extra_keywords): scenes_in_scope.add(sc.get("scene_index")) scope_reasons.setdefault(sc.get("scene_index"), "keyword_fallback") # locations in scope (ground truth from context.scope.locations + ctx.locations 메타) scope_loc_ids = set(ctx_json.get("scope", {}).get("locations") or []) for loc in ctx_json.get("locations", []): if loc.get("short_id") in scope_loc_ids: ctx["locations_in_scope"].append({ "short_id": loc.get("short_id"), "name": loc.get("name"), "description": loc.get("description"), "visual_traits": loc.get("visual_traits") or [], }) # 각 scene의 heading + 전문 추가 (LLM 컨텍스트용) for sc in ctx_json.get("scenes", []): si = sc.get("scene_index") if si in scenes_in_scope: ctx["scenes_in_scope"].append({ "scene_index": si, "heading": sc.get("heading", ""), "text": sc.get("text") or "", "in_scope_reason": scope_reasons.get(si, "unknown"), }) # shot 합집합: scope scene 안의 모든 shot seen: set = set() for s in ctx_json.get("shots", []): si = s.get("scene_index"); sx = s.get("shot_index") key = (si, sx) if key in seen: continue bp, dom = assign_map.get(key, (None, None)) is_in_scope = (bp in bp_set) or (si in scenes_in_scope) if not is_in_scope: continue seen.add(key) # step3 detailed가 있으면 부착 step3_path = run_dir / f"step3_shot_S{si:02d}_Shot{sx}.json" step3 = json.loads(step3_path.read_text(encoding="utf-8")) if step3_path.exists() else None ctx["shots"].append({ "scene_index": si, "shot_index": sx, "shot_id": f"S{si:02d}_Shot{sx}", "description": s.get("description") or "", "characters": s.get("characters") or [], "base_plan_id": bp, "visual_domain": dom, "step3": step3, "in_scope_reason": "shot_assignment" if (bp in bp_set) else "scene_keyword", }) sp1_path = run_dir / "step1_spatial.json" if sp1_path.exists(): ctx["spatial"] = json.loads(sp1_path.read_text(encoding="utf-8")) return ctx def build_user_prompt(ctx: dict) -> str: parts: list[str] = [] parts.append("== BASE PLANS IN SCOPE ==") for p in ctx["base_plans"]: parts.append(f"\n--- base_plan id={p['id']} ---") parts.append(f"label: {p.get('label')}") parts.append(f"visual_domain: {p.get('visual_domain')}") parts.append(f"is_anchor: {p.get('is_anchor')}") parts.append(f"anchor_cluster_id: {p.get('anchor_cluster_id')}") parts.append("\n\n== BASE PHOTOS (existing wide anchors) ==") for ph in ctx["base_photos"]: parts.append(f"\n--- base_photo id={ph['id']} ---") parts.append(f"source_plan_id: {ph.get('source_plan_id')}") parts.append(f"label: {ph.get('label')}") parts.append(f"visual_domain: {ph.get('visual_domain')}") parts.append(f"camera_note: {ph.get('camera_note', '')}") parts.append(f"lighting: {ph.get('lighting', '')}") parts.append("\n\n== LOCATIONS IN SCOPE (ground truth from context.scope.locations) ==") parts.append("Use these as anchor concepts for grouping shots into background nodes. Korean names are CONTEXT ONLY — do not reproduce.") for loc in ctx.get("locations_in_scope", []): parts.append(f"\n--- {loc.get('short_id')} ---") parts.append(f"name (Korean source): {loc.get('name')}") parts.append(f"description: {loc.get('description')}") parts.append(f"visual_traits: {json.dumps(loc.get('visual_traits') or [], ensure_ascii=False)}") parts.append("\n\n== SCENES IN SCOPE (Korean source text, ignore proper names) ==") for sc in ctx.get("scenes_in_scope", []): parts.append(f"\n--- S{sc['scene_index']:02d} reason={sc.get('in_scope_reason')} ---") parts.append(f"heading: {sc.get('heading', '')}") parts.append(f"text:\n{sc.get('text', '')}") parts.append("\n\n== SHOTS (rooftop-related, full inventory) ==") parts.append("Source legend: in_scope_reason = 'shot_assignment' (mapped to a rooftop base_plan) or 'scene_keyword' (scene heading/text contains rooftop terms; base_plan may be missing).") for s in ctx["shots"]: sid = s.get("shot_id") bp = s.get("base_plan_id") or "(unmapped)" dom = s.get("visual_domain") or "(unknown)" reason = s.get("in_scope_reason") parts.append(f"\n--- shot id={sid} base_plan={bp} domain={dom} reason={reason} ---") if s.get("description"): parts.append(f"description: {s['description']}") if s.get("characters"): parts.append(f"characters: {json.dumps(s['characters'], ensure_ascii=False)}") step3 = s.get("step3") if step3: cam = step3.get("camera") or {} parts.append(f"camera position: {cam.get('position', '(none)')}") parts.append(f"camera heading: {cam.get('heading', '(none)')}") parts.append(f"height/fov/lens: {cam.get('height')} / {cam.get('fov')} / {cam.get('lens_note')}") adds = step3.get("additions") or [] if adds: parts.append("additions:") for a in adds: parts.append(f" + [{a.get('type')}] @ {a.get('position', '?')}: {a.get('note', '')}") parts.append("\n\n== INTERIOR / ENVIRONMENT CANON ==") sp = ctx.get("spatial") or {} canon = sp.get("environment_canon") or {} if canon: parts.append(json.dumps(canon, ensure_ascii=False, indent=2)) parts.append("\n\n== TASK ==") parts.append("Use the floor plan IMAGES (attached, each tagged with its base_plan_id immediately before the image) and ALL info above to design the image-chain structure.") parts.append("Output JSON with: rationale_summary, groups, nodes (every shot must appear), execution_order (topological).") parts.append('Use empty string "" for missing source / parent_id (NOT "null").') return "\n".join(parts) def call_gpt_with_plans(client: OpenAI, model: str, plan_pngs: dict[str, Path], system_prompt: str, user_prompt: str, response_schema: dict) -> dict: user_content: list = [{"type": "text", "text": user_prompt}] for plan_id, png_path in plan_pngs.items(): user_content.append({"type": "text", "text": f"\n[Floor plan image follows — base_plan_id={plan_id}]"}) with open(png_path, "rb") as f: b64 = base64.b64encode(f.read()).decode("ascii") user_content.append({ "type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}, }) resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}, ], response_format={ "type": "json_schema", "json_schema": { "name": "ChainStructure", "schema": response_schema, "strict": True, }, }, ) txt = resp.choices[0].message.content or "" return json.loads(txt) def render_tree_ascii(plan: dict) -> str: nodes = {n["id"]: n for n in plan.get("nodes", [])} children: dict[str, list[str]] = {} roots: list[str] = [] for n in plan.get("nodes", []): pid = n.get("parent_id") or "" if pid in ("", "null", "None") or pid not in nodes: roots.append(n["id"]) else: children.setdefault(pid, []).append(n["id"]) lines: list[str] = [] def walk(nid: str, prefix: str, is_last: bool) -> None: n = nodes[nid] connector = "└── " if is_last else "├── " sids = n.get("shot_ids") or [] sid_summary = f" [{len(sids)} shots: {', '.join(sids[:6])}{'...' if len(sids) > 6 else ''}]" if sids else "" lines.append( f"{prefix}{connector}[{n.get('kind', '?')}] {nid} d={n.get('depth', '?')} — {n.get('label', '?')}{sid_summary}") kids = children.get(nid, []) new_prefix = prefix + (" " if is_last else "│ ") for i, k in enumerate(kids): walk(k, new_prefix, i == len(kids) - 1) for i, r in enumerate(roots): walk(r, "", i == len(roots) - 1) return "\n".join(lines) def validate_chain_plan(plan: dict, ctx: dict) -> list[str]: """GPT 응답 직후 의미적 검증 — JSON schema가 보장 못하는 invariant 확인. 검사 항목: 1. 모든 input shot이 정확히 1번 등장 (누락 0, 중복 0) 2. 모든 parent_id가 nodes에 존재 3. execution_order가 모든 노드 정확히 1번 + parent-before-child 4. group node_ids 모두 nodes에 존재 5. base_photo source가 ctx.base_photos에 존재 6. source_plan_id가 ctx.base_plans에 존재 (또는 빈 문자열 — 매우 예외적) """ errors: list[str] = [] nodes = plan.get("nodes", []) nodes_by_id = {n["id"]: n for n in nodes} # 1. 모든 input shot이 정확히 1번 input_shot_ids = {s["shot_id"] for s in ctx.get("shots", [])} seen_in_nodes: dict = {} for n in nodes: for sid in (n.get("shot_ids") or []): if sid in seen_in_nodes: errors.append( f"shot {sid} duplicated: nodes={seen_in_nodes[sid]} and {n['id']}") seen_in_nodes[sid] = n["id"] missing = input_shot_ids - set(seen_in_nodes.keys()) if missing: errors.append(f"shots missing from any node.shot_ids: {sorted(missing)}") extra = set(seen_in_nodes.keys()) - input_shot_ids if extra: errors.append(f"shots in node.shot_ids but not in input ctx.shots: {sorted(extra)}") # 2. parent_id가 nodes에 존재 for n in nodes: pid = n.get("parent_id") or "" if pid and pid not in ("null", "None") and pid not in nodes_by_id: errors.append(f"node {n['id']} parent_id={pid!r} not found in nodes") # 3. execution_order: 모두 한 번씩 + parent-before-child execution_order = plan.get("execution_order", []) seen_in_order: set = set() for nid in execution_order: n = nodes_by_id.get(nid) if not n: errors.append(f"execution_order entry {nid!r} not in nodes") continue if nid in seen_in_order: errors.append(f"execution_order entry {nid!r} duplicated") pid = n.get("parent_id") or "" if pid and pid not in ("null", "None") and pid not in seen_in_order: errors.append(f"execution_order: {nid} comes before its parent {pid}") seen_in_order.add(nid) missing_in_order = set(nodes_by_id.keys()) - set(execution_order) if missing_in_order: errors.append(f"execution_order missing nodes: {sorted(missing_in_order)}") # 4. group node_ids 모두 nodes에 존재 for g in plan.get("groups", []): for nid in g.get("node_ids", []): if nid not in nodes_by_id: errors.append(f"group {g.get('name')!r} references unknown node_id={nid!r}") # 5. base_photo source가 ctx.base_photos에 존재 photo_ids = {ph["id"] for ph in ctx.get("base_photos", [])} for n in nodes: if n.get("kind") == "base_photo": src = n.get("source") or "" if not src: errors.append(f"base_photo node {n['id']} missing source field") elif src not in photo_ids: errors.append(f"base_photo node {n['id']} source={src!r} not in ctx.base_photos {sorted(photo_ids)}") # 6. source_plan_id 검증 plan_ids = {bp["id"] for bp in ctx.get("base_plans", [])} for n in nodes: spid = n.get("source_plan_id") or "" if not spid: # parent가 있으면 inheritance로 OK pid = n.get("parent_id") or "" if not pid or pid in ("null", "None"): errors.append(f"root node {n['id']} missing source_plan_id (cannot find floor plan PNG)") elif spid not in plan_ids: errors.append(f"node {n['id']} source_plan_id={spid!r} not in ctx.base_plans {sorted(plan_ids)}") return errors def render_mermaid(plan: dict) -> str: out = ["flowchart TD"] for n in plan.get("nodes", []): nid = n["id"]; kind = n.get("kind", "?"); label = n.get("label", "?") node_label = f"{nid}
{label}" if kind == "anchor_synthetic": shape = f'(("{node_label}"))' elif kind == "base_photo": shape = f'[/"{node_label}"/]' else: shape = f'["{node_label}"]' out.append(f" {nid}{shape}") for n in plan.get("nodes", []): pid = n.get("parent_id") or "" if pid and pid not in ("null", "None", ""): shared = ", ".join((n.get("shared_visual_anchors_with_parent") or [])[:3]) label = f"|{shared}|" if shared else "" out.append(f" {pid} -->{label} {n['id']}") return "\n".join(out) def main() -> int: p = argparse.ArgumentParser() p.add_argument("--run-dir", required=True) p.add_argument("--base-plan-ids", required=True) p.add_argument("--out-dir", required=True) p.add_argument("--text-model", default="gpt-5.5") p.add_argument("--skip-text", action="store_true") p.add_argument("--skip-validation", action="store_true", help="semantic validator 실패해도 통과 (디버깅용 — 사용 비권장)") args = p.parse_args() run_dir = Path(args.run_dir).resolve() out_dir = Path(args.out_dir).resolve() out_dir.mkdir(parents=True, exist_ok=True) if not os.getenv("OPENAI_API_KEY"): logger.error("OPENAI_API_KEY not set"); return 1 client = OpenAI() base_plan_ids = [s.strip() for s in args.base_plan_ids.split(",") if s.strip()] plan_p = out_dir / "chain_structure.json" if args.skip_text and plan_p.exists(): plan = json.loads(plan_p.read_text(encoding="utf-8")) logger.info("[skip text] reusing chain_structure.json") else: ctx = collect_context(run_dir, base_plan_ids) logger.info("collected: %d base_plans, %d base_photos, %d shots, %d plan_pngs", len(ctx["base_plans"]), len(ctx["base_photos"]), len(ctx["shots"]), len(ctx["plan_pngs"])) if not ctx["shots"] and not ctx["base_photos"]: logger.error("no shots/photos matched base_plan_ids=%s", base_plan_ids); return 1 user_prompt = build_user_prompt(ctx) logger.info("user prompt: %d chars; calling %s", len(user_prompt), args.text_model) plan = call_gpt_with_plans(client, args.text_model, ctx["plan_pngs"], SYSTEM_PROMPT, user_prompt, RESPONSE_SCHEMA) plan["_meta"] = { "base_plan_ids": base_plan_ids, "shot_count": len(ctx["shots"]), "base_photo_count": len(ctx["base_photos"]), "text_model": args.text_model, "provider": "openai", } plan_p.write_text(json.dumps(plan, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("saved chain_structure.json") # === Semantic validation (Medium 6) === errors = validate_chain_plan(plan, ctx) if errors: logger.error("semantic validation FAILED — %d issue(s):", len(errors)) for e in errors: logger.error(" • %s", e) (out_dir / "validation_errors.txt").write_text( "\n".join(errors) + "\n", encoding="utf-8") if not args.skip_validation: logger.error("re-run with --skip-validation to ignore (not recommended).") return 1 else: logger.info("semantic validation OK — all invariants hold") tree_txt = render_tree_ascii(plan) (out_dir / "tree.txt").write_text(tree_txt + "\n", encoding="utf-8") (out_dir / "diagram.mmd").write_text(render_mermaid(plan) + "\n", encoding="utf-8") print("\n=== RATIONALE SUMMARY ===") print(plan.get("rationale_summary", "(missing)")) print("\n=== GROUPS ===") for g in plan.get("groups", []): print(f"\n• {g.get('name')}: {len(g.get('node_ids', []))} nodes") print(f" shared_canon: {g.get('shared_canon', '')}") print(f" node_ids: {', '.join(g.get('node_ids', []))}") print("\n=== TREE ===") print(tree_txt) print("\n=== EXECUTION ORDER ===") for i, nid in enumerate(plan.get("execution_order", []), 1): n = next((x for x in plan.get("nodes", []) if x.get("id") == nid), {}) pid = n.get("parent_id") or "" if pid in ("", "null", "None"): pid = "(root)" print(f" {i:2d}. {nid} [{n.get('kind', '?')}, d={n.get('depth', '?')}] ← parent={pid}") print("\n=== NODE DETAILS ===") for n in plan.get("nodes", []): pid = n.get("parent_id") or "" if pid in ("", "null", "None"): pid = "(root)" print(f"\n[{n['id']}] kind={n.get('kind')} depth={n.get('depth')} parent={pid}") print(f" label: {n.get('label')}") print(f" description: {n.get('description')}") print(f" rationale: {n.get('rationale')}") sva = n.get("shared_visual_anchors_with_parent") or [] if sva: print(f" shared_visual_anchors: {', '.join(sva)}") print(f"\nWrote: {plan_p}") return 0 if __name__ == "__main__": sys.exit(main())