"""LocationFloorPlanStep — planner-driven sequential 도면 PNG 생성. Phase 5 T5 (v3 plan, 2026-04-29). Phase 3 ThreadPoolExecutor 병렬을 planner.floor_plan_order 기반 sequential 처리로 전환. 설계: - 입력: background_planner 체크포인트 (data.floor_plans, floor_plan_order) - 처리: floor_plan_order 순차 — 한 번에 한 도면씩. * 이전 도면 1줄 요약 inject (multi-turn context) * 같은 building_group의 prev PNG ref 첨부 (multi-image edit) - 출력: * PNG 파일 (1 per fp_id) * ImageAsset(asset_type='floor_plan', variant_index=0, variant_label='v00') UPSERT - 토글: settings.background_mode == 'floor_plan_anchored' 체크포인트 data 구조 (Strategy A: 양쪽 shape 동시 보존): - 신 shape: data.floor_plans[fp_id] = {fp_id, status, prompt_text, png_path, primary_location_id, building_group, location_ids, ref_used, failure_reason?} - 호환 alias: data.locations[] = [{id, prompt_text, image_path, status}, ...] (Phase 4의 chain_bg_planning/render readers 호환용) - 메타: data.applicable_count, succeeded_count, failed_count, schema_version, config_hash, floor_plan_order """ from __future__ import annotations import hashlib import json import logging import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 3 # scene_texts inject — v3 prompt # C4 v1 — v4 prompt: room-count cue 의 한국어 방 이름/이동 동사 + slash-zone 한국어 # 예시를 generic semantic principle 로 추상화 (Prompt Closed-List Ban). # config_hash invalidation. SCHEMA_VERSION 3 유지. PROMPT_VERSION = "4" # location_floor_plan v4 (multi-room from scene texts) class LocationFloorPlanStep(StepRunner): """planner-driven sequential 도면 PNG 생성 + DB 등록.""" # ── checkpoint loader ── def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning( "location_floor_plan: %s checkpoint parse failed: %s", step_id, exc, ) return None return None # ── _execute ── def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings from app.core.errors import AppError # 토글 가드 — applicability=if_floor_plan_mode가 차단하더라도 직접 호출 안전망. if settings.background_mode != "floor_plan_anchored": logger.info( "location_floor_plan: skipped — background_mode=%s", settings.background_mode, ) return self._empty_result() from app.core.openai_keys import has_openai_key if not has_openai_key(): raise AppError( code="step.no_openai_key", message="OPENAI_API_KEY 미설정 — OpenAI 텍스트/이미지 모델 호출 불가.", status_code=400, ) planner_cp = self._load_prev_checkpoint("background_planner") planner_data = (planner_cp or {}).get("data", {}) or {} fp_list: List[Dict[str, Any]] = planner_data.get("floor_plans") or [] fp_order: List[str] = list(planner_data.get("floor_plan_order") or []) if not fp_list or not fp_order: logger.info( "location_floor_plan: planner has no floor_plans (cp_exists=%s)", planner_cp is not None, ) return self._empty_result() # spec lookup by id fp_specs: Dict[str, Dict[str, Any]] = {fp.get("id", ""): fp for fp in fp_list if fp.get("id")} # location canon mapping location_canon_by_short, location_label_by_short = self._load_location_canons() # rules text (optional) rules_cp = self._load_prev_checkpoint("visual_world_rules") rules_text = "" if rules_cp: data = rules_cp.get("data", {}) or {} rules_text = data.get("rules_text", "") or data.get("text", "") or "" if not rules_text: logger.warning( "location_floor_plan: visual_world_rules가 비어있음 — " "prompt에 세계관 컨텍스트 누락" ) # prompt 로드 from app.modules.prompt_loader import load_prompt system_prompt = load_prompt("location_floor_plan", "system", db=self.db) user_template = load_prompt("location_floor_plan", "user_template", db=self.db) # scene texts + selected shot descriptions per location # — v3: location 메타만으로는 다중 실 vs 단일 실 추론이 부정확. # 원본 segment 텍스트 + 선택 샷 description을 inject (truncation 절대 금지). loc_to_scenes = self._build_loc_to_scenes() segments_by_idx = self._load_scene_segments() shots_by_scene_idx, selected_shot_idx_by_scene = self._load_shot_data() # 정책: scene_director / scene_save 둘 중 하나라도 누락이면 graceful # downgrade — fp 생성은 진행하지만 LLM에 원본 씬 텍스트 컨텍스트가 빠져 # 다중 실 인식이 v2 수준(label only)으로 회귀한다. 운영자가 인지하도록 # 한 번 더 loud한 warning을 남긴다 (개별 helper의 warning은 silent해 보일 # 수 있음). 실패 처리(raise)는 하지 않는다 — chain_only / off 모드 연동성 # 보장 + 다른 step들의 graceful 패턴과 일관. if not loc_to_scenes: logger.warning( "location_floor_plan: scene_director loc→scenes 비어있음 — " "v3 multi-room 추론이 label-only 수준으로 downgrade. " "scene_director step 재실행 후 force 권장." ) if not segments_by_idx: logger.warning( "location_floor_plan: scene_save segments 비어있음 — " "원본 시나리오 본문 inject 불가. label-only fallback." ) image_dir = ( Path(settings.projects_dir) / self.project_id / "images" / self.episode_id / "floor_plan" ) image_dir.mkdir(parents=True, exist_ok=True) # SEQUENTIAL: planner.floor_plan_order에 따라 한 번에 한 도면 처리 rendered_summaries: Dict[str, str] = {} # fp_id → 1-line summary rendered_paths: Dict[str, Path] = {} # fp_id → PNG Path results: Dict[str, Dict[str, Any]] = {} for fp_id in fp_order: spec = fp_specs.get(fp_id) if not spec: logger.warning( "location_floor_plan: fp_id %s in order but missing in floor_plans", fp_id, ) continue # Phase 5.2 — needs_floor_plan=false면 PNG 생성 skip. # 단일실 단일 state 같이 도면이 불필요한 그룹은 chain_bg만 text-only로 진행. # 호환: 필드 누락 시 default true (이전 v1/v2 plan과 동일 동작). if spec.get("needs_floor_plan", True) is False: rationale = spec.get("complexity_rationale", "") or "(no rationale)" logger.info( "location_floor_plan: %s skipped — needs_floor_plan=false (%s)", fp_id, rationale, ) results[fp_id] = { "status": "skipped", "prompt_text": "", "png_path": "", "fp_id": fp_id, "primary_location_id": spec.get("primary_location_id", ""), "building_group": spec.get("building_group", ""), "location_ids": list(spec.get("location_ids") or []), "ref_used": "skipped", "skip_reason": rationale, "needs_floor_plan": False, } # rendered_summaries / rendered_paths에는 추가하지 않는다 — 후속 그룹의 # multi-image edit ref로도 사용되면 안 되므로. continue # 이전에 처리된 도면들의 1줄 요약 — order 의존 (현재까지 rendered만) prev_summaries: List[Tuple[str, str]] = [ (oid, rendered_summaries[oid]) for oid in fp_order if oid != fp_id and oid in rendered_summaries ] # 같은 building_group의 prev PNG ref cur_group = spec.get("building_group", "") same_group_refs: List[Path] = [] if cur_group: for oid, ospec in fp_specs.items(): if oid == fp_id: continue if oid not in rendered_paths: continue if ospec.get("building_group", "") == cur_group: same_group_refs.append(rendered_paths[oid]) scenes_for_fp, selected_shots_for_fp = self._build_scenes_for_fp( spec=spec, loc_to_scenes=loc_to_scenes, segments_by_idx=segments_by_idx, shots_by_scene_idx=shots_by_scene_idx, selected_shot_idx_by_scene=selected_shot_idx_by_scene, ) try: result = self._process_floor_plan( fp_id=fp_id, spec=spec, location_canon_by_short=location_canon_by_short, location_label_by_short=location_label_by_short, prev_summaries=prev_summaries, same_group_ref_paths=same_group_refs, system_prompt=system_prompt, user_template=user_template, rules_text=rules_text, image_dir=image_dir, scenes=scenes_for_fp, selected_shots=selected_shots_for_fp, ) except Exception as exc: # _process_floor_plan은 자체적으로 RuntimeError 등을 처리해야 하지만, # 예기치 못한 예외(타입 오류 등)도 단일 fp 실패로 격리하여 다른 fp 진행 보장. logger.error( "location_floor_plan: %s unexpected exception: %s", fp_id, exc, ) result = { "status": "failed", "prompt_text": "", "png_path": "", "fp_id": fp_id, "primary_location_id": spec.get("primary_location_id", ""), "building_group": cur_group, "location_ids": list(spec.get("location_ids") or []), "ref_used": "text_only", "failure_reason": f"{type(exc).__name__}: {exc}", } results[fp_id] = result if result.get("status") == "ok": rendered_summaries[fp_id] = _summarize_prompt(result.get("prompt_text", "")) png_path = result.get("png_path", "") if png_path: rendered_paths[fp_id] = Path(png_path) # 카운트 집계 — skipped(needs_floor_plan=false)도 completed로 포함 (업스트림 sync에 # "처리 완료"로 보고). failed만 별도 카운트. applicable = len(fp_order) completed = sum( 1 for r in results.values() if r.get("status") in ("ok", "skipped") ) failed = sum(1 for r in results.values() if r.get("status") == "failed") # DB UPSERT (best-effort) try: self._register_image_assets(results, fp_specs, location_canon_by_short) except Exception as exc: logger.error("location_floor_plan: DB sync failed (files on disk): %s", exc) try: self.db.rollback() except Exception: pass # Phase 4 backward-compat alias: data.locations[] compat_locations = self._build_compat_locations(results, fp_specs) return { "completed_count": completed, "applicable_count": applicable, "failed_count": failed, # Phase 5.1 — step_runner.save_checkpoint가 누락 시에만 fallback하므로 # SCHEMA_VERSION + PROMPT_VERSION-aware config_hash가 top-level에 보존된다. "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "floor_plans": results, "floor_plan_order": fp_order, "applicable_count": applicable, "succeeded_count": completed, "failed_count": failed, # Phase 4 readers (background_chain_planning/render) compatibility: # 이들은 data.locations[] 리스트를 읽어 prompt_text/image_path/status를 추출한다. # T5는 fp 단위로 작동하지만 primary_location 단위 alias를 노출하여 회귀 0건 보장. "locations": compat_locations, }, } # ── helpers ── def _empty_result(self) -> Dict[str, Any]: return { "completed_count": 0, "applicable_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "floor_plans": {}, "floor_plan_order": [], "applicable_count": 0, "succeeded_count": 0, "failed_count": 0, "locations": [], }, } def _config_hash(self) -> str: from app.core.config import settings payload = { "background_mode": settings.background_mode, "model": "gpt-image-2.5-sunburst", "size": "1024x1024", "quality": "high", "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_location_canons(self) -> Tuple[Dict[str, str], Dict[str, str]]: """EntityCanon에서 location 매핑. {short_id: canon_id}, {short_id: label}.""" from app.models.project import EntityCanon canons = ( self.db.query(EntityCanon) .filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "location", ) .all() ) canon_by_short: Dict[str, str] = {} label_by_short: Dict[str, str] = {} for c in canons: if c.short_id: canon_by_short[c.short_id] = c.id label_by_short[c.short_id] = (getattr(c, "name", "") or c.short_id) return canon_by_short, label_by_short # ── scene context loaders (v3) ────────────────────────────────────── def _build_loc_to_scenes(self) -> Dict[str, List[int]]: """scene_director.scenes[].primary_location → {short_id: [scene_index, ...]}. scene_director는 각 scene의 주요 location을 short_id (e.g., "L05") 형태로 보존. primary_location이 dict 형태로 저장된 fallback 케이스도 short_id를 추출해 처리. """ cp = self._load_prev_checkpoint("scene_director") if not cp: logger.warning( "location_floor_plan: scene_director checkpoint missing — " "원본 씬 텍스트 inject 불가" ) return {} out: Dict[str, List[int]] = {} for sc in (cp.get("data", {}) or {}).get("scenes", []): pl = sc.get("primary_location") if isinstance(pl, dict): lid = pl.get("short_id") or pl.get("id") or "" elif isinstance(pl, str): lid = pl else: lid = "" si = sc.get("scene_index") if lid and isinstance(si, int): out.setdefault(lid, []).append(si) for lid in out: out[lid].sort() return out def _load_scene_segments(self) -> Dict[int, Dict[str, Any]]: """scene_save.data.segments[] → {scene_index: segment dict}. segment dict는 {scene_index, heading, start_char, end_char, length, text} 보존. """ cp = self._load_prev_checkpoint("scene_save") if not cp: logger.warning( "location_floor_plan: scene_save checkpoint missing — " "원본 씬 텍스트 inject 불가" ) return {} out: Dict[int, Dict[str, Any]] = {} for seg in (cp.get("data", {}) or {}).get("segments", []): si = seg.get("scene_index") if isinstance(si, int): out[si] = seg return out def _load_shot_data( self, ) -> Tuple[Dict[int, Dict[int, Dict[str, Any]]], Dict[int, List[int]]]: """shot_extract + shot_selection 로드. 반환: shots_by_scene_idx: {scene_index: {shot_index: shot dict}} selected_shot_idx_by_scene: {scene_index: [shot_index, ...]} """ ext_cp = self._load_prev_checkpoint("shot_extract") sel_cp = self._load_prev_checkpoint("shot_selection") shots: Dict[int, Dict[int, Dict[str, Any]]] = {} for sc in ((ext_cp or {}).get("data", {}) or {}).get("scenes", []): si = sc.get("scene_index") if not isinstance(si, int): continue inner: Dict[int, Dict[str, Any]] = {} for sh in sc.get("shots", []) or []: shi = sh.get("shot_index") if isinstance(shi, int): inner[shi] = sh shots[si] = inner selected: Dict[int, List[int]] = {} for sc in ((sel_cp or {}).get("data", {}) or {}).get("scenes", []): si = sc.get("scene_index") if isinstance(si, int): selected[si] = [ int(x) for x in (sc.get("selected_shot_indices") or []) if isinstance(x, int) ] return shots, selected def _build_scenes_for_fp( self, *, spec: Dict[str, Any], loc_to_scenes: Dict[str, List[int]], segments_by_idx: Dict[int, Dict[str, Any]], shots_by_scene_idx: Dict[int, Dict[int, Dict[str, Any]]], selected_shot_idx_by_scene: Dict[int, List[int]], ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """fp의 location_ids 묶음에 속한 scene들을 모아 (scenes, selected_shots) 반환. - location_ids는 building_group의 모든 멤버 (primary + secondary). - scene 본문(segment.text) 무절단 — CLAUDE.md 원본 전달 절대 규칙. - selected_shots는 shot_selection의 selected_shot_indices와 매칭된 shot_extract 항목만. """ member_ids = list(spec.get("location_ids") or []) # 이중 등장 방지 + 시간 순서 보존 (scene_index 오름차순) scene_order: List[int] = [] seen: set = set() for lid in member_ids: for si in loc_to_scenes.get(lid, []): if si not in seen: seen.add(si) scene_order.append(si) scene_order.sort() scenes_out: List[Dict[str, Any]] = [] selected_shots_out: List[Dict[str, Any]] = [] for si in scene_order: seg = segments_by_idx.get(si) if seg: scenes_out.append({ "scene_index": si, "heading": seg.get("heading", "") or "", "text": seg.get("text", "") or "", }) shot_map = shots_by_scene_idx.get(si, {}) for shi in selected_shot_idx_by_scene.get(si, []): sh = shot_map.get(shi) if not sh: continue selected_shots_out.append({ "scene_index": si, "shot_index": shi, "description": sh.get("description", "") or "", }) return scenes_out, selected_shots_out def _process_floor_plan( self, *, fp_id: str, spec: Dict[str, Any], location_canon_by_short: Dict[str, str], location_label_by_short: Dict[str, str], prev_summaries: List[Tuple[str, str]], same_group_ref_paths: List[Path], system_prompt: str, user_template: str, rules_text: str, image_dir: Path, scenes: List[Dict[str, Any]], selected_shots: List[Dict[str, Any]], ) -> Dict[str, Any]: """단일 floor_plan 처리: prompt 생성 + PNG 생성 + 결과 dict 반환. prev_summaries: [(other_fp_id, 1-line summary), ...] — 텍스트 컨텍스트 inject. same_group_ref_paths: 같은 building_group의 prev PNG paths — multi-image edit ref. """ from app.modules.pipeline.location_floor_plan import ( build_user_prompt, generate_floor_plan_prompt, generate_floor_plan_image, ) primary_loc = spec.get("primary_location_id", "") building_group = spec.get("building_group", "") location_ids = list(spec.get("location_ids") or []) result: Dict[str, Any] = { "status": "failed", "prompt_text": "", "png_path": "", "fp_id": fp_id, "primary_location_id": primary_loc, "building_group": building_group, "location_ids": location_ids, "ref_used": "text_only", "failure_reason": None, } try: # multi-turn 컨텍스트 블록 빌드 previous_block = _build_previous_floor_plans_block(prev_summaries) building_context = _build_building_group_context( same_group_ref_paths, prev_summaries, building_group ) label = location_label_by_short.get(primary_loc, primary_loc) or primary_loc user_prompt = build_user_prompt( template=user_template, location_short_id=primary_loc, location_label=label, scenes=scenes, selected_shots=selected_shots, visual_world_rules=rules_text, previous_floor_plans_block=previous_block, building_group_context=building_context, ) prompt_text = generate_floor_plan_prompt( system_prompt=system_prompt, user_prompt=user_prompt, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) png_bytes = generate_floor_plan_image( prompt=prompt_text, openai_client=_get_openai_client(), ref_paths=list(same_group_ref_paths) if same_group_ref_paths else None, ) png_path = image_dir / f"{fp_id}.png" png_path.write_bytes(png_bytes) # ref_used 분류 — generate_floor_plan_image의 valid 필터 정책과 정합: # 파일이 1024 bytes 이상 & 존재 시에만 실제로 edit API 사용. if not same_group_ref_paths: ref_used = "text_only" elif len(same_group_ref_paths) == 1: ref_used = "single_ref" else: ref_used = "multi_ref" result.update({ "status": "ok", "prompt_text": prompt_text, "png_path": str(png_path), "ref_used": ref_used, }) result.pop("failure_reason", None) logger.info( "location_floor_plan: %s ok (prompt=%d chars, png=%d bytes, refs=%d, ref_used=%s)", fp_id, len(prompt_text), len(png_bytes), len(same_group_ref_paths), ref_used, ) except Exception as exc: logger.warning("location_floor_plan: %s failed: %s", fp_id, exc) result["failure_reason"] = f"{type(exc).__name__}: {exc}" return result def _build_compat_locations( self, results: Dict[str, Dict[str, Any]], fp_specs: Dict[str, Dict[str, Any]], ) -> List[Dict[str, Any]]: """Phase 4 readers 호환용 alias: primary_location 기준 list shape. background_chain_planning_step._load_floor_plan_prompts 와 background_chain_render_step._load_floor_plan_paths 가 data.locations[] 리스트에서 {id (short_id), prompt_text, image_path, status} 필드를 읽는다. 회귀 0건 보장. 주의: 같은 primary_location_id를 가진 floor_plan이 여러 개면(planner는 보통 하나만 만들지만), order 첫 등장만 alias한다 — 마지막 덮어쓰기 회피. """ seen: set = set() compat: List[Dict[str, Any]] = [] for fp_id, result in results.items(): spec = fp_specs.get(fp_id, {}) primary = spec.get("primary_location_id", "") or fp_id if primary in seen: continue seen.add(primary) # image_path: Phase 4 reader는 projects_dir.parent 기준 relative path를 기대. # 실패 시 빈 문자열 — Phase 4 reader가 graceful skip. # to_relative_image_path 는 root 외부면 절대 그대로 반환 (e2e tmp_path 호환). from app.core.file_paths import to_relative_image_path png_path = result.get("png_path", "") or "" rel_path = to_relative_image_path(png_path) if png_path else "" compat.append({ "id": primary, "prompt_text": result.get("prompt_text", "") or "", "image_path": rel_path, "status": result.get("status", "failed"), # 디버그 traceability — fp_id로 신 shape 역참조 "fp_id": fp_id, }) return compat def _register_image_assets( self, results: Dict[str, Dict[str, Any]], fp_specs: Dict[str, Dict[str, Any]], location_canon_by_short: Dict[str, str], ) -> None: """ImageAsset(asset_type='floor_plan', variant_index=0, variant_label='v00') UPSERT. match key (application-level): (project_id, asset_type, entity_id, variant_index). DB constraint 추가 없음 — application UPSERT. Episode 분리: episode_id도 같이 매칭하여 cross-episode 충돌 방지. """ from app.models.project import ImageAsset now = datetime.now(timezone.utc).isoformat() registered = 0 for fp_id, result in results.items(): if result.get("status") != "ok": continue png_path_str = result.get("png_path", "") if not png_path_str: continue # Phase 3 컨벤션: file_path 는 projects_root 기준 relative. # to_relative_image_path 는 root 외부면 절대 그대로 반환 (e2e tmp_path 호환). from app.core.file_paths import to_relative_image_path rel_png_path = to_relative_image_path(png_path_str) spec = fp_specs.get(fp_id, {}) primary_loc = spec.get("primary_location_id", "") canon_id = location_canon_by_short.get(primary_loc) if not canon_id: logger.warning( "location_floor_plan: %s primary_location %s has no canon — skip DB", fp_id, primary_loc, ) continue existing = ( self.db.query(ImageAsset) .filter_by( project_id=self.project_id, asset_type="floor_plan", entity_id=canon_id, variant_index=0, episode_id=self.episode_id, ) .first() ) if existing: existing.file_path = rel_png_path existing.prompt_used = result.get("prompt_text", "") existing.variant_label = "v00" existing.t2i_guide = None existing.is_primary = 1 existing.status = "generated" existing.generation_model = "gpt-image-2.5-sunburst" else: self.db.add(ImageAsset( id=str(uuid.uuid4()), project_id=self.project_id, asset_type="floor_plan", entity_id=canon_id, episode_id=self.episode_id, variant_index=0, variant_label="v00", t2i_guide=None, file_path=rel_png_path, prompt_used=result.get("prompt_text", ""), generation_model="gpt-image-2.5-sunburst", status="generated", is_primary=1, created_at=now, )) registered += 1 if registered: self.db.commit() logger.info("location_floor_plan: registered %d ImageAssets", registered) # ────────────────────────────────────────────── # 모듈 레벨 helpers (step 인스턴스 의존 없음 → 직접 테스트 가능) # ────────────────────────────────────────────── def _summarize_prompt(prompt_text: str) -> str: """LLM이 생성한 floor_plan prompt에서 1줄 요약 추출. CLAUDE.md no-truncation: 원본 prompt는 자르지 않는다 (별도 보존). 이 함수가 반환하는 것은 _다음 LLM 호출에 컨텍스트로 inject할 1줄 요약_으로, 원본 데이터의 일부 슬라이스가 아닌 별개의 표현이다. 추출 규칙: 첫 문장 (마침표/줄바꿈까지) — 원본을 잃지 않으므로 truncation 위반 아님. fallback: 첫 문장 추출 실패 시 빈 줄까지의 첫 단락. """ text = (prompt_text or "").strip() if not text: return "" # 첫 문장 (마침표 + 공백/줄바꿈) for sep in (". ", ".\n", "\n\n"): idx = text.find(sep) if 0 < idx < 200: return text[:idx + 1].strip() # fallback: 첫 줄 first_line = text.split("\n", 1)[0].strip() return first_line def _build_previous_floor_plans_block( prev_summaries: List[Tuple[str, str]], ) -> str: """이전 도면 요약 블록 빌드. 비어있으면 빈 문자열.""" if not prev_summaries: return "" lines = ["[PREVIOUS FLOOR PLANS]"] for fp_id, summary in prev_summaries: # summary는 1줄 — 줄바꿈 보호 cleaned = summary.replace("\n", " ").strip() lines.append(f"- {fp_id}: {cleaned}") return "\n".join(lines) def _build_building_group_context( same_group_ref_paths: List[Path], prev_summaries: List[Tuple[str, str]], building_group: str, ) -> str: """같은 building_group의 prev PNG ref가 있을 때만 컨텍스트 블록 생성.""" if not same_group_ref_paths: return "" # 같은 group의 prev fp_id만 추출 (filename = .png) ref_ids = [p.stem for p in same_group_ref_paths if p.stem] ids_str = ", ".join(ref_ids) if ref_ids else "previously rendered floor plans" return ( "[BUILDING GROUP CONTEXT]\n" f"The previously rendered floor plans {ids_str} belong to the SAME\n" "building_group as this one. Their PNG outputs are attached as references.\n" "Use them as spatial layout authority." ) def _get_openai_client(): """OpenAI 클라이언트 lazy 생성 — 테스트에서 monkeypatch 가능. timeout: settings.llm_timeout_image_gen (env LLM_TIMEOUT_IMAGE_GEN override 가능) — single source. default 600s 의 long hang 회귀 가드. """ from app.core.openai_keys import openai_client from app.core.config import settings return openai_client( timeout=float(settings.llm_timeout_image_gen), )