"""location_aerial — W-L (2026-07-03): 야외 location 항공뷰 배치도(야외 fp) SOT. 같은 야외 location 의 bg plate 들이 샷 관점별로 쪼개진 평면도들을 1순위 ref 로 받아(같은 loc 에 fp 가 여러 장 = fp 분열) plate 끼리 서로 다른 위상으로 갈라지는 결함의 근본 대응. 야외 location 의 위상 SOT 는 **위에서 내려다본 배치도(실내 floor plan 과 같은 도면 언어의 야외판) 1장/loc** 이며, 야외 bg 렌더는 분열된 개별 fp 대신 이것을 1순위 ref 로 받는다(실내 loc 은 기존 실내 fp 1순위 유지 — 이 모듈 무관). ★사용자 7차 정정(2026-07-03) 반영: "항공뷰"는 실사 드론 사진이 아니라 실내 fp 처럼 **위에서 그린 도면형 배치도**를 뜻한다. 또한 건물/구조물에 속한 상층 location 은 그 **구조물 레벨의 평면**으로 그려야 한다 (지면 위 단독 부지로 상상 금지 — 판단은 classify 산출 데이터가 이끈다). - ``build_loc_aerial_context`` — background_classify ``building_groups`` 구조 조인(순수): loc별 is_indoor/group_mixed/label/summary. 파일명/라벨 파싱 0. - ``build_aerial_establishing_prompt`` — 도면형 site-plan T2I 프롬프트(generic 템플릿, location label/summary 는 데이터로만 주입, 시나리오 토큰 0, 마커/텍스트 금지). - ``render_location_aerial`` — loc당 1장 생성(T2I; 같은 building 그룹에 실내가 공존(mixed)하면 indoor floor plan 을 I2I ref 로 footprint 정합). 기존 마커형 항공뷰(outdoor_site_layout_provider.generate_location_aerial_base — composition guide 전용 다이어그램)와 별개 producer. - ``AERIAL_SITE_GUIDANCE`` — bg plate 렌더에 aerial 배치도를 1순위 ref 로 첨부할 때 프롬프트에 덧붙이는 위상 계약(중립 — 도면 스타일 복제 금지). 소비자(background_render)가 flag ``outdoor_aerial_reference_enabled`` ON 일 때만 호출한다(OFF = 미호출 = 기존 경로 byte-identical). """ from __future__ import annotations import logging 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 from app.modules.prompt_sanitizer import PromptSanitizer logger = logging.getLogger(__name__) __all__ = [ "AERIAL_SITE_GUIDANCE", "AERIAL_FP_COHERENCE_GUIDANCE", "PLATE_ORDER_SCHEMA", "build_loc_aerial_context", "build_aerial_establishing_prompt", "compute_aerial_context_hash", "decide_group_plate_order", "render_location_aerial", "validate_plate_order", ] # 2.x — 7차 정정: 실사 establishing → 도면형 top-down site plan. # 3.x — 8차 정정: loc당 1장 → 같은 장소 그룹당 통합 1장(건물 지붕뷰+주변, # 탈것은 전체 형태, 간판 판 위치 표기). # 4.x — W-M(9차): 조금 넓은 프레이밍(주변 컨텍스트 마진) + 개구부 동그라미 # 기호(실내 fp 정합점) + 상층 레벨 시각 단서(옥상형 레벨이 지상 단독 부지로 # 읽히는 결함 대응). 4.202607040120 — canary 육안 후속: 개구부 circle 외 # 원형 장식/무늬 발명 금지(배치도의 장식성 원형 패턴이 plate 표면 문양으로 # 전이된 사례 차단). 이 값이 바뀌면 소비자(background_render)의 aerial 캐시 # 재사용이 무효화된다(이전 의미의 PNG 재사용 방지 — Codex NARROW). AERIAL_PROMPT_VERSION = "4.202607040120" # W-L — bg plate 렌더에 aerial 배치도를 **1순위(첫 번째) ref** 로 첨부할 때 # 프롬프트 끝에 덧붙이는 위상 계약. generic — 부지 배치/진입부/레벨 관계만 # 고정, 앵글/조명/시간대/날씨는 plate 프롬프트가 SOT. 도면의 flat 스타일을 # photoreal 출력에 복제하는 것을 명시 금지. adapter 의 fp_seeded_anchor # guidance("supplied base floor plan ...")가 프롬프트 앞부분에 남는 경우가 # 있어(W-L 은 개별 fp 첨부를 이 배치도로 교체) 마지막 문장이 그 지시의 base # reference 를 이 배치도로 명시 연결한다. AERIAL_SITE_GUIDANCE = ( "\n\nThe FIRST attached image is a flat TOP-DOWN SITE PLAN of this " "location (an outdoor floor plan) — the location's spatial MASTER " "reference. It fixes the real layout: each structure's footprint, size " "and position, the entrances and approach paths, level changes, and " "where the permanently installed elements sit. Keep this render " "consistent with that layout — anything shown here must exist where the " "plan puts it, and the scene must read as standing somewhere INSIDE " "that plan. Do NOT copy the plan's drawing style: no plan lines, flat " "colour fills, symbols or top-down map view in the output — render a " "fully photoreal scene at the camera position and angle this prompt " "describes. The plan marks doorways, windows and other openings with " "small hollow-circle symbols — those circles are drafting notation " "ONLY: never draw circles, rings or marker overlays in this render, " "and never reproduce any circular or ring-shaped marking from the " "plan as a painted pattern, decal or decoration on any surface; at " "each marked position render the real door, window, gate or opening " "itself, on the wall the plan puts it on. " "Where the plan marks blank lettering panels, this render " "shows the place's own lettering in those positions — the same " "characters and mark wherever it appears. Lighting, weather and time " "of day follow " "THIS prompt. If " "any earlier instruction calls a floor plan the base visual reference, " "that means THIS attached site plan." ) # W-L — mixed 그룹(같은 building 그룹에 실내 loc 공존)의 aerial 배치도를 # 생성할 때 indoor floor plan 을 I2I ref 로 첨부하는 정합 계약. 같은 도면 # 언어 산출용 — 기존 BUILDING_FP_AERIAL_GUIDANCE(마커형 schematic map 용)와 # 별개. AERIAL_FP_COHERENCE_GUIDANCE = ( "\n\nThe ATTACHED image is the TOP-DOWN INTERIOR FLOOR PLAN of the " "building this location belongs to — a STRUCTURAL REFERENCE ONLY, never " "content to copy. Draw the site plan described above as a fresh drawing " "in the same drafting style. Keep the building's footprint size, shape, " "proportions and the placement of its doors, windows and level access " "consistent with that interior plan, so the two plans read as the same " "single building. Place the site plan's hollow-circle opening markers " "so they COINCIDE with the interior plan's door, window and stair " "positions on the shared walls — the two plans' openings must line up " "as the same physical openings. Do NOT copy the interior plan's room " "contents, furniture symbols, text or numbers into the site plan." ) def build_loc_aerial_context( building_groups: List[Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: """background_classify ``building_groups`` → loc별 aerial 컨텍스트 (순수). 반환: ``{loc_sid: {"group_id", "is_indoor", "group_mixed", "label", "summary", "anchor_loc", "group_members"}}``. ★8차 정정: aerial 산출 단위는 loc 가 아니라 **같은 장소 그룹** (building_groups 그룹 = 하나의 건물/구조물/부지) — 같은 그룹의 야외 loc 들은 통합 배치도 1장을 공유한다. ``group_members`` 는 그룹 전 멤버(실내 포함)의 {loc_id, label, summary, is_indoor} 리스트로, 통합 배치도 프롬프트의 place 데이터가 된다. building_fp_link.build_group_membership_by_loc 와 같은 구조 조인 규칙 (``is_indoor`` 가 bool 아닌 멤버 제외 — 추측 금지, group_id 없는 그룹 제외, 같은 loc 여러 그룹이면 first-wins = classify 산출 순서 결정론). """ out: Dict[str, Dict[str, Any]] = {} for group in building_groups or []: if not isinstance(group, dict): continue members = [ m for m in (group.get("members") or []) if isinstance(m, dict) and isinstance(m.get("is_indoor"), bool) ] if not members: continue group_id = str(group.get("group_id") or "") if not group_id: continue has_indoor = any(m["is_indoor"] for m in members) has_outdoor = any(not m["is_indoor"] for m in members) mixed = has_indoor and has_outdoor member_data = [ { "loc_id": str(m.get("loc_id") or ""), "label": str(m.get("label") or ""), "summary": str(m.get("summary") or ""), "is_indoor": bool(m["is_indoor"]), } for m in members if str(m.get("loc_id") or "") ] anchor_loc = str(group.get("anchor_loc") or "") for m in members: loc = str(m.get("loc_id") or "") if not loc or loc in out: continue out[loc] = { "group_id": group_id, "is_indoor": bool(m["is_indoor"]), "group_mixed": mixed, "label": str(m.get("label") or ""), "summary": str(m.get("summary") or ""), "anchor_loc": anchor_loc, "group_members": member_data, } return out def compute_aerial_context_hash( *, group_id: str, members: List[Dict[str, Any]], anchor_loc: str, fp_sot: str, ) -> str: """그룹 통합 배치도의 **입력 구조 컨텍스트** 해시 (순수, sha256 16자). Codex 8차 NARROW: 캐시 재사용을 prompt_version 만으로 판정하면 같은 group_id 가 유지된 채 building_groups 멤버/label/summary/is_indoor/ anchor_loc 또는 mixed fp 링크가 바뀌는 resume 에서 낡은 배치도가 부활한다. 이 해시는 구조 데이터만 접는다: group_id + 멤버 리스트 (loc_id/label/summary/is_indoor, classify 산출 순서 보존) + anchor_loc + mixed fp SOT(fp_id — 링크 유무·정체 변화 감지). 소비자는 prompt_version AND context_hash 둘 다 일치할 때만 캐시를 재사용한다. """ import hashlib import json as _json payload = { "group_id": str(group_id or ""), "anchor_loc": str(anchor_loc or ""), "fp_sot": str(fp_sot or ""), "members": [ { "loc_id": str(m.get("loc_id") or ""), "label": str(m.get("label") or ""), "summary": str(m.get("summary") or ""), "is_indoor": bool(m.get("is_indoor")), } for m in (members or []) ], } return hashlib.sha256( _json.dumps(payload, sort_keys=True, ensure_ascii=False) .encode("utf-8") ).hexdigest()[:16] def build_aerial_establishing_prompt( *, members: List[Dict[str, Any]], ) -> str: """같은 장소 그룹의 통합 배치도(site plan) T2I 프롬프트 (generic 템플릿). 실내 floor plan 과 같은 도면 언어의 야외판 — 위에서 수직으로 내려다본 flat 컬러 배치도(실사 사진/3D 아님). ★8차 정정: 그룹(하나의 건물/ 구조물/부지) 전 멤버를 **한 장에 통합** — 건물은 지붕뷰(roof plane, 상층 요소가 그 위에 보임)+주변 지상 공간, 탈것/구조물은 전체 형태. 상업/공공 시설의 간판 판 위치도 표기(도면이므로 빈 판 — 문자 금지). ``members`` 는 classify 산출 데이터({label, summary, is_indoor})로만 주입 — 템플릿 자체는 어떤 시나리오에도 범용(특정 장소 명사 0, 마커/라벨/텍스트 0 강제). """ lines_parts: List[str] = [] for m in members or []: tag = "indoor part" if m.get("is_indoor") else "outdoor part" body = " ".join( x for x in ( str(m.get("label") or "").strip(), str(m.get("summary") or "").strip(), ) if x ) if body: lines_parts.append(f"- ({tag}) {body}") place_block = "\n".join(lines_parts) if lines_parts else ( "- an ordinary real-world outdoor site." ) return "\n".join([ "ONE flat TOP-DOWN SITE PLAN of a single real-world place, drawn " "in the exact drafting style of a clean architectural floor plan: " "flat colour fills, thin dark outlines, simple fixture symbols, " "straight-down orthographic view — NOT a photograph, NOT a 3D " "render, NOT an oblique aerial photo; no perspective, no photo " "textures, no cast shadows.", "", "This ONE place consists of these connected parts — draw them " "TOGETHER as one coherent whole (one building/structure and its " "immediate surroundings), never as separate lots:", place_block, "", "Draw the WHOLE place in one plan: the full outline of the main " "building or structure seen from directly above — its roof plane, " "with whatever stands on that top level drawn in place — plus " "every open ground or floor area around and on it, the entrances, " "stairs and approach paths that connect the parts, and every " "permanently installed element as a simple plan symbol in its " "true position. Indoor parts stay under the roof outline — do NOT " "expose their interior rooms; the roof plane and its top-level " "features represent the building. If the place is a vehicle or " "vessel, draw the ENTIRE craft from above as one complete shape. " "Where the place carries lettering, include those lettering panels " "at their true positions, drawn as blank panels. Keep the geometry " "simple, unambiguous and physically plausible — this plan is the " "place's master spatial reference that later renders must stay " "consistent with.", "", # W-M(9차): 주변 컨텍스트 마진 — 배치도가 부지 안쪽만 그려 장소가 # 진공에 뜨는 것을 방지(조금 넓게). "Frame the plan a little WIDER than the place itself: include a " "modest margin of the immediate surroundings — the adjacent " "street, ground, water or neighbouring structures the place " "actually touches — so the place reads in its real context. The " "place itself stays the dominant subject of the plan.", "", # W-M(9차): 개구부 동그라미 기호 — 실내 fp 와의 정합점이자 후속 # plate 렌더의 문/창 위치 SOT. 기호는 도면 표기 전용(plate 누출 # 가드는 AERIAL_SITE_GUIDANCE 가 담당). "Mark EVERY opening of the place — each doorway, gate, window " "and level-access point (the top of a stair, a ladder, a hatch) " "— with a small hollow CIRCLE symbol placed exactly ON the " "wall, parapet, railing or hull line that opening belongs to. " "An opening symbol must sit on a real physical boundary line — " "never floating in open space — and openings between two parts " "must be shared by both. Doors that lead into an enclosed part " "sit on that part's outline at the side where the entrance " "really is. These hollow opening circles must be the ONLY " "circular marks anywhere on the plan — do not add any other " "round motifs, ring patterns or circular decorations to roofs, " "decks, floors or walls.", "", # W-M(9차): 상층 레벨 시각 단서 — 건물 위 레벨이 지상 단독 부지처럼 # 읽히는 결함(레벨 오독 → plate 전파) 대응. generic 문장 — 판단은 # place 데이터(label/summary)가 이끈다. "If the place's main level sits ON TOP of a building or " "structure (an upper level above the ground), the plan must make " "that unmistakable: draw the supporting building's full roof " "outline as that level's boundary, keep a clearly visible " "roof-edge/parapet line around it, place the level-access " "opening symbol where one arrives onto that level, and let the " "surrounding ground or street read as OUTSIDE and BELOW that " "outline — never draw an upper level as if it were its own " "ground-level lot with a yard of its own.", "", "STRICT: NO people, NO animals, NO text, letters, numbers, " "labels, arrows, logos or watermarks anywhere in the image. The " "ONLY marker symbols allowed are the hollow-circle opening " "markers and the standard flat plan fixture symbols described " "above.", ]) def render_location_aerial( *, openai_client: Any, image_model: str, place_id: str, members: List[Dict[str, Any]], out_path: Path, building_fp_path: Optional[Path] = None, sanitizer: Optional[PromptSanitizer] = None, size: str = "1536x864", quality: str = "high", max_attempts: int = 2, capture_input_image_ids: Optional[List[str]] = None, ) -> Dict[str, Any]: """같은 장소 그룹 1곳의 통합 항공뷰 배치도(site plan) PNG 생성. - ``place_id`` = building_groups 의 group_id(로그/캡처 메타 전용 — 프롬프트에는 넣지 않는다). ``members`` = 그룹 전 멤버의 {label, summary, is_indoor} 데이터. - ``building_fp_path`` 가 존재하면(mixed 그룹) I2I edit(fp ref) + AERIAL_FP_COHERENCE_GUIDANCE — 건물 footprint/개구부를 실내 fp 와 정합 (같은 도면 언어). 없으면 순수 T2I. - moderation 류 차단은 PromptSanitizer 로 bounded retry (render_one_ background 미러). 실패는 raise 하지 않고 status='failed' dict 반환 — 호출자(background_render)는 해당 그룹의 aerial 없이 기존 경로로 fail-safe 진행한다. Returns dict: ``{status, attempts, strategies, building_fp_used, prompt_used, prompt_version, final_block_reason, png_path?}``. """ prompt = build_aerial_establishing_prompt(members=members) fp_attached = building_fp_path is not None and building_fp_path.exists() if fp_attached: prompt = prompt + AERIAL_FP_COHERENCE_GUIDANCE info: Dict[str, Any] = { "status": "failed", "attempts": 0, "strategies": [], "building_fp_used": fp_attached, "prompt_used": prompt, "prompt_version": AERIAL_PROMPT_VERSION, "final_block_reason": None, } current_prompt = prompt if sanitizer is None: sanitizer = PromptSanitizer() for attempt in range(1, max_attempts + 1): info["attempts"] = attempt try: call_kwargs = { "model": image_model, "size": size, "quality": quality, "n": 1, } if fp_attached: reserve_current_call( source="background_render.location_aerial_edit") png = call_gpt_image_bytes( openai_client, mode="edit", prompt=current_prompt, ref_paths=[building_fp_path], call_kwargs=call_kwargs, capture_role="location_aerial", capture_metadata={ "budget_source": "background_render.location_aerial_edit", "place_id": place_id, "building_fp_used": True, }, capture_input_image_ids=capture_input_image_ids, ) else: reserve_current_call( source="background_render.location_aerial_generate") png = call_gpt_image_bytes( openai_client, mode="generate", prompt=current_prompt, ref_paths=None, call_kwargs=call_kwargs, capture_role="location_aerial", capture_metadata={ "budget_source": "background_render.location_aerial_generate", "place_id": place_id, "building_fp_used": False, }, capture_input_image_ids=capture_input_image_ids, ) if not png: raise RuntimeError("empty b64 response") out_path.write_bytes(png) info["status"] = "ok" info["png_path"] = str(out_path) info["prompt_used"] = current_prompt return info except ImageCallBudgetExceeded: raise except Exception as exc: msg = str(exc).lower() is_moderation = any( k in msg for k in ( "moderation", "safety", "content_policy", "prohibited", "policy", "blocked", "violates", "violation", ) ) if is_moderation and attempt < max_attempts: try: sr = sanitizer.sanitize( original_prompt=current_prompt, block_reason="SAFETY", block_categories=[], attempt=min(attempt, 3), ) sanitized = sr.get("sanitized_prompt", "") if sanitized: current_prompt = sanitized info["strategies"].append(sr.get("strategy")) continue except Exception as se: # noqa: BLE001 — 비차단 producer 계약 info["final_block_reason"] = ( f"sanitize_failed: {se}"[:200]) return info logger.error( "location_aerial %s attempt %d/%d: %s", place_id, attempt, max_attempts, str(exc)[:200], ) if attempt >= max_attempts: info["final_block_reason"] = str(exc)[:200] return info return info # ── W-M (2026-07-03): 같은 장소 그룹 plate 생성 순서 — LLM 결정 ── # # 그룹 체인 lane(background_render)은 그룹의 야외 plate 들을 한 줄로 세워 # 순차 생성한다: 첫 plate = 통합 배치도(aerial) 1장만 ref, 이후 plate = # 이전 ok plate + 배치도(최대 2장). 어떤 plate 를 먼저/다음에 생성할지는 # LLM 이 결정한다(사용자 계약: "첫/다음 생성 순서는 LLM 이"). 검증은 # 결정론(순열 + depends_on 제약) — 실패하면 bounded retry 후 구조 순서 # fallback(비차단). PLATE_ORDER_SCHEMA: Dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "order": { "type": "array", "items": {"type": "string"}, "description": ( "bg_id generation order — a permutation of the input list" ), }, "first_reason": { "type": "string", "description": ( "one sentence: why the first bg_id best establishes the " "whole place" ), }, }, "required": ["order", "first_reason"], } PLATE_ORDER_SYSTEM = ( "You order the outdoor background renders of ONE real-world place for " "sequential image generation.\n\n" "The renders are generated one at a time. The FIRST render is generated " "with only the place's top-down site plan as reference, so it must be " "the view that best ESTABLISHES the whole place — the widest, most " "representative view of the main structure and its surroundings. Each " "LATER render is generated with the PREVIOUS successful render as its " "visual reference, so order the rest to maximize visual continuity: " "views that overlap or sit next to each other should be adjacent in " "the order, and wider views should generally come before tighter " "partial views of the same area.\n\n" "Rules:\n" "- Return EVERY bg_id from the input exactly once (a permutation).\n" "- If a render lists depends_on entries, it must come AFTER all of " "them (it depicts a later state of that render).\n" "- Judge only from the given data; do not invent anything about the " "place." ) def validate_plate_order( order: List[str], candidate_ids: List[str], depends_on_by_bg: Dict[str, List[str]], ) -> Optional[str]: """LLM 이 낸 순서의 결정론 검증 (순수). 위반 사유 문자열 / 정상 None. - 순열: candidate_ids 와 정확히 같은 집합, 중복/누락/외부 id 금지. - 의존: ``depends_on_by_bg[bid]`` 의 각 dep(후보 집합 내의 것만)가 bid 보다 먼저 와야 한다. """ cand = list(candidate_ids) if sorted(order) != sorted(cand): return ( f"not a permutation: got={sorted(order)} " f"expected={sorted(cand)}" ) pos = {bid: i for i, bid in enumerate(order)} cand_set = set(cand) for bid, deps in (depends_on_by_bg or {}).items(): if bid not in pos: continue for dep in deps or []: if dep in cand_set and pos.get(dep, -1) > pos[bid]: return f"dependency violated: {bid!r} before its dep {dep!r}" return None def decide_group_plate_order( *, group_id: str, candidates: List[Dict[str, Any]], call_structured_fn: Any, project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 2, ) -> Dict[str, Any]: """같은 장소 그룹 1곳의 plate 생성 순서를 LLM 으로 결정 (비차단). ``candidates`` 항목: ``{bg_id, location_label, location_summary, surface_role, state_label, depends_on, t2i_prompt}`` — t2i_prompt 는 **전문**(자르기 금지 계약). ``depends_on`` 은 후보 집합 내 bg_id 만. 반환: ``{"order", "order_source"('llm'|'fallback_structural'), "first_reason", "attempts", "error"}``. LLM 실패/검증 실패가 ``max_retries`` 를 소진하면 입력 순서 그대로 fallback (구조 순서 = 기존 lane 순서 — 결정론). 예외를 밖으로 던지지 않는다. """ import json as _json candidate_ids = [str(c.get("bg_id") or "") for c in candidates] depends_on_by_bg = { str(c.get("bg_id") or ""): [ str(d) for d in (c.get("depends_on") or []) ] for c in candidates } out: Dict[str, Any] = { "order": list(candidate_ids), "order_source": "fallback_structural", "first_reason": "", "attempts": 0, "error": "", } if len(candidates) < 2: # 1장 이하는 순서 결정이 무의미 — LLM 호출 0. out["order_source"] = "trivial" return out user_prompt = _json.dumps( { "place_renders": candidates, "instruction": ( "Order these renders for sequential chained generation. " "Return every bg_id exactly once." ), }, ensure_ascii=False, indent=2, ) last_err = "" for attempt in range(1, max_retries + 1): out["attempts"] = attempt try: result = call_structured_fn( step="background_plate_order", system_prompt=PLATE_ORDER_SYSTEM, user_prompt=user_prompt, response_schema=PLATE_ORDER_SCHEMA, project_config=project_config, schema_name="background_plate_order", opik_metadata=opik_metadata, ) order = [str(x) for x in (result.get("order") or [])] violation = validate_plate_order( order, candidate_ids, depends_on_by_bg) if violation is not None: raise ValueError(violation) out["order"] = order out["order_source"] = "llm" out["first_reason"] = str(result.get("first_reason") or "") out["error"] = "" return out except Exception as exc: # noqa: BLE001 — 비차단 fallback 계약 last_err = str(exc)[:200] logger.warning( "plate_order group=%s attempt %d/%d failed: %s", group_id, attempt, max_retries, last_err, ) out["error"] = last_err logger.warning( "plate_order group=%s exhausted — 구조 순서 fallback", group_id) return out