"""파이프라인 단계 게이트 — 모든 단계 전환 시 전 단계 100% 완성 검증. 각 단계가 시작되기 전에 이전 단계의 완성도를 확인하고, 미완성이면 구체적인 에러 메시지와 함께 차단한다. """ import json import logging import re from typing import Dict, List, Optional, Tuple from sqlalchemy.orm import Session as OrmSession from app.core.errors import AppError from app.models.project import ( EntityCanon, EntityEpisodeLink, CharacterOutlook, ImageAsset, SceneStill, Episode, ) logger = logging.getLogger(__name__) def episode_reference_entities( db: OrmSession, project_id: str, episode_id: str, ) -> List[EntityCanon]: """이 에피소드가 **참조 이미지를 써야 하는** character/prop canon. 대상 범위는 `EntityEpisodeLink` — 「이번 에피소드에 나오는가」다. location 은 제외 (배경 참조 이미지 생성이 꺼져 있다). """ # ★★보류(`shelved`) 링크는 대상이 아니다 (2026-09-04). 저빈도로 걸러진 # 요소는 이제 **지우는 대신** 링크에 그렇게 적어 남긴다. 여기서 안 빼면 # 그 행들이 참조 대상으로 세어져 게이트가 영영 안 풀리고, 만들려 들면 # 유료 호출이 는다. # ★조건을 여기 다시 적지 않는다 — 정본 하나를 부른다. from app.core.entity_identity import active_episode_canon_ids linked_ids = active_episode_canon_ids(db, project_id, episode_id) if not linked_ids: return [] # ★생산자와 **같은 정본**에서 갈래를 받는다 — `canonical_ref_owner_types()` # (집행 계약의 정책 갈래). 종전의 `notin_(["location", "outlook"])` 는 # location_part 를 대상에 넣어 sidecar 와 **집행자가 둘**이 됐다 # (Codex BLOCK 2026-09-02). 새 갈래가 생기면 계약이 먼저 선다. from app.modules.pipeline.grounding_entity_contract import ( canonical_ref_owner_types, ) return db.query(EntityCanon).filter( EntityCanon.id.in_(linked_ids), EntityCanon.entity_type.in_(list(canonical_ref_owner_types())), ).all() def _primary_reference_rows(db: OrmSession, project_id: str, entity_ids): """정본 참조 행 — **필터 정의는 여기 한 곳뿐이다.** 파일 존재는 보지 않는다(호출자가 본다). 행 수와 파일 수를 갈라 봐야 「행이 없다」와 「파일이 사라졌다」를 구별할 수 있다. """ return db.query(ImageAsset).filter( ImageAsset.project_id == project_id, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ImageAsset.entity_id.in_(list(entity_ids or [])), ).all() def reference_asset_row_count(db: OrmSession, project_id: str, entity_ids) -> int: """정본 참조 **행** 수 — 진단용(파일 존재 무관).""" ids = list(entity_ids or []) return len(_primary_reference_rows(db, project_id, ids)) if ids else 0 def entities_with_reference( db: OrmSession, project_id: str, entity_ids, ) -> set: """정본 참조 이미지가 **실제로 있는** canon id 집합. ★자산 범위는 **canon/project** 다 — episode 가 아니다. 참조 이미지의 primary 는 프로젝트 안에서 canon 당 **하나**이고, 새 참조를 쓸 때 `reference_phase1_service.py:198-202` 가 `project_id + entity_id` 범위로 옛 primary 를 전부 내린다. 같은 함수의 합성(composite) 게이트도 이미 project 범위로 본다. episode 범위로 물으면 **앞 에피소드에서 만든 canon 을 다시 쓸 때** 교착이 난다 — 생성 쪽(`reference_pipeline_orchestrator.py:202-208`)은 project 범위로 「이미 있다」고 건너뛰는데 게이트는 이 에피소드 행을 요구해서, 다시 태워도 영영 안 풀린다 (2026-08-29 실측: 에피소드 3화의 C01·P03). ★DB 행만으로는 안 된다 — **파일이 실제로 있어야** 한다. """ ids = list(entity_ids or []) if not ids: return set() from app.core.file_paths import resolve_image_path found = set() for ra in _primary_reference_rows(db, project_id, ids): p = resolve_image_path(ra.file_path) if p and p.exists(): found.add(ra.entity_id) else: logger.warning("참조 이미지 파일 누락: entity=%s, path=%s", ra.entity_id, ra.file_path) return found def reference_targets( db: OrmSession, project_id: str, episode_id: str, ) -> List[EntityCanon]: """참조가 **실제로 있어야 하는** 대상 — 저빈도 스킵을 뺀 뒤. ★게이트와 진행률이 이 하나를 같이 써야 한다. 종전에는 게이트만 저빈도 스킵을 빼고 진행률은 안 빼서, 「누락 없음」인데 화면은 `3/4 미완성` 으로 나왔다 (2026-08-29 실측, 에피소드 2화). """ from app.core.low_freq_skip import load_low_freq_skip_ids skipped = load_low_freq_skip_ids(project_id, episode_id) return [e for e in episode_reference_entities(db, project_id, episode_id) if e.id not in skipped] def missing_reference_entities( db: OrmSession, project_id: str, episode_id: str, ) -> List[EntityCanon]: """준비가 안 된 대상 — 게이트와 진행률이 **같은 물음**을 쓰게 하는 정본. 「대상은 episode, 자산은 canon/project」. """ targets = reference_targets(db, project_id, episode_id) if not targets: return [] have = entities_with_reference(db, project_id, [e.id for e in targets]) return [e for e in targets if e.id not in have] def check_analysis_ready(db: OrmSession, project_id: str, episode_id: str) -> None: """분석 시작 전 검증: fulltext 존재.""" ep = db.query(Episode).filter(Episode.id == episode_id, Episode.project_id == project_id).first() if not ep: raise AppError(code="episode.not_found", message="에피소드를 찾을 수 없습니다", status_code=404) if not ep.fulltext: raise AppError(code="gate.no_fulltext", message="시나리오 텍스트가 없습니다. PDF를 먼저 업로드하세요.", status_code=400) def ensure_analysis_projection_current( db: OrmSession, project_id: str, episode_id: str, ) -> None: """W20F10 — readiness gate 진입부에서 checkpoint → DB projection 을 최신화. image/ref readiness gate 가 stale ``Episode.status`` 만 신뢰해 막히는 회귀 방지용. ``orchestrate_full_sync`` 는 idempotent — step 경로가 이미 sync 한 경우에도 안전하게 재실행된다. API/script 경로처럼 sync 없이 gate 만 통과하는 호출자가 stale ``error`` projection 에 갇히는 것을 막는다. 실패 시 fail-closed: 주 세션 rollback 후 ``gate.projection_sync_failed`` AppError 로 상위 호출자에 보고. orchestrator 자체가 별도 세션에 sync 실패를 이미 기록하므로 여기서는 호출 세션 정리만 책임진다. """ try: from app.services.checkpoint_sync import orchestrate_full_sync orchestrate_full_sync(project_id, episode_id, db) except AppError: raise except Exception as exc: try: db.rollback() except Exception: logger.warning("rollback after sync failure failed", exc_info=True) raise AppError( code="gate.projection_sync_failed", message=f"분석 projection 동기화 실패: {exc}", status_code=500, ) from exc def check_ref_images_ready(db: OrmSession, project_id: str, episode_id: str) -> None: """참조 이미지 생성 전 검증: 분석 완료 + 요소 존재.""" ep = db.query(Episode).filter(Episode.id == episode_id, Episode.project_id == project_id).first() if not ep: raise AppError(code="episode.not_found", message="에피소드를 찾을 수 없습니다", status_code=404) # W20F10: stale Episode.status 로 막히지 않도록 checkpoint projection 을 # 동기화한 뒤 ep 를 refresh 해서 status 를 다시 본다. ensure_analysis_projection_current(db, project_id, episode_id) db.refresh(ep) if ep.status != "analyzed": raise AppError( code="gate.analysis_incomplete", message=f"분석이 완료되지 않았습니다 (현재: {ep.status}). 분석을 먼저 완료하세요.", status_code=400, ) # ★보류(shelved)는 세지 않는다 — 「연결된 요소가 없다」의 뜻은 # 「만들 것이 없다」이지 「행이 없다」가 아니다 (2026-09-04). from app.core.entity_identity import active_episode_canon_ids entity_count = len(active_episode_canon_ids(db, project_id, episode_id)) if entity_count == 0: raise AppError(code="gate.no_entities", message="이 에피소드에 연결된 요소가 없습니다. 분석을 먼저 실행하세요.", status_code=400) def check_scene_images_ready(db: OrmSession, project_id: str, episode_id: str) -> Dict: """씬 이미지 생성 전 검증: 분석 + 참조 이미지 + 합성 이미지 100% 완성. Returns: 검증 결과 요약 dict (UI 표시용) """ ep = db.query(Episode).filter(Episode.id == episode_id, Episode.project_id == project_id).first() if not ep: raise AppError(code="episode.not_found", message="에피소드를 찾을 수 없습니다", status_code=404) # W20F10: stale Episode.status 로 막히지 않도록 checkpoint projection 을 # 동기화한 뒤 ep 를 refresh 해서 status 를 다시 본다. ensure_analysis_projection_current(db, project_id, episode_id) db.refresh(ep) # 1) 분석 완료 if ep.status != "analyzed": raise AppError( code="gate.analysis_incomplete", message=f"분석이 완료되지 않았습니다 (현재: {ep.status}). 분석을 먼저 완료하세요.", status_code=400, ) # 2) 씬 존재 scene_count = db.query(SceneStill).filter( SceneStill.project_id == project_id, SceneStill.episode_id == episode_id, ).count() if scene_count == 0: raise AppError(code="gate.no_scenes", message="씬이 없습니다. 분석을 먼저 완료하세요.", status_code=400) # 3) 캐릭터/소품 참조 이미지 전수 확인 — **대상은 episode, 자산은 canon/project** # (정본 helper 하나로 게이트와 진행률이 같은 물음을 쓴다) non_outlook = reference_targets(db, project_id, episode_id) missing_refs = missing_reference_entities(db, project_id, episode_id) if missing_refs: names = ", ".join(f"{e.name}({e.entity_type})" for e in missing_refs[:5]) more = f" 외 {len(missing_refs)-5}개" if len(missing_refs) > 5 else "" raise AppError( code="gate.incomplete_references", message=f"참조 이미지 미완성: {len(missing_refs)}개 누락 ({names}{more}). '참조 이미지 생성'을 먼저 완료하세요.", status_code=400, ) # 4) 인물+아웃룩 합성 이미지 전수 확인 (이 에피소드에 연결된 캐릭터만) # O00 (Null Outlook) 쌍은 composite 불필요 → 제외 char_ids_in_episode = {e.id for e in non_outlook if e.entity_type == "character"} # ★이 화의 배정만 (Codex BLOCK 2026-09-04). 상태 조회와 **같은 함수**를 쓴다. all_combos = episode_composite_combos( db, project_id, episode_id, char_ids_in_episode) if all_combos: existing_keys = existing_composite_keys(db, project_id) # 배치 쿼리 — 누락 combo의 entity 이름을 한 번에 로드 (N+1 방지) # ★저빈도 스킵 조건을 여기서 또 걸지 않는다 — `char_ids_in_episode` 가 # `reference_targets` 에서 왔고 그쪽이 이미 뺐다. 두 번 빼면 같은 규칙이 # 두 자리에 살아 한쪽만 바뀌는 날 어긋난다. _missing_pairs = [ combo for combo in all_combos if f"{combo.character_id}:{combo.outlook_id}" not in existing_keys ] missing_combos: List[str] = [] if _missing_pairs: _needed_ids: set = set() for combo in _missing_pairs: _needed_ids.add(combo.character_id) _needed_ids.add(combo.outlook_id) _name_map: Dict[str, str] = { e.id: e.name for e in db.query(EntityCanon).filter(EntityCanon.id.in_(_needed_ids)).all() } for combo in _missing_pairs: missing_combos.append( f"{_name_map.get(combo.character_id, '?')}+{_name_map.get(combo.outlook_id, '?')}" ) if missing_combos: names = ", ".join(missing_combos[:5]) more = f" 외 {len(missing_combos)-5}개" if len(missing_combos) > 5 else "" raise AppError( code="gate.incomplete_composites", message=f"인물+아웃룩 합성 이미지 미완성: {len(missing_combos)}개 누락 ({names}{more}). '참조 이미지 생성'을 먼저 완료하세요.", status_code=400, ) return { # 여기까지 왔으면 누락 게이트를 통과한 자리다 — 대상이 곧 준비된 수다. "entities": len(non_outlook), "refs": len(non_outlook), "combos": len(all_combos), "combo_done": len(all_combos) - len(missing_combos) if all_combos else 0, "scenes": scene_count, } def episode_composite_combos(db: OrmSession, project_id: str, episode_id: str, char_ids: set) -> List: """이 화에서 **합성이 있어야 하는** 인물↔아웃룩 쌍. ★게이트와 상태 조회가 **같은 함수**를 봐야 한다 (Codex BLOCK 2026-09-04). 따로 적으면 한쪽만 화 범위로 바뀌고, EP1 합성만 있어도 EP2 가 「다 됐다」로 보인다. `O00`(Null Outlook)은 합성이 필요 없으니 뺀다. """ if not char_ids: return [] from app.core.entity_identity import episode_outlook_rows o00 = { e.id for e in db.query(EntityCanon).filter( EntityCanon.project_id == project_id, EntityCanon.short_id == "O00", EntityCanon.entity_type == "outlook", ).all() } return [c for c in episode_outlook_rows(db, project_id, episode_id) if c.character_id in char_ids and c.outlook_id not in o00] def existing_composite_keys(db: OrmSession, project_id: str) -> set: """실제로 **파일까지 있는** 합성의 ``"{char_id}:{outlook_id}"`` 집합. ★DB 행만 세면 파일이 없어진 것을 「됐다」로 읽는다. """ from sqlalchemy import or_ from app.core.file_paths import resolve_image_path keys = set() for ca in db.query(ImageAsset).filter( ImageAsset.project_id == project_id, ImageAsset.asset_type == "reference", or_(ImageAsset.prompt_used.like("%composite:%"), ImageAsset.prompt_used.like("%outlook_id:%")), ).all(): pu = ca.prompt_used or "" # 새 형식: composite:{char}:{outlook} · 레거시: outlook_id:{outlook} m = re.search(r'composite:([a-f0-9-]+):([a-f0-9-]+)', pu) m_legacy = re.search(r'outlook_id:([a-f0-9-]+)', pu) if not m else None char_id = m.group(1) if m else (ca.entity_id if m_legacy else None) outlook_id = m.group(2) if m else (m_legacy.group(1) if m_legacy else None) if not (char_id and outlook_id): continue resolved = resolve_image_path(ca.file_path) if resolved and resolved.exists(): keys.add(f"{char_id}:{outlook_id}") else: logger.warning("합성 이미지 파일 누락: entity=%s, path=%s", ca.entity_id, ca.file_path) return keys def get_pipeline_status(db: OrmSession, project_id: str, episode_id: str) -> Dict: """전체 파이프라인 단계별 완성도 조회 (UI 표시용).""" ep = db.query(Episode).filter(Episode.id == episode_id, Episode.project_id == project_id).first() if not ep: return {} # 요소 — ★게이트와 **같은 물음**(보류 제외). 한쪽만 빼면 「누락 없음」인데 # 화면은 미완성으로 나온다 (2026-08-29 에 그 판이 있었다). from app.core.entity_identity import active_episode_canon_ids linked_ids = active_episode_canon_ids(db, project_id, episode_id) entity_total = db.query(EntityCanon).filter(EntityCanon.id.in_(linked_ids)).count() if linked_ids else 0 # 참조 이미지 — ★게이트와 **같은 물음**을 쓴다. 종전에는 이 수가 # ①이 에피소드 범위이고 ②linked 대상으로 좁히지도 않아서, 게이트는 # 막는데 화면은 「다 됐다」로 보이는 오독이 가능했다 (2026-08-29). _ref_targets = reference_targets(db, project_id, episode_id) _ref_missing = missing_reference_entities(db, project_id, episode_id) ref_done = len(_ref_targets) - len(_ref_missing) # 합성 이미지 — ★게이트와 **같은 물음**(이 화의 쌍, O00 제외, 파일까지 확인). # 종전에는 분모도 분자도 **프로젝트 전체**라, EP1 합성만 있어도 EP2 가 # 「다 됐다」로 보이고 다른 화의 쌍이 EP2 분모에 들어갔다 # (Codex BLOCK 2026-09-04). # ★인물 집합은 `linked_ids` 를 다시 조회하지 않고 **게이트가 쓰는 것과 # 같은 `_ref_targets`** 에서 뽑는다 (Codex NON-BLOCK 2026-09-04) — # 그래야 분모가 `check_reference_images_ready` 와 문자 그대로 같고, # 저빈도 스킵 규칙이 한 자리에만 산다. _chars_in_episode = { e.id for e in _ref_targets if e.entity_type == "character"} _combos_status = episode_composite_combos( db, project_id, episode_id, _chars_in_episode) combo_total = len(_combos_status) if combo_total: _keys_status = existing_composite_keys(db, project_id) composite_done = sum( 1 for c in _combos_status if f"{c.character_id}:{c.outlook_id}" in _keys_status) else: composite_done = 0 # 씬 scene_total = db.query(SceneStill).filter( SceneStill.project_id == project_id, SceneStill.episode_id == episode_id, ).count() scene_img_done = db.query(ImageAsset.still_id).filter( ImageAsset.project_id == project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ).distinct().count() # 단계별 완성 여부 analysis_done = ep.status == "analyzed" # ★분모도 게이트와 같은 대상 집합이다. 종전 분모는 location 을 포함했는데 # 게이트 대상은 character/prop 뿐이라, 장소가 있는 에피소드는 화면이 # 영영 「미완성」으로 남았다. ref_target_total = len(_ref_targets) # ★게이트가 통과시키는 조건과 **같은 식**이다 — 누락이 없으면 완료다. # ★대상이 0 이어도 완료다. 게이트는 누락 0 이면 통과시키는데 여기서만 # `> 0` 을 걸면, 장소만 있는 에피소드나 char/prop 이 전부 저빈도 스킵인 # 에피소드에서 게이트는 지나가고 화면은 영영 미완성으로 남는다 # (2026-08-29 Codex BLOCK-3). 이것은 준비됨(readiness) 계약이다. ref_complete = not _ref_missing combo_complete = composite_done >= combo_total if combo_total > 0 else ref_complete scene_complete = scene_img_done >= scene_total and scene_total > 0 return { "analysis": {"status": ep.status, "done": analysis_done}, "reference_images": { "entity_total": ref_target_total, "ref_done": ref_done, "complete": ref_complete, "can_start": analysis_done, }, "composite_images": { "combo_total": combo_total, "composite_done": composite_done, "complete": combo_complete, "can_start": ref_complete, }, "scene_images": { "scene_total": scene_total, "scene_done": scene_img_done, "complete": scene_complete, "can_start": ref_complete and combo_complete, }, }