"""EntitySyncService — entity_t2i 체크포인트 → EntityCanon + EntityEpisodeLink UPSERT. 기존 `steps.py:_sync_checkpoints_to_db` line 806-905에서 이관. """ from __future__ import annotations import json import logging import uuid from typing import Dict from app.core.errors import AppError from app.services.checkpoint_sync._base import BaseSyncService, is_cp_syncable logger = logging.getLogger(__name__) class EntitySyncService(BaseSyncService): """entity_t2i 체크포인트 기반 canon/link UPSERT. - C/L/P 엔티티만 대상 (outlook은 OutlookSyncService에서 처리). - 기존 link의 t2i_appearance_count 등 메타데이터 보존. - 체크포인트에 없는 link만 제거 (다른 step의 link는 보존). """ def sync_from_checkpoint(self) -> Dict[str, int]: """Returns: {"synced": n, "removed": n, "skipped": n}""" from app.models.project import EntityCanon, EntityEpisodeLink t2i_cp = self._load_cp("entity_t2i") # M2 Fix 2 단일 표준: partial이라도 데이터 있으면 sync. # 빈 데이터(cascade 직후)면 skip — 기존 row 손상 방지. if not is_cp_syncable(t2i_cp, ["characters", "locations", "props"]): logger.info( "Skipping entity sync: entity_t2i 데이터 없음 (cascade or pre-analysis)" ) return {"synced": 0, "removed": 0, "skipped": 1} # Codex P1-2: partial 시 stale 제거 skip — 부분 성공 cp가 실패한 # entity의 기존 link를 삭제하지 않도록 보호. completed cp만 authoritative. is_partial = (t2i_cp.get("status") == "partial") data = t2i_cp.get("data", {}) # Area B (Task 4 / C4): location-only pre-validation 폐기 → sync loop 의 validate_entity_metadata_shape 단일 호출로 통합. # 기존 canon 매핑 (name → id) existing_canons = {e.name: e for e in self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type.in_(["character", "location", "prop"]), ).all()} # UPSERT: 기존 link를 canon_id 기준으로 로드 (메타데이터 보존). # 중요 — C/L/P 캐논만 대상. outlook(O##) link는 OutlookSyncService가 관리. _clp_canon_ids = {e.id for e in existing_canons.values()} _existing_links = { lnk.canon_id: lnk for lnk in self.db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, EntityEpisodeLink.canon_id.in_(_clp_canon_ids) if _clp_canon_ids else False, ).all() } if _clp_canon_ids else {} _seen_canon_ids: set = set() # 벌크 UPSERT — canon은 name 기준 존재 시 update, 없으면 insert _counters = {"C": 0, "L": 0, "P": 0, "O": 0} _prefix_map = {"character": "C", "location": "L", "prop": "P", "outlook": "O"} # 사전 스캔: 기존 canon + 체크포인트 short_id로 카운터 시드 for e in existing_canons.values(): if e.short_id and len(e.short_id) > 1: try: prefix = e.short_id[0] _counters[prefix] = max(_counters.get(prefix, 0), int(e.short_id[1:])) except (ValueError, IndexError): logger.warning("malformed existing short_id skipped: %s", e.short_id) for etype in ["characters", "locations", "props"]: prefix = _prefix_map[etype[:-1]] for ent in data.get(etype, []): sid = ent.get("short_id", "") if sid and len(sid) > 1: try: _counters[prefix] = max(_counters[prefix], int(sid[1:])) except (ValueError, IndexError): logger.warning("malformed checkpoint short_id skipped: %s", sid) # Pre-pass: cp의 short_id와 같은 기존 row의 short_id를 NULL로 임시 비움. # PostgreSQL partial index `uq_entity_canon_short_id WHERE short_id IS NOT NULL`이 # NULL을 unique 검사에서 제외하므로 같은 트랜잭션 내 short_id swap/재할당 안전. # (이 동작은 PostgreSQL 한정 — SQLite 등 partial index 미지원 DB에서는 작동 X.) # # 회귀 가드 (force 모드 sync blocker): # DB: row_A(L18), row_B(L20). cp가 row_A→L20 또는 새 row→L18 등 충돌 매핑 시 # 직접 UPDATE는 UNIQUE VIOLATION → dispatcher abort. # name 매칭이 아닌 short_id 매칭 기준으로 비워야 cp에 없는 row의 짠 short_id가 # cp의 새 short_id와 충돌하는 케이스도 회피. cp_short_ids: set = set() for etype in ["characters", "locations", "props"]: for ent in data.get(etype, []): s = ent.get("short_id", "") if s: cp_short_ids.add(s) orphan_warn_ids: set = set() if cp_short_ids: conflicting = self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type.in_(["character", "location", "prop"]), EntityCanon.short_id.in_(cp_short_ids), ).all() for row in conflicting: # cp에 같은 name이 있으면 pass 2에서 다시 short_id 할당됨 → 정상 swap. # cp에 name이 없으면 NULL로 잔존 → orphan, 후속 short_id lookup 깨질 위험. # name 비교는 pass 2 직전에만 의미. 일단 모두 NULL로 → 후처리에서 경고. orphan_warn_ids.add(row.id) row.short_id = None if conflicting: self.db.flush() # Area B (Task 4 / C4, 2026-05-13): all entity_type normalized shape # DB write. 기존 `if singular == "location"` 분기 폐기 — character/prop # 도 cp metadata_json 그대로 EntityCanon.metadata_json column 에 # serialize. DB write 전 validate_entity_metadata_shape 호출로 L2 # boundary fail-fast (silent fallback / default '{}' 주입 0). from app.core.entity_metadata import validate_entity_metadata_shape # Area B (Task 4 review I1 fix): plural/singular/prefix 통일 — slice fragility # (etype[:-1]) 제거 + future entity_type 추가 시 명시적 매핑 의무. _TYPE_DEFS = ( ("characters", "character", "C"), ("locations", "location", "L"), ("props", "prop", "P"), ) synced = 0 for etype, singular, prefix in _TYPE_DEFS: for ent in data.get(etype, []): name = ent["name"] existing = existing_canons.get(name) # Area B: metadata_json 강제 — None 시 strict raise. # entity_t2i ENTITY_DETAIL_SCHEMA (Task 1 / C1) + entity_t2i # post-validate (Task 3 / C3) 가 normalized shape 강제하므로 # 신 cp 에서는 도달 불가. legacy cp (pre-Area-B) 진입 시 fail-fast # — force re-run entity_t2i 의무. _md_raw = ent.get("metadata_json") if _md_raw is None: raise AppError( code="entity_metadata.shape_violation", message=( f"EntitySyncService.sync_from_checkpoint: entity {name!r} " f"({singular}, short_id={ent.get('short_id', '')!r}) " f"missing metadata_json in entity_t2i checkpoint. " "Force re-run entity_extract / entity_t2i for this episode " "with Area B schema." ), ) # L2 boundary — shape + entity_type 별 invariant 검증. # D6 SpaceProfileError 는 helper 내부에서 그대로 propagate # (D6 error code 보존). validate_entity_metadata_shape( singular, _md_raw, short_id=ent.get("short_id", "") or name, ) metadata_json_str = json.dumps(_md_raw, ensure_ascii=False) if existing: # UPDATE: 기존 canon 갱신 (short_id, description, t2i_prompt, metadata_json) canon_id = existing.id cp_short = ent.get("short_id", "") if cp_short and cp_short != existing.short_id: existing.short_id = cp_short existing.description = ent.get("description", existing.description) existing.t2i_prompt = ent.get("t2i_prompt", existing.t2i_prompt) existing.stable_traits = json.dumps(ent.get("visual_traits", []), ensure_ascii=False) existing.metadata_json = metadata_json_str # Area B: 항상 write. existing.updated_at = self.now else: # INSERT: 새 canon 생성 canon_id = str(uuid.uuid4()) short = ent.get("short_id", "") if not short: _counters[prefix] += 1 short = f"{prefix}{_counters[prefix]:02d}" insert_kwargs = dict( id=canon_id, project_id=self.project_id, short_id=short, name=name, entity_type=singular, description=ent.get("description", ""), t2i_prompt=ent.get("t2i_prompt", ""), stable_traits=json.dumps(ent.get("visual_traits", []), ensure_ascii=False), metadata_json=metadata_json_str, # Area B: 항상 write. created_at=self.now, updated_at=self.now, ) self.db.add(EntityCanon(**insert_kwargs)) # UPSERT: 기존 link 있으면 유지 (t2i_appearance_count 보존), 없으면 생성 if canon_id not in _existing_links: self.db.add(EntityEpisodeLink( id=str(uuid.uuid4()), canon_id=canon_id, project_id=self.project_id, episode_id=self.episode_id, )) _seen_canon_ids.add(canon_id) synced += 1 # 이번 체크포인트에 나타나지 않은 기존 link만 제거 (다른 캐논은 보존). # Codex P1-2: partial 상태일 때는 stale 제거 보류 — 부분 cp는 비포괄적이라 # 미포함 link가 단순 누락일 수 있음. 다음 completed 동기화에서 정리. removed = 0 if is_partial: self.logger.info( "Entity sync: partial checkpoint — upsert %d, stale cleanup deferred " "(awaiting completed checkpoint)", len(_seen_canon_ids), ) else: for cid, lnk in _existing_links.items(): if cid not in _seen_canon_ids: self.db.delete(lnk) removed += 1 self.db.flush() # Orphan 가드: pre-pass에서 NULL set 된 row 중 cp pass 2가 다시 short_id를 # 채우지 않은 row (= cp에 name 매칭 없음)는 후속 short_id lookup이 silent miss. # 데이터 손실 위험 명시 — cleanup은 별도 단계에서. if orphan_warn_ids: still_null = self.db.query(EntityCanon).filter( EntityCanon.id.in_(orphan_warn_ids), EntityCanon.short_id.is_(None), ).all() if still_null: self.logger.warning( "entity_canon orphan after sync — %d row(s) left short_id=NULL " "(cp swap took their short_id, no name match in cp). " "Downstream short_id lookups will silently miss these. ids=%s", len(still_null), [r.id for r in still_null][:5], ) # Orphan 정리 (feedback_never_delete_images 규칙): # entity 분석 step 재실행으로 entity_id 가 바뀌어 image_asset.entity_id # 가 존재하지 않는 entity 를 가리키는 reference/composite 만 삭제. # 평상 UPSERT 흐름에서는 EntityCanon 자체가 보존되므로 0 건이 정상. # 외부 개입 / 프로젝트 재import 등 예외 케이스 대비 방어 net. from app.services.scene_persistence_service import ScenePersistenceService try: sps = ScenePersistenceService(self.db, self.project_id) removed_orphans = sps.delete_orphan_entity_assets() if removed_orphans: self.logger.warning( "entity sync: cleaned %d orphan reference/composite asset row(s) " "(unexpected — investigate parent entity removal source)", removed_orphans, ) except Exception as exc: self.logger.error( "entity sync: orphan cleanup failed (non-fatal): %s", exc ) self.logger.info( "Synced %d entities to DB from checkpoint (upsert); removed %d stale links", synced, removed, ) return {"synced": synced, "removed": removed, "skipped": 0}