"""location_floor_plan pipeline 모듈 — 도면 prompt 생성 + PNG 호출. Phase 3 (v3 plan, 2026-04-29). spike test 6 패턴 production화. Phase 5 T4 (2026-04-29). prompt v2 multi-turn context + ref_paths multi-image edit. 순수 함수: - build_user_prompt: location 메타 + scenes + shots → user_prompt - validate_prompt_length: 500 ≤ len ≤ 6000 LLM 호출: - generate_floor_plan_prompt: gpt-5.5 → 영문 도면 prompt - generate_floor_plan_image: gpt-image-2 → PNG bytes (text-only or multi-image edit) """ from __future__ import annotations import logging import time from pathlib import Path from typing import Any, Dict, List, Optional from app.core.image_call_budget import ( ImageCallBudgetExceeded, reserve_current_call, ) from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes logger = logging.getLogger(__name__) PROMPT_LEN_MIN = 500 PROMPT_LEN_MAX = 6000 LLM_RETRY_MAX = 3 LLM_RETRY_BACKOFF_STEP = 2 # linear: 2, 4, 6 sec (linear progression, not exponential) IMAGE_RETRY_MAX = 3 IMAGE_SIZE = "1024x1024" IMAGE_QUALITY = "high" def build_user_prompt( template: str, location_short_id: str, location_label: str, scenes: List[Dict[str, Any]], selected_shots: List[Dict[str, Any]], visual_world_rules: str, previous_floor_plans_block: str = "", building_group_context: str = "", ) -> str: """user_template에 placeholder 치환. v2 prompt는 multi-turn 컨텍스트용 placeholder 2개 추가: - previous_floor_plans_block: 이전 도면 요약 (없으면 "") - building_group_context: 같은 building_group 도면 ref 참조 안내 (없으면 "") v1 template에는 이 placeholder가 없으므로 .format()이 무시 — backward-compat. """ scenes_lines: List[str] = [] for sc in scenes: scenes_lines.append(f"### Scene {sc['scene_index']} — {sc.get('heading', '')}") scenes_lines.append(sc.get("text", "")) scenes_lines.append("") scenes_text = "\n".join(scenes_lines).rstrip() by_scene: Dict[int, List[Dict[str, Any]]] = {} for sh in selected_shots: by_scene.setdefault(sh["scene_index"], []).append(sh) shots_lines: List[str] = [] for si in sorted(by_scene): shots_lines.append(f"### Scene {si} selected shots:") for sh in by_scene[si]: shots_lines.append(f"- Shot {sh['shot_index']}: {sh.get('description', '')}") shots_lines.append("") selected_shots_text = "\n".join(shots_lines).rstrip() return template.format( location_short_id=location_short_id, location_label=location_label, scenes_text=scenes_text, selected_shots_text=selected_shots_text, visual_world_rules=(visual_world_rules or "").strip(), previous_floor_plans_block=(previous_floor_plans_block or "").strip(), building_group_context=(building_group_context or "").strip(), ) def validate_prompt_length(text: str) -> None: """gpt-image-2 prompt 길이 검증. 500 ≤ len ≤ 6000.""" n = len(text) if n < PROMPT_LEN_MIN: raise ValueError(f"floor plan prompt too short: {n} < {PROMPT_LEN_MIN}") if n > PROMPT_LEN_MAX: raise ValueError(f"floor plan prompt too long: {n} > {PROMPT_LEN_MAX}") def generate_floor_plan_prompt( *, system_prompt: str, user_prompt: str, project_config: Any, opik_metadata: Optional[Dict[str, Any]] = None, call_text_fn=None, ) -> str: """gpt-5.5에 도면 prompt 자동 작성 요청. retry + length 검증. call_text_fn: 테스트용 주입. None이면 app.modules.llm.llm_client.call_text 사용. """ if call_text_fn is None: from app.modules.llm.llm_client import call_text call_text_fn = call_text last_exc: Optional[Exception] = None for attempt in range(LLM_RETRY_MAX + 1): try: text = call_text_fn( step="location_floor_plan", system_prompt=system_prompt, user_prompt=user_prompt, project_config=project_config, opik_metadata=opik_metadata or {}, ) text = (text or "").strip() validate_prompt_length(text) return text except Exception as exc: last_exc = exc if attempt < LLM_RETRY_MAX: delay = LLM_RETRY_BACKOFF_STEP * (attempt + 1) logger.warning( "generate_floor_plan_prompt retry %d/%d (sleep %ds): %s", attempt + 1, LLM_RETRY_MAX, delay, exc, ) time.sleep(delay) continue break raise RuntimeError( f"generate_floor_plan_prompt failed after {LLM_RETRY_MAX} retries: {last_exc}" ) def generate_floor_plan_image( *, prompt: str, openai_client: Any, ref_paths: Optional[List[Path]] = None, model: str = "gpt-image-2", size: str = IMAGE_SIZE, quality: str = IMAGE_QUALITY, capture_role: Optional[str] = "floor_plan_image", capture_input_image_ids: Optional[List[str]] = None, capture_extra_metadata: Optional[Dict[str, Any]] = None, ) -> bytes: """gpt-image-2 호출 → PNG bytes 반환. retry 3회. ref_paths가 None/빈 리스트/모두 invalid면 text-only(images.generate). 1+ valid이면 multi-image edit(images.edit). 유효성 기준: 파일 존재 + size >= 1024 bytes. Phase 5 T4: building_group multi-turn ref 도입. backward-compat — ref_paths 미지정 시 Phase 3 동작과 동일. ``capture_role`` — 이 함수는 floor_plan / outdoor aerial·blocking·sketch / registered pose guide 등 여러 용도의 수렴점이므로, capture pipeline_role 은 호출자가 결정한다(Phase C scope 배선). default-off 라 scope 미개방 시 미사용. """ valid_refs: List[Path] = [] for p in (ref_paths or []): if p is None: continue try: if p.exists() and p.stat().st_size >= 1024: valid_refs.append(p) except OSError as exc: logger.warning("generate_floor_plan_image: ref_path stat failed %r: %s", p, exc) last_exc: Optional[Exception] = None # gpt-image 호출 + b64 decode 는 primitive wrapper 로 위임(생성물 capture 동시, # scope 미배선이면 no-op). reserve/too-small 검증/retry 는 여기 유지. ★이 사이트는 # ref 1개라도 image=[리스트](항상 multi-edit) → edit_image_as_list=True 로 보존. call_kwargs = {"model": model, "size": size, "quality": quality, "n": 1} def _capture_meta(source: str, ref_count: int) -> Dict[str, Any]: # 호출자 추가 메타(group_id/stage 등) + 사이트 고정 메타(budget_source/ref_count). # ★고정 메타가 우선 — caller extra 가 budget_source/ref_count 를 덮지 못한다 # (budget label 불변 규칙, Codex minor 보강). capture_extra_metadata None(default) # 이면 기존과 동일 = byte-identical. meta: Dict[str, Any] = dict(capture_extra_metadata or {}) meta["budget_source"] = source meta["ref_count"] = ref_count return meta for attempt in range(IMAGE_RETRY_MAX + 1): try: if not valid_refs: reserve_current_call(source="location_floor_plan.generate") png_bytes = call_gpt_image_bytes( openai_client, mode="generate", prompt=prompt, ref_paths=None, call_kwargs=call_kwargs, capture_role=capture_role, capture_input_image_ids=capture_input_image_ids, capture_metadata=_capture_meta("location_floor_plan.generate", 0), ) else: reserve_current_call(source="location_floor_plan.edit") png_bytes = call_gpt_image_bytes( openai_client, mode="edit", prompt=prompt, ref_paths=valid_refs, call_kwargs=call_kwargs, capture_role=capture_role, capture_input_image_ids=capture_input_image_ids, edit_image_as_list=True, capture_metadata=_capture_meta( "location_floor_plan.edit", len(valid_refs)), ) if not png_bytes or len(png_bytes) < 1024: raise RuntimeError( f"empty or too-small PNG returned: {len(png_bytes)} bytes" ) return png_bytes except ImageCallBudgetExceeded: raise except Exception as exc: last_exc = exc if attempt < IMAGE_RETRY_MAX: delay = LLM_RETRY_BACKOFF_STEP * (attempt + 1) logger.warning( "generate_floor_plan_image retry %d/%d (sleep %ds): %s", attempt + 1, IMAGE_RETRY_MAX, delay, exc, ) time.sleep(delay) continue break raise RuntimeError( f"generate_floor_plan_image failed after {IMAGE_RETRY_MAX} retries: {last_exc}" )