"""OutlookSyncService — outlook_phase3 체크포인트 → EntityCanon(outlook) + CharacterOutlook. Phase 4.5: CharacterOutlook의 "해당 에피소드 character에 연결된 행 전체 DELETE 후 재INSERT" 경로를 `(character_id, outlook_id)` 키 기준 delta sync로 전환. 안정성 향상 + DELETE→INSERT 패턴 회피. """ from __future__ import annotations import uuid from typing import Dict, Set, Tuple from sqlalchemy import text as sql_text from app.core.entity_identity import (CANON_STATUS_ORPHANED, NULL_OUTLOOK_SHORT_ID) from app.core.name_matcher import build_name_index, lookup_name from app.modules.pipeline import episode_carry as _ec from app.services.checkpoint_sync._base import BaseSyncService, is_cp_syncable class OutlookSyncService(BaseSyncService): def sync_from_checkpoint(self) -> Dict[str, int]: """outlook_phase3 (또는 legacy outlook_extraction) → outlook EntityCanon + CharacterOutlook. Phase 4.5: CharacterOutlook는 (character_id, outlook_id) 기준 delta sync. Returns: { "outlooks": n_canon, "links": n_current_pairs, "links_inserted": n, "links_deleted": n, "orphans_marked": n, } """ from app.models.project import EntityCanon, EntityEpisodeLink # M2 Fix 2 단일 표준: partial이라도 데이터 있으면 sync. # 빈 데이터(cascade 직후)면 phase3 skip → legacy outlook_extraction fallback. ol_cp = self._load_cp("outlook_phase3") if not is_cp_syncable(ol_cp, ["outlooks"]): ol_cp = self._load_cp("outlook_extraction") outlook_count = 0 link_count = 0 links_inserted = 0 links_deleted = 0 # Codex P1-2: partial 시 stale 제거 skip — 부분 cp가 미포함 outlook을 # 단순 누락으로 처리해 옛 row를 삭제하지 않도록. completed만 authoritative. is_partial = (ol_cp.get("status") == "partial") if ol_cp else False if is_cp_syncable(ol_cp, ["outlooks"]): ol_data = ol_cp.get("data", {}) outlooks = ol_data.get("outlooks", []) # ★★★새 계약(앞 화 명부)을 탄 CP 인가 — 그러면 **ID 만** 쓴다. # 이름 fallback 은 프로젝트 전역에서 **다른 사람의 같은 이름 옷**을 # 한 canon 으로 접는다. carry 가 같은 ID 경쟁을 둘 다 NEW 로 # 돌려도 여기서 이름으로 다시 합쳐지면 소용이 없다 # (Codex BLOCK 2026-09-04). # ★표식이 없는 옛 CP 에서는 종전대로 이름을 본다 — 그때는 # `short_id` 가 화마다 다시 매겨져 ID 만으로는 못 잇는다. _id_only = _ec.LEDGER_KEY in ol_data _existing_ols = self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "outlook", ).all() existing_ol_by_short = {e.short_id: e for e in _existing_ols if e.short_id} existing_ol_by_name = {e.name: e for e in _existing_ols} # Pre-pass: cp의 outlook short_id와 같은 기존 row를 NULL로 임시 비움. # PostgreSQL partial index `uq_entity_canon_short_id WHERE short_id IS NOT NULL`이 # NULL을 unique 검사에서 제외 → 같은 트랜잭션 내 swap/재할당 안전 (PG 한정). # # NOTE — outlook swap intent 한계: line 77 `existing_ol_by_short.get(ol_short)`가 # short_id 우선 매칭이고, dict는 pre-pass 이전(line 51) 빌드라 옛 short_id로 # 매핑된다. 즉 cp가 row_A↔row_B의 short_id를 swap 의도했더라도 코드는 옛 매핑 # 기준으로 처리해 name swap만 수행 (short_id는 그대로). 본 hotfix는 force # 시 UNIQUE VIOLATION 회피만 보장 — outlook swap 정확성은 별도 후속 작업. # NULL 후 cp 처리에서 다시 채워지지 않으면 NULL 잔존. ★2026-09-04: # 종전 주석은 `_remove_orphan_outlooks`가 그 행을 「자동 삭제」한다고 # 적혀 있었는데, 그 삭제가 **다른 화의 아웃룩까지 지우던 자리**였다. # 지금은 `_mark_orphan_outlooks`가 표시만 한다. cp_short_ids: set = set() for ol in outlooks: s = ol.get("outlook_id") or ol.get("short_id", "") if s: cp_short_ids.add(s) if cp_short_ids: conflicting = self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "outlook", EntityCanon.short_id.in_(cp_short_ids), ).all() for row in conflicting: row.short_id = None if conflicting: self.db.flush() # 이 에피소드에 이미 달린 outlook link — 중복 INSERT 방지용. _linked_canon_ids: Set[str] = { row[0] for row in self.db.query(EntityEpisodeLink.canon_id).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, ).all() } for ol in outlooks: ol_name = ol.get("name", "") ol_short = ol.get("outlook_id") or ol.get("short_id", "") ol_desc = ol.get("description", "") existing = existing_ol_by_short.get(ol_short) if existing is None and not _id_only: existing = existing_ol_by_name.get(ol_name) if existing: ol_id = existing.id existing.short_id = ol_short existing.name = ol_name existing.description = ol_desc existing.updated_at = self.now else: ol_id = str(uuid.uuid4()) self.db.add(EntityCanon( id=ol_id, project_id=self.project_id, short_id=ol_short, name=ol_name, entity_type="outlook", description=ol_desc, created_at=self.now, updated_at=self.now, )) # UPSERT: 프로젝트에 이미 있던 canon 이라도 **이 에피소드** link 는 # 따로 필요하다. 예전에는 이 줄이 `else`(새 canon) 안에만 있어, # 재사용된 outlook 이 episode link 없이 남았다. 그러면 # `load_episode_entity_dicts` 가 그 canon 을 안 싣고, # `still_recipe_service._norm_uuid('O01')` 이 None 이 되어 # outfit_assignments 가 `unknown outlook_id` 로 fail-closed 된다. # entity_sync_service 의 같은 자리(분기 밖 upsert)와 모양을 맞춘다. if ol_id not in _linked_canon_ids: self.db.add(EntityEpisodeLink( id=str(uuid.uuid4()), canon_id=ol_id, project_id=self.project_id, episode_id=self.episode_id, )) _linked_canon_ids.add(ol_id) outlook_count += 1 # 체크포인트에 없는 기존 outlook 삭제 — Codex P1-2: partial 시 보류. if is_partial: self.logger.info( "Outlook sync: partial checkpoint — upsert %d, stale outlook canon " "cleanup deferred (awaiting completed checkpoint)", outlook_count, ) else: # ★★★범위는 **이 에피소드의 링크**다 — 프로젝트 전체가 아니다. # # 종전에는 `_existing_ols`(프로젝트 전체 아웃룩)를 훑어 이번 화 # 체크포인트에 없는 것을 canon 째 DELETE 했다. 그래서 3화를 # sync 하면 **1화의 아웃룩이 사라졌다** — 실측(골목 끝 # da049582): 1화가 만든 O01 남색작업복·O02 노란우비·O03 회색외투 # 중 O03 은 DB 에서 없어지고, DB 의 아웃룩 집합이 「마지막에 # sync 한 화의 집합」과 한 글자도 안 달랐다. # # ★C/L/P 쪽 `EntitySyncService` 는 처음부터 **링크만** 지운다. # 여기만 canon 을 지우고 있었다. 같은 계약으로 맞춘다. _matched_ids = set() for ol in outlooks: _s = ol.get("outlook_id") or ol.get("short_id", "") _n = ol.get("name", "") _e = existing_ol_by_short.get(_s) if _e is None and not _id_only: _e = existing_ol_by_name.get(_n) if _e: _matched_ids.add(_e.id) # 이 에피소드에 붙어 있는 아웃룩 링크 중 이번 CP 에 없는 것만. _stale_links = self.db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, EntityEpisodeLink.canon_id.in_( [e.id for e in _existing_ols] or [""]), ).all() _unlinked = 0 for lnk in _stale_links: if lnk.canon_id in _matched_ids: continue # ★링크는 **이 에피소드 자신의 기록**이라 지워도 남의 것을 # 안 건드린다. canon·alias·relation·character_outlook 은 # 그대로 둔다 — 다른 화가 쓰고 있을 수 있다. self.db.delete(lnk) _unlinked += 1 if _unlinked: self.logger.info( "Outlook sync: 이번 화 링크 %d개 해제 (canon 은 보존)", _unlinked, ) self.db.flush() # CharacterOutlook delta sync — (character_id, outlook_id) 기준 # char_name_id는 _delta_sync_character_outlook 내부에서 local 구축 (out-param 제거 — v0.5.17 Claude 리뷰) # Codex P1-2: partial이면 stale pair 삭제 skip (insert만). ( link_count, links_inserted, links_deleted, ) = self._delta_sync_character_outlook( outlooks, ol_data, skip_stale=is_partial, id_only=_id_only ) self.logger.info( "Synced %d outlooks + %d character-outlook links (Δ +%d / -%d)", outlook_count, link_count, links_inserted, links_deleted, ) # 미연결 아웃룩 제거 (O00 null_outlook은 보호) — Codex P1-2: partial 시 보류. if is_partial: self.logger.info( "Outlook sync: partial checkpoint — orphan outlook cleanup deferred" ) orphans_removed = 0 else: orphans_removed = self._mark_orphan_outlooks() return { "outlooks": outlook_count, "links": link_count, "links_inserted": links_inserted, "links_deleted": links_deleted, "orphans_marked": orphans_removed, } def _delta_sync_character_outlook( self, outlooks: list, ol_data: dict, skip_stale: bool = False, id_only: bool = False, ) -> Tuple[int, int, int]: """character_outlook을 (character_id, outlook_id) 쌍 기준 delta sync. 기대 쌍을 scene_assignments에서 파싱 → 현재 DB 쌍과 diff → 추가 INSERT / 제거 DELETE. UPSERT는 PK(id)가 임의 uuid이므로 불필요. **name_matcher**: outlook/character 모두 원본 + 괄호·공백 정규화 키로 인덱스 구축. out-param을 쓰지 않고 local에서만 빌드해 caller dict 오염 없음. Args: skip_stale: True면 missing 쌍의 DELETE skip (Codex P1-2 — partial 시 부분 cp가 옛 link 정상 누락하지 않도록 보호). """ from app.models.project import EntityCanon # outlook → id (최신 상태 재조회) ol_entries = self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "outlook", ).all() ol_short_to_id = {e.short_id: e.id for e in ol_entries if e.short_id} ol_name_to_id = build_name_index( ol_entries, key_fn=lambda e: e.name, value_fn=lambda e: e.id, ) # character → id (최신 상태 재조회, local only) char_entries = self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "character", ).all() char_short_to_id: Dict[str, str] = { e.short_id: e.id for e in char_entries if e.short_id } char_name_id = build_name_index( char_entries, key_fn=lambda e: e.name, value_fn=lambda e: e.id, ) # 기대 상태: (char_id, outlook_id) 집합 desired_pairs: Set[Tuple[str, str]] = set() for sa in ol_data.get("scene_assignments", []): chars = sa.get("assignments", sa.get("characters", [])) for c in chars: csid_raw = c.get("character_id", "") csid = ( csid_raw.split("O")[0] if "O" in csid_raw and csid_raw.startswith("C") else csid_raw ) cname = c.get("character_name", "") osid = c.get("outlook_id", "") oname = c.get("outlook_name", "") # ★새 계약 CP 는 **ID 만**. 이름으로 되짚으면 같은 이름의 다른 # 옷이 한 canon 으로 접힌다 (Codex BLOCK 2026-09-04). cid = char_short_to_id.get(csid) oid = ol_short_to_id.get(osid) if not id_only: cid = cid or lookup_name(char_name_id, cname) oid = oid or lookup_name(ol_name_to_id, oname) if cid and oid: desired_pairs.add((cid, oid)) # ★★★현재 상태의 범위는 **이 화의 배정**이다 (alembic 014). # # 종전 조회는 「이 화에 나오는 인물」로만 좁히고 그 인물의 **프로젝트 # 전체 옷 배정**을 가져왔다. 그래서 같은 인물이 다음 화에 다른 옷을 # 입으면 `existing - desired` 에 **앞 화 배정**이 걸려 지워졌다. # `character_outlook` 에 화 칸이 없어서 갈라낼 방법이 없었다. existing_rows = self.db.execute( sql_text( "SELECT co.id, co.character_id, co.outlook_id " "FROM character_outlook co " "WHERE co.project_id = :pid AND co.episode_id = :eid" ), {"pid": self.project_id, "eid": self.episode_id}, ).fetchall() existing_map: Dict[Tuple[str, str], str] = {} for row in existing_rows: row_id, cid, oid = row[0], row[1], row[2] key = (cid, oid) # 혹시라도 동일 키 중복 행이 있으면 최신 1개만 유지, 나머지 정리 (dedup) if key in existing_map: self.db.execute( sql_text("DELETE FROM character_outlook WHERE id = :id"), {"id": row_id}, ) else: existing_map[key] = row_id existing_pairs = set(existing_map.keys()) # ★화 칸이 없는 legacy 행 — **지우지 않고 이어받는다.** 이번 화가 # 원하는 쌍과 같으면 그 행에 화를 적어 제 것으로 삼는다. 그러지 않으면 # 같은 쌍이 두 행(NULL + 이 화)으로 늘어난다. legacy_rows = self.db.execute( sql_text( "SELECT co.id, co.character_id, co.outlook_id " "FROM character_outlook co " "WHERE co.project_id = :pid AND co.episode_id IS NULL" ), {"pid": self.project_id}, ).fetchall() legacy_map: Dict[Tuple[str, str], str] = {} for row in legacy_rows: legacy_map.setdefault((row[1], row[2]), row[0]) to_insert = desired_pairs - existing_pairs claimed = 0 for key in sorted(to_insert & set(legacy_map)): self.db.execute( sql_text("UPDATE character_outlook SET episode_id = :eid " "WHERE id = :id"), {"eid": self.episode_id, "id": legacy_map[key]}, ) existing_map[key] = legacy_map[key] claimed += 1 to_insert = to_insert - set(legacy_map) # Codex P1-2: partial 시 stale pair 삭제 skip — 부분 cp의 missing이 # 단순 누락일 수 있음 (다음 completed 동기화에서 정리). # ★지우는 것은 **이 화가 적은 행**뿐이다. 다른 화 것도, legacy 도 아니다. to_delete = set() if skip_stale else (existing_pairs - desired_pairs) for key in to_delete: self.db.execute( sql_text("DELETE FROM character_outlook WHERE id = :id"), {"id": existing_map[key]}, ) for (cid, oid) in to_insert: self.db.execute( sql_text( "INSERT INTO character_outlook " "(id, project_id, episode_id, character_id, outlook_id, created_at) " "VALUES (:id, :pid, :eid, :cid, :oid, :now)" ), { "id": str(uuid.uuid4()), "pid": self.project_id, "eid": self.episode_id, "cid": cid, "oid": oid, "now": self.now, }, ) if to_insert or to_delete or claimed: self.db.flush() if claimed: self.logger.info( "character_outlook: 화 칸 없는 legacy 행 %d개를 이 화로 이어받음", claimed, ) return len(desired_pairs), len(to_insert), len(to_delete) def _mark_orphan_outlooks(self) -> int: """어디에도 안 붙은 아웃룩을 **표시만** 한다. ★지우지 않는다. ★★★종전 이름은 `_remove_orphan_outlooks` 였고 실제로 `entity_canon` 까지 DELETE 했다. 이 파이프라인에서 아웃룩 canon 은 **프로젝트 자산** 이라 한 화의 sync 가 지울 것이 아니다. 실측(골목 끝 da049582)에서 1화 아웃룩 셋 중 하나가 이 경로로 사라졌다. ★★참조를 **넷 다** 본다 (Codex 2026-09-04). `character_outlook` 만 보면 「관계는 없는데 링크·관계사실·이미지가 붙어 있는」 행을 고아로 잘못 읽는다. 그런 행을 지우던 것이 앞 판의 손실이었다. """ rows = self.db.execute(sql_text( "SELECT ec.id, ec.name, ec.short_id, ec.status FROM entity_canon ec " "WHERE ec.project_id = :pid AND ec.entity_type = 'outlook' " " AND ec.short_id IS DISTINCT FROM :null_outlook " " AND NOT EXISTS (SELECT 1 FROM character_outlook co " " WHERE co.outlook_id = ec.id) " " AND NOT EXISTS (SELECT 1 FROM entity_episode_link el " " WHERE el.canon_id = ec.id) " " AND NOT EXISTS (SELECT 1 FROM relation_participant rp " " WHERE rp.canon_id = ec.id) " " AND NOT EXISTS (SELECT 1 FROM image_asset ia " " WHERE ia.entity_id = ec.id)" ), {"pid": self.project_id, "null_outlook": NULL_OUTLOOK_SHORT_ID}).fetchall() marked = 0 for row in rows: if row[3] == CANON_STATUS_ORPHANED: continue self.db.execute(sql_text( "UPDATE entity_canon SET status = :st, updated_at = :now WHERE id = :oid" ), {"st": CANON_STATUS_ORPHANED, "now": self.now, "oid": row[0]}) marked += 1 if marked: self.db.flush() self.logger.info( "아무 데도 안 붙은 아웃룩 %d개를 %s 로 표시 (삭제 안 함): %s", marked, CANON_STATUS_ORPHANED, [r[1] for r in rows][:5], ) return marked