"""엔티티 신원 계약 — `short_id` 발급과 수명 상태를 **한 곳**에서 정한다. ## 왜 이 모듈이 있나 2026-09-04 실측(골목 끝 `da049582`, 3화). 화마다 `short_id` 를 **위치 기준으로 `01` 부터** 다시 매기고 있었다 (`entity_steps._assign_short_ids`, `outlook_steps`). `entity_canon` 의 유일성은 `(project_id, short_id)` 라 2화의 `C01` 이 1화의 `C01` **행을 찾아 덮었다** — 1화 CP C01=민수 2·3화 CP C01=정임 지금 DB C01=정임 1화 CP O01=남색작업복 3화 CP O01=감색차장제복 → DB O01 은 이름이 「감색차장제복」인데 붙어 있는 참조 이미지는 **1화가 만든 남색작업복**이다. 그림과 이름이 갈렸다. 그래서 번호는 **모델도 스텝도 아니고 프로젝트 장부가 소유한다.** 한 번 쓴 번호는 다시 안 나온다. ## 이 모듈이 지키는 것 - 발급은 `project_short_id_counter` 의 **줄지 않는 값**에서 나온다. - 첫 값(seed)은 DB 최대값**만으로는 안 된다** — 이미 지워진 번호를 다시 발급하게 된다(실측: 1화 `O03 회색외투` 는 DB 에 없지만 1화 체크포인트에는 살아 있다). **살아 있는 모든 에피소드 체크포인트**까지 같이 본다. - `O00` 은 Null Outlook **예약값**이다. 발급기는 절대 내지 않는다. - 상한은 999 다. 소비자 정규식 20여 곳이 ``[CLPO]\\d{2,3}`` 로 세 자리까지만 받는다. 넓히는 것은 별도 범위이고, 여기서는 **넘기 전에 선다.** """ from __future__ import annotations import json import logging import pathlib from typing import Any, Dict, Iterable, List, Optional from sqlalchemy import or_, text as sql_text from sqlalchemy.orm import Session as OrmSession from app.modules.pipeline.grounding_entity_contract import OWNER_PREFIX logger = logging.getLogger(__name__) #: Null Outlook 예약값 — 「옷 배정이 없다」는 뜻의 sentinel 이고 실체가 아니다. #: `image_steps` · `reference_phase3_service` · `reference_pipeline_orchestrator` #: 가 이 값으로 찾는다. ★발급기가 내면 그 자리들이 실제 아웃룩을 sentinel 로 #: 읽는다. NULL_OUTLOOK_SHORT_ID = "O00" #: 발급 범위. 하한이 1인 이유는 위 sentinel 이 0 을 쓰기 때문이다. SHORT_ID_MIN = 1 #: 상한. ★임의로 고른 수가 아니다 — 소비자 정규식이 세 자리까지만 받는다. #: `visible_entities_validator` · `detail_steps` · `subject_reference_policy` · #: `gaze_direction` · `frame_spatial_contract` · `outlook_dedup` · #: `visual_continuity_anchor_plan` · 프런트까지 같은 모양이다. SHORT_ID_MAX = 999 #: `entity_canon.status` — **집계된 수명**이다. 어느 화가 쓰는지는 링크가 안다. CANON_STATUS_ACTIVE = "active" CANON_STATUS_ORPHANED = "orphaned" #: `entity_episode_link.presence_status` — **이 화에서** 어떤 자리인가. #: ★프로젝트 전역 상태가 아니다. 1화에서 저빈도였다고 5화에서도 저빈도가 #: 아니다. PRESENCE_ACTIVE = "active" PRESENCE_SHELVED = "shelved" #: 체크포인트에서 번호를 주울 때 볼 **키 이름들**. ★글자 훑기가 아니라 #: 키 기준이다 — 본문에서 ID 처럼 생긴 것을 주우면 프롬프트 예시까지 삼킨다. _SHORT_ID_KEYS = ("short_id", "outlook_id") class SeedUnreadable(RuntimeError): """seed 를 정할 근거(체크포인트)를 못 읽었다. ★낮은 seed 로 가면 안 된다.""" class ShortIdBadValue(ValueError): """앞 단계가 준 `short_id` 가 계약 밖이다. ★그대로 믿으면 남을 덮는다.""" class ShortIdExhausted(RuntimeError): """발급 가능한 번호를 다 썼다. ★조용히 넘어가면 남의 번호를 덮는다.""" def format_short_id(owner: str, num: int) -> str: """`("character", 7)` → `"C07"`. 100 부터는 자연히 세 자리가 된다.""" prefix = OWNER_PREFIX.get(owner) if prefix is None: raise ValueError(f"모르는 갈래: {owner!r}") if not (SHORT_ID_MIN <= num <= SHORT_ID_MAX): raise ShortIdExhausted( f"{owner} 번호 {num} 은 발급 범위 밖이다 " f"({SHORT_ID_MIN}~{SHORT_ID_MAX}). 소비자 정규식이 세 자리까지만 " f"받으므로 여기서 선다 — 넓히려면 그 정규식들을 먼저 고쳐야 한다.") return f"{prefix}{num:02d}" def parse_short_id(owner: str, short_id: str) -> Optional[int]: """`("character", "C07")` → `7`. 그 갈래의 것이 아니면 `None`. ★`split_final_id` 를 쓴다 — 접두표가 prefix-free 가 아니라서 (`L` 과 `LP`) 직접 자르면 `LP01` 을 `L` 것으로 읽는다. """ from app.modules.pipeline.grounding_entity_contract import split_final_id got = split_final_id(str(short_id or "")) if got is None: return None got_owner, num = got return num if got_owner == owner else None def _walk_short_ids(node: Any, out: List[str]) -> None: """JSON 을 훑어 `short_id`/`outlook_id` **키의 값**만 모은다.""" if isinstance(node, dict): for k, v in node.items(): if k in _SHORT_ID_KEYS and isinstance(v, str) and v: out.append(v) else: _walk_short_ids(v, out) elif isinstance(node, list): for v in node: _walk_short_ids(v, out) def checkpoint_high_water(project_id: str, owner: str) -> int: """살아 있는 **모든 에피소드 체크포인트**에서 그 갈래의 최대 번호. ★없으면 0. 읽다 깨진 파일은 건너뛰되 **조용히 넘어가지 않는다** — 못 읽은 파일이 있으면 그만큼 seed 가 낮아져 남의 번호를 다시 발급할 수 있다. """ from app.core.config import settings root = pathlib.Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" if not root.is_dir(): return 0 best = 0 unreadable: List[str] = [] # ★보관본(`manifest_*.json`)은 집지 않는다 — 현행 판만 본다. for mf in root.glob("*/*/manifest.json"): try: data = json.loads(mf.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 unreadable.append(str(mf)) continue found: List[str] = [] _walk_short_ids(data.get("data"), found) for sid in found: num = parse_short_id(owner, sid) if num is not None: best = max(best, num) if unreadable: # ★★★못 읽고 **계속 가면** seed 가 실제보다 낮아져 이미 쓴 번호를 # 다시 발급한다 — 그러면 남의 행을 덮는다. 경고로 넘길 일이 아니다 # (Codex BLOCK 2026-09-04). 여기서 선다. raise SeedUnreadable( f"살아 있는 체크포인트 {len(unreadable)}개를 못 읽어 `{owner}` 번호 " f"seed 를 못 정한다 — 낮은 seed 로 발급하면 이미 쓴 번호를 다시 낸다. " f"먼저 고쳐라: {unreadable[:3]}") return best def db_high_water(db: OrmSession, project_id: str, owner: str) -> int: """`entity_canon` 에 이미 있는 그 갈래의 최대 번호. 없으면 0.""" rows = db.execute(sql_text( "SELECT short_id FROM entity_canon " "WHERE project_id = :pid AND short_id IS NOT NULL" ), {"pid": project_id}).fetchall() best = 0 for (sid,) in rows: num = parse_short_id(owner, sid) if num is not None: best = max(best, num) return best def reserve_short_ids( db: OrmSession, project_id: str, owner: str, count: int, ) -> List[str]: """그 갈래의 다음 번호 `count` 개를 **원자적으로** 발급한다. ★조회와 발급이 따로면 동시에 도는 둘이 같은 번호를 받는다. 한 문장으로 올리고 그 반환값을 쓴다 — 락을 따로 잡지 않아도 PostgreSQL 이 이 행을 직렬화한다. ★첫 호출에서 seed 를 넣는다. seed = max(DB, 살아 있는 체크포인트). 둘이 나란히 seed 를 넣어도 `ON CONFLICT DO NOTHING` 으로 하나만 들어가고, 둘이 같은 값을 계산했으므로 결과가 같다. """ if count <= 0: return [] if owner not in OWNER_PREFIX: raise ValueError(f"모르는 갈래: {owner!r}") seed = max(db_high_water(db, project_id, owner), checkpoint_high_water(project_id, owner)) db.execute(sql_text( "INSERT INTO project_short_id_counter (project_id, owner, high_water) " "VALUES (:pid, :owner, :seed) ON CONFLICT (project_id, owner) DO NOTHING" ), {"pid": project_id, "owner": owner, "seed": seed}) # ★seed 는 「이미 쓴 최대」이므로 기존 행이 더 낮으면 끌어올린다. # (앞 판에서 낮게 잡힌 장부가 남아 있을 수 있다.) row = db.execute(sql_text( "UPDATE project_short_id_counter " "SET high_water = GREATEST(high_water, :seed) + :count " "WHERE project_id = :pid AND owner = :owner " "RETURNING high_water" ), {"pid": project_id, "owner": owner, "seed": seed, "count": count}).fetchone() if row is None: raise RuntimeError( f"short_id 장부를 못 올렸다 (project={project_id}, owner={owner})") end = int(row[0]) start = end - count + 1 if end > SHORT_ID_MAX: raise ShortIdExhausted( f"{owner} 번호가 {SHORT_ID_MAX} 를 넘는다 (요청 {count}개, 끝 {end}). " f"소비자 정규식이 세 자리까지만 받는다 — 넓히는 것은 별도 범위다.") return [format_short_id(owner, n) for n in range(start, end + 1)] def assign_short_ids( db: OrmSession, project_id: str, owner: str, entities: Iterable[Dict], ) -> List[Dict]: """`short_id` 가 **없는** 것에만 프로젝트 번호를 채운다. ★이미 있는 것은 건드리지 않는다 — 앞 화에서 물려받은 신원이다. """ rows = list(entities) # ★★★이미 든 것도 **믿지 않는다** (Codex BLOCK 2026-09-04). 앞 단계나 # 옛 CP 가 준 값이 갈래가 다르거나(`L01` 이 prop 줄에), 범위 밖이거나, # 예약값(`O00`)이거나, 한 판 안에서 겹칠 수 있다. 그대로 쓰면 남의 행을 # 덮는다 — 그것이 이 판이 고치려는 결함 자체다. seen: Dict[str, int] = {} for i, e in enumerate(rows): sid = str(e.get("short_id") or "").strip() if not sid: continue if sid == NULL_OUTLOOK_SHORT_ID: raise ShortIdBadValue( f"예약값 {sid} 이 일반 신원으로 들어왔다 (자리 {i})") num = parse_short_id(owner, sid) if num is None: raise ShortIdBadValue( f"`{sid}` 는 `{owner}` 의 신원이 아니다 (자리 {i}) — " f"갈래가 섞였거나 형식이 틀렸다") if not (SHORT_ID_MIN <= num <= SHORT_ID_MAX): raise ShortIdBadValue( f"`{sid}` 가 발급 범위 밖이다 ({SHORT_ID_MIN}~{SHORT_ID_MAX})") if sid in seen: raise ShortIdBadValue( f"`{sid}` 를 두 줄이 갖고 있다 (자리 {seen[sid]}·{i}) — " f"한 판 안에서 신원이 겹치면 뒤가 앞을 덮는다") seen[sid] = i need = [e for e in rows if not str(e.get("short_id") or "").strip()] if need: for e, sid in zip(need, reserve_short_ids(db, project_id, owner, len(need))): e["short_id"] = sid return rows def active_episode_canon_ids(db, project_id: str, episode_id: str) -> List[str]: """**이 화에서 살아 있는** canon id — 보류(`shelved`)를 뺀 것. ★★★이것이 「이 화에 무엇이 나오나」의 **정본**이다. 소비자마다 링크를 직접 읽으면 보류 제외가 한 곳만 붙고 나머지는 샌다 — 실측 2026-09-04: 이 화 링크를 읽는 자리가 **21곳**인데 거르는 곳은 **1곳**이었다. ★쓰는 쪽(sync 서비스)은 이것을 **안 쓴다.** 그쪽은 보류 링크도 봐야 `active` 로 되살릴 수 있다. """ from app.models.project import EntityEpisodeLink rows = db.query(EntityEpisodeLink.canon_id).filter( EntityEpisodeLink.project_id == project_id, EntityEpisodeLink.episode_id == episode_id, EntityEpisodeLink.presence_status != PRESENCE_SHELVED, ).all() return [r[0] for r in rows] def episode_outlook_rows(db, project_id: str, episode_id: str) -> List[Any]: """**이 화의** 인물↔아웃룩 배정 행. ★★★`character_outlook` 은 2026-09-04 (alembic 014) 전까지 화 범위가 없었다. 그래서 같은 인물이 1화에 O01, 2화에 O02 를 입으면 **2화가 둘 다** 생성·검증·프롬프트 재료로 봤고, 한 화의 `O00` 이 다른 화 판단까지 오염시켰다 (Codex BLOCK 2026-09-04). ## legacy(`episode_id IS NULL`) 를 어떻게 다루나 ★★앞 판은 exact 와 **모든 NULL 행을 OR 로 무조건 합쳤다.** 그런데 014 는 기존 행을 전부 NULL 로 남긴다 — 그러면 손상 복구 대상인 **다중 화 legacy 프로젝트에서 과거 옷과 `O00` 이 모든 화로 다시 들어온다.** 고치려던 것이 그대로 돌아오는 셈이다 (Codex BLOCK 2026-09-04 재지적). 그래서 계약을 좁힌다 — 이 화의 exact 행이 **하나라도** 있으면 → **exact 만** 하나도 없으면 → NULL 행 중 **이 화의 active 링크 양 끝** (인물·아웃룩 **둘 다**)에 걸린 것만 ★양 끝을 요구하는 까닭: 한쪽만 걸린 행은 「이 화의 배정」이라고 볼 근거가 없다. 모호하면 넓히지 않는다. """ from app.models.project import CharacterOutlook exact = db.query(CharacterOutlook).filter( CharacterOutlook.project_id == project_id, CharacterOutlook.episode_id == episode_id, ).all() if exact: return exact legacy = db.query(CharacterOutlook).filter( CharacterOutlook.project_id == project_id, CharacterOutlook.episode_id.is_(None), ).all() if not legacy: return [] active = set(active_episode_canon_ids(db, project_id, episode_id)) return [r for r in legacy if r.character_id in active and r.outlook_id in active] def episode_outlook_pairs(db, project_id: str, episode_id: str) -> List[Any]: """`(character_id, outlook_id)` 쌍 — 위와 **같은 범위**.""" return [(r.character_id, r.outlook_id) for r in episode_outlook_rows(db, project_id, episode_id)]