"""C(c) 산출 → `entity_merge` 모양. ★**호출 0 · 결정적** · 아직 안 켠다. ## 왜 adapter 인가 `entity_merge` 체크포인트를 읽는 모듈이 **12개**다. 모양을 바꾸면 그 전부가 같이 바뀌어야 하고, 그러면 「원자적 전환」이 아니라 대공사가 된다. 그래서 새 producer 는 **기존 모양으로 낸다**. ## 무엇을 옮기고 무엇을 안 옮기나 — 근거 실제 CP 로 쟀다 (2026-08-31). entity_merge 행 = {name, description, visual_traits, short_id} C(c) 행 = {surface_form, visual_brief, evidence_quotes, 두 축, …} ★**`visual_brief` 를 `description` 자리에 넣지 않는다.** 둘 다 「겉모습」을 말하지만 `description` 에는 `visual_traits` 와 함께 **단일 시각 형태 원칙**이 붙어 있다 — 「A이거나 B」 같은 표현 금지. 그 글이 **T2I 프롬프트에 그대로** 들어가기 때문이다(`entity_extract_v4/prop.md`). `visual_brief` 는 그 계약을 **받은 적이 없다.** 옮기면 생성 모델이 임의 발현을 골라 엉뚱한 물체를 그린다. ★**비워 두는 것이 안전한 까닭**도 실측이다 — entity_detail(14.0) 의 입력 queue = (name, entity_type, short_id) 뿐 → merge 의 description 을 **안 받는다**. 이름·갈래·세계 규칙으로 새로 만든다 background_chain_planning: 「entity_detail 에서 description/visual_traits 보강 (있으면 우선)」 → **detail 이 이긴다** 그래서 merge 의 `description` 은 **detail 이 없을 때의 대비책**이고, 정본이 아니다. C(c) 는 그 자리를 비우고 `entity_detail` 이 채운다. 겉모습 글은 **버리지 않고** 이름이 다른 칸(`grounding_visual_brief`)에 남긴다 — 그래야 누가 T2I 정본으로 잘못 읽지 않는다. ## 안 하는 것 - facet(`outlook`·`location_part`)을 base 갈래로 **우회 등록**하지 않는다 - 등록 **미확정** 행을 「미등록」으로 접지 않는다 — 따로 돌려준다 - 이름·부분문자열로 기존 행과 짝짓지 않는다 ## ★★신원은 **장부가 이미 정했다** — 다시 정하지 않는다 `reduce_episode` 의 등록 장부가 `final_id` 를 준다(`C01`·`L01`·`P01`). 실제 유료 산출에서 92행 전부 값이 있고, 겹침 0 이며 접두가 갈래와 다 맞았다. ★그런데 앞 판 adapter 는 **번호를 새로 발급했다** (Codex BLOCK 2026-08-31). 그러면 같은 대상의 신원이 **둘**이 되고, `final_id` 로 걸린 참조 의무·하류 결속이 엉뚱한 행을 가리킨다. `short_id = final_id` 다. ★그래서 이미 있는 행과 번호가 겹치면 **다시 매기지 않고 선다.** 다시 매기는 순간 장부와 갈라지기 때문이다. chunk 모드에서 `entity_extract_*` 는 no-call 이라 기존 행이 애초에 없지만, 있으면 그것은 **두 벌이 선 상태**라 사람이 봐야 한다. """ from __future__ import annotations import copy from app.modules.pipeline import grounding_entity_contract as _ec import unicodedata from typing import Any, Dict, List, Optional, Sequence from app.modules.pipeline.grounding_entity_contract import ( MATERIALIZABLE_OWNER_TYPES, OWNER_PREFIX, owner_of_final_id) #: ★값의 주인은 **계약 모듈**이다 — 여기 다시 적으면 두 벌이 되고, 결속을 #: 검증하는 쪽(`grounding_carry`)은 C(c) 를 못 부르므로 그쪽이 못 본다. ADAPTER_CONTRACT_VERSION = _ec.PRODUCER_CONTRACT_VERSION ADAPTER_ISSUER = _ec.PRODUCER_ISSUER #: ★후보에 링크를 싣는 칸 이름. 받는 쪽과 **한 벌**이다. LINK_KEY = "producer_link" #: owner → `entity_merge` 의 복수 키. ★**base 갈래만** 자리가 있다. #: facet 두 갈래는 여기 없다 — 없는 것이 계약이다. OWNER_TO_MERGE_KEY: Dict[str, str] = { o: f"{o}s" for o in MATERIALIZABLE_OWNER_TYPES if o != "outlook" } #: 못 옮긴 사유. SKIP_NOT_REGISTERED = "not_registered" SKIP_UNRESOLVED = "registration_unresolved" SKIP_NO_MERGE_KEY = "no_entity_merge_key" class CandidateIdCollision(RuntimeError): """emitter 가 낸 후보 id 가 겹친다. ★조용히 둘로 만들지 않는다.""" class MissingSourceSpan(RuntimeError): """등록된 행에 원문 좌표가 없다. ★빈 anchor 로 후보를 안 만든다.""" class ShortIdCollision(RuntimeError): """장부의 `final_id` 가 **이미 있는 행**과 겹친다. ★다시 매기지 않고 선다.""" class LedgerContractViolation(RuntimeError): """들어온 것이 **구조화 계약**을 어긴다. ★소비 전에 선다. `reduce_episode` 는 행과 장부를 1:1 로 만들고 `final_id` 접두를 갈래에 맞춘다. 그러나 adapter 는 **durable CP·재생 경계**에서도 불린다 — 거기서 어긋난 것이 오면 조용히 흘려보내면 안 된다 (Codex BLOCK 2026-08-31). 조용히 흘리면 두 가지가 난다 — ①장부에만 있는 등록 행이 **사라진다** (엔티티가 안 생긴다) ②`character` 인데 `P01` 같은 어긋난 신원이 그대로 나간다 → 참조 의무가 **엉뚱한 행**에 걸린다 """ def validate_ledger(reduced: Dict[str, Any]) -> None: """구조화 입력 계약. ★**중앙 ID 계약**으로 본다 — 이름·regex 안 쓴다. Raises: LedgerContractViolation: 아래 중 하나라도 어긋나면. =========================== ========================================== 행 `local_id` 집합 장부 key 집합과 **같아야** 한다 `registered is True` `final_id` 가 **반드시** 있어야 한다 `final_id` 접두 `OWNER_PREFIX[owner]` 와 **같아야** 한다 등록된 `final_id` 전역 중복 **0** =========================== ========================================== """ rows = list(reduced.get("rows") or ()) ledger = dict(reduced.get("registered") or {}) by_id: Dict[str, Dict[str, Any]] = {} for r in rows: lid = str(r.get("local_id") or "") # ★★빈 신원을 그냥 두면 **행 여럿이 같은 빈 키**가 되고, 장부와 # 「집합이 같다」가 우연히 맞는다 (Codex 2026-08-31). if not lid.strip(): raise LedgerContractViolation( "행에 `local_id` 가 없다 — 빈 신원은 서로 같아 보인다") if lid in by_id: raise LedgerContractViolation(f"행 {lid} 이 두 번 있다") by_id[lid] = r for k in ledger: if not str(k or "").strip(): raise LedgerContractViolation("장부에 빈 key 가 있다") only_rows = sorted(set(by_id) - set(ledger)) only_ledger = sorted(set(ledger) - set(by_id)) if only_rows or only_ledger: raise LedgerContractViolation( f"행과 장부가 1:1 이 아니다 — 행에만 {only_rows[:5]} · " f"장부에만 {only_ledger[:5]}. 장부에만 있는 등록 행은 **조용히 " "사라진다**") seen: Dict[str, str] = {} for lid, rec in sorted(ledger.items()): # ★★`registered` 는 **정확히** True/False/None 이어야 한다. # `"true"` 같은 문자열을 `is not True` 로 접으면 **등록된 것이 # 조용히 미등록이 된다** — truthiness 부류다 (Codex 2026-08-31). got = rec.get("registered") if got is not True and got is not False and got is not None: raise LedgerContractViolation( f"{lid}: `registered` 가 {got!r}({type(got).__name__}) 다 — " "True/False/None 만 쓴다. 다른 값을 조용히 접으면 등록이 " "미등록으로 바뀐다") if got is not True: continue fid = str(rec.get("final_id") or "") if not fid: raise LedgerContractViolation( f"{lid} 은 등록됐다는데 `final_id` 가 없다 — 이름으로 지어내지 " "않는다. 사람이 장부를 봐야 한다") owner = str(by_id[lid].get("owner_type") or "") if owner not in OWNER_PREFIX: raise LedgerContractViolation(f"{lid}: 모르는 갈래 {owner!r}") # ★★`startswith` 로 보면 `location` 행에 `LP01` 이 통과한다 — # 접두표가 **prefix-free 가 아니다**(`L` ⊂ `LP`, Codex 2026-08-31). # **가장 긴 접두**로 가르는 계약 함수를 쓴다. got = owner_of_final_id(fid) if got != owner: raise LedgerContractViolation( f"{lid}: 갈래는 {owner!r} 인데 신원 {fid!r} 는 " f"{got!r} 의 것이다 — 참조 의무가 엉뚱한 행에 걸린다") if fid in seen: raise LedgerContractViolation( f"신원 {fid!r} 를 {seen[fid]} 와 {lid} 이 함께 쓴다 — " "전역 중복은 0 이어야 한다") seen[fid] = lid def _clean(text: str) -> str: """**보이는 이름**만 다듬는다 — NFKC + 공백 축약. ★대소문자는 안 건드린다. ★신원은 여기서 안 정한다. 신원은 장부의 `final_id` 다 — 이 함수 결과가 무엇이든 어느 행이 어느 행인지는 안 바뀐다. 비교용 정규화를 그대로 이름에 쓰면 `iPhone` 이 `iphone` 이 된다. 여기서 막으려는 것은 **보이지 않는 공백·호환 문자 차이**뿐이다. """ return " ".join(unicodedata.normalize("NFKC", str(text or "")).split()) def _provenance(lid: str, final_id: str, owner: str, row: Dict[str, Any], rec: Dict[str, Any]) -> Dict[str, Any]: """이 엔티티가 **어디서 왔는지** 원형 보존 (Codex BLOCK 2026-08-31). D 에서 `entity_merge` 호환 CP 가 하류 정본이 되면, 「왜 이 엔티티·의무가 생겼나」를 **그 행에서** 되짚을 수 있어야 한다. 앞 판은 겉모습 글·샷·두 축·사유만 실어서 **정본 인용(`source_span`+`source_quote`)과 겉모습 근거, 처분이 통째로 빠졌다.** ★`description`·`visual_traits` 에는 **한 자도 안 섞는다.** 저기 섞이면 T2I 프롬프트로 새어 나간다. """ return { "adapter_contract": ADAPTER_CONTRACT_VERSION, "local_id": lid, "final_id": final_id, "owner_type": owner, # ★부른 자리 — 인용과 **자리(span)** 를 함께. 자리가 없으면 나중에 # 원문에서 되찾을 수 없다. "occurrences": [ {"source_span": dict(o.get("source_span") or {}), "source_quote": o.get("source_quote")} for o in (row.get("occurrences") or ())], "evidence_quotes": list(row.get("evidence_quotes") or ()), "visual_brief": str(row.get("visual_brief") or ""), "shot_binding_status": row.get("shot_binding_status"), "shot_appearance_ids": list(row.get("shot_appearance_ids") or ()), "hard_to_generate": row.get("hard_to_generate"), "viewers_would_notice": row.get("viewers_would_notice"), "search_terms_native": list(row.get("search_terms_native") or ()), "language_lock_native": row.get("language_lock_native"), # ★장부 기록 **통째로** — 처분·사유·신원이 다 든다. "registration": dict(rec), } def to_entity_rows( reduced: Dict[str, Any], *, existing: Optional[Dict[str, Sequence[Dict[str, Any]]]] = None, ) -> Dict[str, Any]: """등록된 C(c) 행을 `entity_merge` 모양으로 낸다. Args: reduced: `grounding_chunk_merge.reduce_episode(...)` 산출. existing: 갈래 → 이미 있는 행. **겹치는지 보려고만** 쓴다 — 겹치면 선다. 기존 행은 **한 줄도 안 건드린다**. Raises: LedgerContractViolation: 들어온 것이 구조화 계약을 어긴다. ShortIdCollision: 장부의 `final_id` 가 이미 있는 행과 겹친다. Returns: ``{"rows": {갈래: [...]}, "skipped": [...], "contract": ...}`` `skipped` 에는 **왜 안 옮겼는지**가 행마다 들어간다. 조용히 사라지는 것이 없어야 다섯 갈래 끝점을 셀 수 있다. """ # ★★소비 **전에** 계약을 본다 — 어긋난 것을 흘려보내지 않는다. validate_ledger(reduced) rows = list(reduced.get("rows") or ()) registered = dict(reduced.get("registered") or {}) have = dict(existing or {}) used: Dict[str, set] = {} for key, olds in have.items(): used[key] = {str(o.get("short_id") or "") for o in (olds or ())} out: Dict[str, List[Dict[str, Any]]] = {} skipped: List[Dict[str, Any]] = [] # ★`local_id` 로 정렬한다 — 같은 입력이면 **같은 바이트**여야 지문이 선다. for r in sorted(rows, key=lambda x: str(x.get("local_id") or "")): lid = str(r.get("local_id") or "") owner = str(r.get("owner_type") or "") rec = registered.get(lid) or {} # ★★**안 옮긴 행에도 근거를 붙인다** (Codex 2026-08-31). 안 붙이면 # facet 23개와 미확정 30개가 「이름과 사유」만 남아, 다섯 갈래 감사가 # adapter 출력에서 **끊긴다** — 어느 인용에서 온 무엇인지 못 되짚는다. base = {"local_id": lid, "owner_type": owner, "surface_form": r.get("surface_form"), "reason": rec.get("reason"), "grounding_provenance": _provenance( lid, str(rec.get("final_id") or ""), owner, r, rec)} if rec.get("registered") is None: # ★**미확정은 미등록이 아니다.** 사람이 봐야 한다. skipped.append({**base, "skip": SKIP_UNRESOLVED}) continue if rec.get("registered") is not True: skipped.append({**base, "skip": SKIP_NOT_REGISTERED}) continue key = OWNER_TO_MERGE_KEY.get(owner) if not key: # ★facet 은 여기 자리가 없다. base 갈래로 **우회 등록 안 한다** — # `grounding_facet_binding` 이 제 자리(또는 빚)로 보낸다. skipped.append({**base, "skip": SKIP_NO_MERGE_KEY, "note": "base 갈래로 우회 등록하지 않는다"}) continue # ★★신원은 **장부가 이미 정했다.** 새로 발급하면 두 벌이 된다. # ★비었거나 접두가 어긋난 것은 `validate_ledger` 가 **이미 세웠다** — # 여기서 다시 거르면 규칙이 두 곳이 된다. sid = str(rec["final_id"]) seen = used.setdefault(key, set()) if sid in seen: raise ShortIdCollision( f"{sid} 가 이미 있다 ({key}). 장부가 정한 신원이라 **다시 매기지 " "않는다** — 다시 매기면 참조 의무·하류 결속이 엉뚱한 행을 " "가리킨다. 두 벌이 선 상태이니 사람이 봐야 한다") seen.add(sid) out.setdefault(key, []).append({ "short_id": sid, "name": _clean(r.get("surface_form")), # ★비워 둔다 — `entity_detail` 이 만들어 덮는다. 여기 겉모습 글을 # 넣으면 **단일 시각 형태 계약 없이** T2I 로 들어간다. "description": "", "visual_traits": [], # ★★**왜 이 엔티티가 생겼는지**를 원형 그대로 남긴다. # 한 벌로 묶어 `description`·`visual_traits` 와 섞이지 않게 한다. "grounding_provenance": _provenance(lid, sid, owner, r, rec), }) return {"rows": out, "skipped": skipped, "contract": ADAPTER_CONTRACT_VERSION} # ───────────────────────────────────────────────────────────────────── # ★★★엔티티와 후보를 **같은 reduced 행에서 함께** 낸다 (Codex · 09-01) # ───────────────────────────────────────────────────────────────────── def project_rows_and_candidates( reduced: Dict[str, Any], *, project_id: str, episode_id: str, existing: Optional[Dict[str, Sequence[Dict[str, Any]]]] = None, ) -> Dict[str, Any]: """등록된 행과 **그 행을 가리키는 후보**를 한 번에 낸다. ★왜 한 함수인가 — 둘을 따로 만들면 링크가 어느 행을 가리키는지 **다시 짝지어야** 하고, 그 짝짓기가 곧 이름 결속이 된다. 같은 reduced 행에서 **동시에** 내면 링크는 짝짓기가 아니라 **그 행의 사실**이다. ★후보에 실리는 `producer_link` 는 `grounding_carry.verified_link` 가 받는 바로 그 모양이다 — `{issuer, contract_version, final_id, local_id}`. 두 좌표를 **다** 싣는다: 받는 쪽이 「낸 좌표는 대상에도 있어야 한다」로 보므로, 한쪽만 실으면 대상에 그 칸이 없을 때 거절된다. ★**결정적**이다 — 같은 입력이면 같은 차례·같은 내용. ★이 함수는 **아무 스텝도 안 부른다**(adapter 는 아직 inert). manifest 배선은 D cutover 몫이다. Returns: ``{"rows": …, "skipped": …, "candidates": [...], "contract": …}`` """ # ★id 는 **production 발급 함수**가 낸다 — 여기서 지어내면 두 벌이 된다. from app.modules.pipeline.grounding_subject import build_subject # ★★facet 결속의 **구조 좌표**를 같은 reduced 에서 함께 낸다 — 따로 # 만들면 어느 행의 것인지 다시 짝지어야 하고 그것이 곧 이름 결속이다. from app.modules.pipeline import grounding_facet_binding as _fb facet = _fb.bind(list((reduced or {}).get("rows") or ()), list((reduced or {}).get("part_of") or ()), dict((reduced or {}).get("registered") or {})) by_facet = {str(b.get("local_id") or ""): b for b in (facet.get("bindings") or ())} got = to_entity_rows(reduced, existing=existing) by_local = {} for _key, rows in (got.get("rows") or {}).items(): for r in rows: pv = r.get("grounding_provenance") or {} lid = str(pv.get("local_id") or "") if lid: by_local[lid] = (r, pv) cands: List[Dict[str, Any]] = [] for r in (reduced or {}).get("rows") or (): lid = str((r or {}).get("local_id") or "") hit = by_local.get(lid) # ★★★후보를 내는 문은 **장부의 `registered` 하나**다 (Codex · 09-01). # owner 로 가르면 heuristic 이 된다 — 정본은 `reduce_episode` 가 낸 # `registered[local_id].registered` 다. # # registered is True + entity row 있음 → 링크 달아 낸다 # registered is True + entity row 없음 → 증거만(링크 없이). # `outlook` 이 그 자리다 # False / None → **후보 0**. owner 무관. # # ★그 밖의 미등록 행에 후보를 내면 `promoted` 로 승격되어 **없던 참조 # 의무가 생긴다** — 실측: 등록 1인데 의무 2. rec = ((reduced or {}).get("registered") or {}).get(lid) or {} if rec.get("registered") is not True: continue row, pv = hit if hit else ({}, {}) occ = list(r.get("occurrences") or ()) first = (occ[0] if occ else {}) or {} owner = str(pv.get("owner_type") or (r or {}).get("owner_type") or "") surface = str(r.get("surface_form") or "") # ★★★anchor 를 **정본 span 좌표**로 만든다 (Codex BLOCK · 09-01). # 앞 판은 `segment_id` 만 써서, 같은 씬 안의 **같은 owner·같은 표면형** # 두 행이 서로 다른 span·`local_id`·`final_id` 를 가져도 **같은 # `research_subject_id`** 를 받았다. 같은 이름의 다른 것은 한 씬에도 # 있다. ★이름을 해석하지 않는다 — 좌표를 그대로 붙일 뿐이다. anchor = _span_anchor(first.get("source_span") or {}) if not anchor: # ★빈 anchor 로 후보를 만들지 않는다 — 그것이 곧 id 충돌의 씨앗이다 raise MissingSourceSpan( f"{lid} 에 원문 좌표가 없다 — 빈 anchor 로 후보를 만들면 " "같은 이름의 다른 것과 신원이 겹친다") cands.append({ **build_subject(project_id=project_id, episode_id=episode_id, source_anchor=anchor, surface_form=surface, owner_type=owner, provenance={"source_step": "grounding_chunk"}), "owner_type": owner, "surface_form": surface, "source_anchor": anchor, "source_quote": str(first.get("source_quote") or ""), "planned_occurrences": len(occ), "why_candidate": str(pv.get("visual_brief") or ""), # ★있는 것만 싣는다 — 없는 좌표를 만들지 않는다 **({"occurrences": copy.deepcopy(occ)} if occ else {}), # ★★producer 가 **유료로 낸 구조화 판정**을 원형 그대로 나른다. # 원문 증거와 **섞지 않는다** — 한 덩어리로 따로 싣는다. **({_ec.PRODUCER_PAYLOAD: _pl} if (_pl := _ec.producer_payload(r)) else {}), # ★★facet 좌표 — **덩어리 그대로**. 없으면(빚이면) 칸도 안 만든다. # ★맥락 칸도 **그대로** 나른다 — 의무 계획이 이것으로 맥락 # 조사 재료를 얻는다. 없으면 칸도 안 만든다. **({_ec.HOST_CONTEXT: copy.deepcopy(r["host_context"])} if isinstance(r.get("host_context"), dict) else {}), **({_ec.FACET_BINDING: copy.deepcopy(_fbrow)} if (_fbrow := by_facet.get(lid)) else {}), # ★★**링크** — 가리킬 행이 있을 때만, 두 좌표를 다 싣는다 **({LINK_KEY: { "issuer": ADAPTER_ISSUER, "contract_version": ADAPTER_CONTRACT_VERSION, "final_id": str(row.get("short_id") or ""), "local_id": lid, }} if row.get("short_id") else {}), }) # ★★emitter 스스로 **신원이 겹치지 않는지** 본다 — 받는 쪽이 세우기 전에. ids = [str(c.get("research_subject_id") or "") for c in cands] dup = sorted({i for i in ids if ids.count(i) > 1}) if dup: raise CandidateIdCollision( f"후보 신원이 겹친다: {dup} — 같은 좌표의 행이 둘이라는 뜻이다. " "조용히 둘로 만들지 않는다") return {**got, "candidates": cands, "facet_debt": facet.get("debt") or []} def _span_anchor(span: Dict[str, Any]) -> str: """정본 span 좌표. ★`segment_id` 만으로는 **한 씬 안에서 겹친다**.""" seg = str((span or {}).get("segment_id") or "").strip() start, end = (span or {}).get("start"), (span or {}).get("end") if not seg or start is None or end is None: return "" return f"{seg}:{start}-{end}"