"""구조 관계를 `entity_relation` CP 모양으로 **투영**한다. ★inert. 유료 0. ## 무엇인가 `location_part` → `location` 의 `part_of` 를 **관계 표**에 넣는다. ★새 표를 안 만든다 — `RelationFact`/`RelationParticipant` 가 이미 `relation_family`·`relation_type`·참가자 `role`/`order` 를 갖는다. ## 지켜야 할 것 (Codex 2026-08-31) ★**한 CP = 한 producer.** `entity_relation` CP 는 활성 `EntityRelationStep`(13.6)이 만든다. adapter 가 그 CP 를 **직접 쓰면** `visual_variant` 와 `part_of` 가 실행·재개 차례에 따라 **서로 덮는다**. → `v2_chunk` 에서는 그 스텝이 **결정적 no-call projection** 으로 이것을 불러 CP **한 벌**을 만든다. ★**타입별로 가른다.** 한 타입의 sync 가 다른 타입을 **stale 로 지우면 안 된다**. 실측: 기존 조회는 이미 `WHERE relation_type='visual_variant'` 로 잠겨 있다 — 새 `part_of` 갈래도 **제 타입만** 봐야 한다. ★`outlook → character` 는 **여기 안 온다.** `CharacterOutlook` 이 SOT 이고 orchestrator 가 RelationSync(2) → OutlookSync(4) 라 관계를 쓸 때 outlook canon 이 **아직 없을 수 있다**. ## 아직 안 쓰인다 `app/` 어디서도 부르지 않는다. """ from __future__ import annotations from typing import Any, Dict, List, Sequence, Tuple from app.modules.pipeline.grounding_entity_contract import ( RELATION_OWNER_PAIRS, owner_of_final_id) #: 이 투영이 내는 관계 종류. ★**이것만** 낸다. RELATION_PART_OF = "part_of" #: ★이 투영이 **무엇으로 만든 것인지**. 산출에 실려 뒤가 되짚는다. #: 2026-09-01 — production 배선(관계 스텝의 호출 0 갈래)이 이것을 적는다. #: ★2026-09-02 판 2 — DB 가 받는 짝을 **고르고** 나머지는 까닭과 함께 #: 남긴다(`select_db_part_of`). 앞 판은 안 받는 짝에서 **주행을 세웠다**. PROJECTION_CONTRACT_VERSION = "2.202609020300" #: 참가자 역할. ROLE_PART = "part" ROLE_WHOLE = "whole" #: `part_of` 가 이을 수 있는 갈래 짝. ★이름을 안 본다 — **갈래**로 가른다. #: `outlook → character` 는 **여기 없다**(CharacterOutlook 이 SOT). #: ★★★**여기서 다시 적지 않는다** (Codex BLOCK 2026-09-01). 계약이 두 벌이면 #: DB 조회 owner 와 투영 검사가 갈린다 — 하나만 고쳐진다. ALLOWED_PART_OF: Dict[str, str] = { part: whole for part, whole in RELATION_OWNER_PAIRS[RELATION_PART_OF]} class RelationProjectionError(RuntimeError): """투영이 계약을 어겼다. ★조용히 넘어가지 않는다.""" def select_db_part_of(links: Sequence[Dict[str, Any]] ) -> Dict[str, Any]: """`part_of` 짝들 → **DB 가 받는 것**과 **안 받는 것**을 갈라 낸다. ★★★producer 의 `part_of` 는 **한 벌인데 소비자가 둘**이다 (실측 2026-09-02 유료 재개): `grounding_facet_binding` 아웃룩·장소부분을 주인에 붙이는 **재료**. `O01 → C01` 같은 짝이 **거기 쓰인다** 이 투영 DB `relation_fact` — 계약이 받는 짝은 `RELATION_OWNER_PAIRS["part_of"]` 뿐 그래서 producer 가 `O01 → C01` 을 내는 것은 **옳고**, 좁은 쪽은 이 투영이다. 앞 판은 그것을 받고 `assert_pair` 로 **주행을 세웠다** — 실제 유료 주행이 `entity_relation` 에서 죽었다. ★**조용히 버리지 않는다** — 안 받는 짝도 까닭과 함께 돌려준다. Returns: `{"for_db": [...], "not_for_db": [{"part","whole","owners","why"}]}` """ from app.modules.pipeline.grounding_entity_contract import ( RELATION_OWNER_PAIRS, owner_of_final_id) ok = set(RELATION_OWNER_PAIRS.get(RELATION_PART_OF, ())) for_db, other = [], [] for x in links: part = str((x or {}).get("part") or "") whole = str((x or {}).get("whole") or "") if not part or not whole: raise RelationProjectionError(f"짝이 비었다: {x!r}") pair = (owner_of_final_id(part), owner_of_final_id(whole)) # ★★★**비정본 ID 는 감사 목록으로 넘기지 않는다** (Codex 2026-09-02). # `Pfoo` 같은 글자는 `owner_of_final_id` 가 `None` 을 주는데, 그것을 # 「DB 계약 밖」으로 접으면 **모르는 것이 조용히 지나간다**. # 감사 목록에 남는 것은 「정본이지만 DB 가 안 받는 짝」뿐이어야 한다. if pair[0] is None or pair[1] is None: raise RelationProjectionError( f"정본 ID 가 아니다: part={part!r}({pair[0]}) · " f"whole={whole!r}({pair[1]})") if part == whole: raise RelationProjectionError(f"자기 자신을 잇는다: {part!r}") if pair in ok: for_db.append({"part": part, "whole": whole}) else: other.append({"part": part, "whole": whole, "owners": list(pair), "why": ("이 갈래 짝은 DB `part_of` 계약이 안 받는다 " f"— 받는 것 {sorted(ok)}")}) return {"for_db": for_db, "not_for_db": other} def project_part_of(links: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: """`{part, whole}` 정본 ID 짝들 → `entity_relation` CP 행들. ★**결정적**이다 — 같은 입력이면 같은 차례·같은 내용. Args: links: `{"part": "LP01", "whole": "L03"}` 꼴. ★**정본 ID** 다. Returns: CP 행 목록. 각 행은 `relation_type` 을 **명시**한다 — 그래야 sync 가 타입별로 가른다. """ seen: Dict[Tuple[str, str], None] = {} for x in links: part = str((x or {}).get("part") or "") whole = str((x or {}).get("whole") or "") if not part or not whole: raise RelationProjectionError(f"짝이 비었다: {x!r}") assert_pair(part, whole) # ★투영과 sync 가 **같은 함수**를 쓴다 seen[(part, whole)] = None return [{ "relation_type": RELATION_PART_OF, "participants": [{"short_id": p, "role": ROLE_PART, "order": 1}, {"short_id": w, "role": ROLE_WHOLE, "order": 2}], } for p, w in sorted(seen)] def assert_pair(part: str, whole: str) -> None: """이 짝이 **갈래 계약 안**인가. ★아니면 선다 — 조용히 넘어가지 않는다. ★★★`desired_keys` 가 이걸 안 부르던 것이 결함이었다 (Codex BLOCK 2026-09-01). `project_part_of` 만 검사하고 sync 는 참가자 역할·존재만 봐서, 망가진 CP 의 `C01(part) → P01(whole)` 도 양쪽 canon 만 있으면 **DB 에 `part_of` 로 저장**됐다. """ po, wo = owner_of_final_id(part), owner_of_final_id(whole) if po is None or wo is None: raise RelationProjectionError( f"정본 ID 가 아니다: part={part!r}({po}) whole={whole!r}({wo})") want = ALLOWED_PART_OF.get(po) if want is None: raise RelationProjectionError( f"`{po}` 는 `part_of` 로 잇지 않는다 — {sorted(ALLOWED_PART_OF)}") if wo != want: raise RelationProjectionError( f"`{po}` 의 부모는 `{want}` 여야 하는데 `{wo}` 다") if part == whole: raise RelationProjectionError(f"제 자신에 붙었다: {part}") def desired_keys(rows: Sequence[Dict[str, Any]], *, relation_type: str) -> List[Tuple[str, str]]: """CP 행들에서 **그 타입만** 골라 짝을 낸다. ★★한 타입의 delta 가 다른 타입을 **stale 로 지우면 안 된다** (Codex). 부르는 쪽이 타입을 **명시**하게 해서 실수로 넓히지 못하게 한다. ★★★그리고 **갈래 짝을 여기서도 검사한다** — DB sync 가 실제로 부르는 자리가 여기다. 투영에서만 검사하면 CP 가 어디서 왔든 믿는 것이 된다. """ out: List[Tuple[str, str]] = [] for r in rows: if str((r or {}).get("relation_type") or "") != relation_type: continue by_role = {str(p.get("role")): str(p.get("short_id")) for p in (r.get("participants") or ())} if relation_type == RELATION_PART_OF: p, w = by_role.get(ROLE_PART), by_role.get(ROLE_WHOLE) if not p or not w: raise RelationProjectionError( f"`{RELATION_PART_OF}` 에 참가자가 모자란다: {by_role}") assert_pair(p, w) # ★저장 직전 자리에서도 본다 out.append((p, w)) else: raise RelationProjectionError( f"이 투영이 모르는 타입 {relation_type!r}") return sorted(set(out))