"""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.name_matcher import build_name_index, lookup_name 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_removed": 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", []) _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 잔존이지만 outlook은 # `_remove_orphan_outlooks`가 character_outlook 미연결 row를 자동 삭제 → # 별도 orphan 경고 불필요 (entity_canon character/location/prop과 다른 점). 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() 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) or existing_ol_by_name.get(ol_name) if existing: 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, )) self.db.add(EntityEpisodeLink( id=str(uuid.uuid4()), canon_id=ol_id, project_id=self.project_id, episode_id=self.episode_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: _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) or existing_ol_by_name.get(_n) if _e: _matched_ids.add(_e.id) for e in _existing_ols: if e.id not in _matched_ids: # entity_canon을 참조하는 모든 child FK 정리 (RESTRICT 모두). # 누락 시 force 재실행 → DELETE FROM entity_canon UNIQUE/FK # violation으로 dispatcher abort. Claude opus IMPORTANT-1 적용: # entity_canon 참조 FK 전수 grep 후 4개 테이블 모두 처리. self.db.execute(sql_text( "DELETE FROM entity_episode_link WHERE canon_id = :cid" ), {"cid": e.id}) self.db.execute(sql_text( "DELETE FROM character_outlook WHERE outlook_id = :cid" ), {"cid": e.id}) self.db.execute(sql_text( "DELETE FROM entity_alias WHERE canon_id = :cid" ), {"cid": e.id}) self.db.execute(sql_text( "DELETE FROM relation_participant WHERE canon_id = :cid" ), {"cid": e.id}) self.db.execute(sql_text( "DELETE FROM entity_canon WHERE id = :eid" ), {"eid": e.id}) 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 ) 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._remove_orphan_outlooks() return { "outlooks": outlook_count, "links": link_count, "links_inserted": links_inserted, "links_deleted": links_deleted, "orphans_removed": orphans_removed, } def _delta_sync_character_outlook( self, outlooks: list, ol_data: dict, skip_stale: 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", "") cid = char_short_to_id.get(csid) or lookup_name(char_name_id, cname) oid = ol_short_to_id.get(osid) or lookup_name(ol_name_to_id, oname) if cid and oid: desired_pairs.add((cid, oid)) # 현재 상태: 이 에피소드 character에 연결된 character_outlook만 범위로 existing_rows = self.db.execute( sql_text( "SELECT co.id, co.character_id, co.outlook_id " "FROM character_outlook co " "JOIN entity_episode_link eel ON eel.canon_id = co.character_id " "JOIN entity_canon ec ON ec.id = co.character_id " "WHERE co.project_id = :pid AND eel.episode_id = :eid " "AND ec.entity_type = 'character'" ), {"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()) to_insert = desired_pairs - existing_pairs # Codex P1-2: partial 시 stale pair 삭제 skip — 부분 cp의 missing이 # 단순 누락일 수 있음 (다음 completed 동기화에서 정리). 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, character_id, outlook_id, created_at) " "VALUES (:id, :pid, :cid, :oid, :now)" ), { "id": str(uuid.uuid4()), "pid": self.project_id, "cid": cid, "oid": oid, "now": self.now, }, ) if to_insert or to_delete: self.db.flush() return len(desired_pairs), len(to_insert), len(to_delete) def _remove_orphan_outlooks(self) -> int: orphan_result = self.db.execute(sql_text( "SELECT ec.id, ec.name, ec.short_id FROM entity_canon ec " "WHERE ec.project_id = :pid AND ec.entity_type = 'outlook' " "AND ec.id NOT IN (SELECT outlook_id FROM character_outlook WHERE project_id = :pid)" ), {"pid": self.project_id}).fetchall() if not orphan_result: return 0 orphan_ids = [r[0] for r in orphan_result if r[2] != "O00"] orphan_names = [r[1] for r in orphan_result if r[2] != "O00"] for oid in orphan_ids: # entity_canon 참조 4 child 테이블 모두 정리 (RESTRICT FK). # 본 함수의 SELECT가 character_outlook 미참조만 추리지만 entity_alias / # relation_participant FK는 보장 안 됨 — Codex IMPORTANT 적용. self.db.execute(sql_text( "DELETE FROM entity_episode_link WHERE canon_id = :oid" ), {"oid": oid}) self.db.execute(sql_text( "DELETE FROM entity_alias WHERE canon_id = :oid" ), {"oid": oid}) self.db.execute(sql_text( "DELETE FROM relation_participant WHERE canon_id = :oid" ), {"oid": oid}) # character_outlook은 위 SELECT가 비참조만 추렸지만 안전상 명시. self.db.execute(sql_text( "DELETE FROM character_outlook WHERE outlook_id = :oid" ), {"oid": oid}) self.db.execute(sql_text( "DELETE FROM entity_canon WHERE id = :oid" ), {"oid": oid}) self.db.flush() if orphan_ids: self.logger.info("Removed %d orphan outlooks: %s", len(orphan_ids), orphan_names[:5]) return len(orphan_ids)