"""앞단이 남긴 outlook **의무·증거**를 phase3 산출에 **구조로** 잇는다. ## 왜 뒤에서 하나 스텝 차례가 이유다 (실측 2026-09-01) — grounding_screen 13.68 ← 의무·증거를 정한다 reference_acquisition 13.8 outlook_phase1~3 19 ~ 19.2 ← ★outlook **실물**이 여기서 생긴다 즉 앞단이 참조를 살 때 outlook 엔티티는 **아직 없다**. 그래서 앞단은 **의무와 증거만** 남기고, 실제 결속은 여기서 한다. ★참조를 **사지 않는다** — 구매자는 중앙 획득 한 곳이다. ## 무엇으로 잇나 ★**이름도 부분문자열도 안 쓴다.** 앞단 줄과 phase3 산출이 **공유하는 구조**만 본다 — 지금 그것은 `scene_assignments` 의 `(scene_index, character_id, outlook_id)` 조합이고, 앞단 줄에는 `source_evidence.occurrences` 의 `segment_id` 가 있다. 두 좌표가 **결정적으로** 같은 것을 가리킬 때만 잇는다. 한 source id → 최대 한 outlook 한 outlook ← merge 된 여러 source id 허용 그 밖(기대 부모 없음·그 씬에 그 부모 없음·둘 이상·enum 밖) → **unresolved** ★★★**씬이 같다는 것은 소유가 아니다** (Codex BLOCK · 09-01). 앞 판은 occurrence 의 씬에 배정이 하나뿐이면 그 배정에 붙였다 — 그러면 그 씬의 다른 인물(`C02`) 몫 근거가 `C01` 것으로 **잘못 붙는다**. 그래서 앞단이 낸 **기대 부모 좌표**(`grounding_facet_binding.parent_final_id`)를 들고 와, 그 씬의 배정 중 **`character_id` 가 그 부모와 같은 것만** 후보로 본다. 좌표가 없으면 **짐작하지 않고 `unresolved`** 다. ## 못 이으면 ★사람을 기다리지 않는다. `reference_unavailable` 로 적고 참조 없이 내려간다 — 그것이 이 판의 확정 정책이다(HITL 0). """ from __future__ import annotations import copy from typing import Any, Dict, List, Optional, Sequence # ★★★갈래 중립인 것은 **여기서 정의하지 않는다** (Codex BLOCK · 09-01). # 회계·자리 가름·상태 이름은 중앙 입구 계약 한 벌이고, 이 모듈은 outlook 을 # **잇는 법**만 갖는다. 앞 판은 이것들을 여기 적어 두어서 중앙 입구가 # outlook 전용이 됐다 — location·location_part 가 들어올 길이 없었다. from app.modules.pipeline.grounding_acquisition_ledger import ( # noqa: E402,F401 LANE_AUTO_DONE, LANE_BUY, LANE_NOT_APPLICABLE, STATUSES, UNRESOLVED, accounting, acquisition_targets, auto_completed, known_screens, lane_of, unresolved_outcome) from app.modules.pipeline.grounding_acquisition_ledger import ( # noqa: E402 LedgerContractError) from app.modules.pipeline.grounding_acquisition_ledger import ( # noqa: E402 RESOLVED as BOUND) #: ★앞단이 싣는 **facet 좌표 덩어리**의 칸 이름. 계약에서 온다. #: 기대 부모는 그 안의 `parent_final_id` 다 — **따로 실은 값은 안 받는다**. from app.modules.pipeline.grounding_entity_contract import ( # noqa: E402 FACET_BINDING) #: 못 정한 까닭. WHY_NO_PARENT = "no_expected_parent_coordinate" WHY_PARENT_NOT_IN_SCENE = "expected_parent_not_assigned_here" WHY_NO_MATCH = "no_structural_match" WHY_AMBIGUOUS = "more_than_one_outlook" WHY_NOT_IN_CATALOG = "outlook_id_not_in_catalog" CONTRACT_VERSION = "1.202609011300" class OutlookBindingError(LedgerContractError): """outlook 을 **잇는 자리**의 계약 위반. ★조용히 넘어가지 않는다. ★중앙 입구 계약(`LedgerContractError`)의 갈래다 — 부르는 쪽이 둘 중 무엇을 잡아도 장부 계약 위반은 다 걸린다. """ def catalog(phase3: Optional[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: """phase3 산출의 **runtime enum** — `outlook_id` → 그 항목. ★코드에 이름을 안 적는다. 무엇이 있는지는 **산출이 정한다**. """ got: Dict[str, Dict[str, Any]] = {} for ol in ((phase3 or {}).get("outlooks") or ()): if not isinstance(ol, dict): continue oid = str(ol.get("outlook_id") or ol.get("short_id") or "").strip() if not oid: continue if oid in got: raise OutlookBindingError( f"phase3 에 `{oid}` 가 두 번 있다 — 어느 것인지 못 정한다") got[oid] = ol return got def _segments_of(row: Dict[str, Any]) -> set: """앞단 줄이 가리키는 **원문 구간들**. ★없으면 빈 집합이다.""" ev = (row or {}).get("source_evidence") or {} out = set() for o in (ev.get("occurrences") or ()): seg = str(((o or {}).get("source_span") or {}).get("segment_id") or "") if seg: out.add(seg) return out def _scene_key(scene_index: Any) -> str: """씬 번호 → 구간 id. ★한 곳에서만 만든다 — 계약 모듈이 그 곳이다.""" from app.modules.pipeline.grounding_entity_contract import scene_key return scene_key(scene_index) def assignments_index(phase3: Optional[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]: """구간 id → 그 씬의 배정들. ★`scene_assignments` 를 **그대로** 읽는다.""" out: Dict[str, List[Dict[str, Any]]] = {} for sa in ((phase3 or {}).get("scene_assignments") or ()): if not isinstance(sa, dict): continue key = _scene_key(sa.get("scene_index")) for a in (sa.get("assignments") or ()): if not isinstance(a, dict): continue oid = str(a.get("outlook_id") or "").strip() cid = str(a.get("character_id") or "").strip() if not oid: continue out.setdefault(key, []).append({"outlook_id": oid, "character_id": cid}) return out def bind(rows: Sequence[Dict[str, Any]], phase3: Optional[Dict[str, Any]]) -> Dict[str, Any]: """앞단 outlook 줄들 → **결속 장부**. ★참조는 사지 않는다. Args: rows: `grounding_screen` 이 낸 정본 행 중 `owner_type == "outlook"` 인 것들. 각 행은 `source_evidence`(원문 증거)와, 있으면 `grounding_producer_payload`(producer 판정)를 갖는다. phase3: `outlook_phase3` 체크포인트의 `data`. Returns: ``{"contract_version": …, "rows": [...], "counts": {...}}`` 각 줄은 `status`(bound/unresolved)와 **왜 그런지**를 갖는다. ★못 이은 것도 **한 줄씩 남는다** — 조용히 사라지지 않는다. """ cat = catalog(phase3) idx = assignments_index(phase3) out: List[Dict[str, Any]] = [] for r in rows or (): rsid = str((r or {}).get("research_subject_id") or "") base: Dict[str, Any] = { "research_subject_id": rsid, "owner_type": str((r or {}).get("owner_type") or ""), # ★증거와 판정을 **원형 그대로** 들고 간다 — 중앙 획득이 쓴다. "source_evidence": copy.deepcopy((r or {}).get("source_evidence") or {}), # ★못 이은 줄도 **칸은 있다** — 없으면 소비하는 쪽이 KeyError 로 # 깨지고, 「없다」와 「칸이 없다」가 뒤섞인다 "final_id": None, "parent_final_id": None, } payload = (r or {}).get("grounding_producer_payload") if payload: base["grounding_producer_payload"] = copy.deepcopy(payload) # ★★★판별이 **무엇이라 했는지**를 원형 그대로 들고 간다 (Codex · 09-01). # 버리면 「참조 불필요」로 판정된 것까지 사게 된다 — 실제로 그랬다. base["screen"] = str((r or {}).get("screen") or "") if (r or {}).get("reason"): base["screen_reason"] = str(r["reason"]) # ★★앞단이 실은 **facet 좌표 덩어리**에서 읽는다 — 한 칸만 떼서 들고 # 오면 출처와 계약을 되짚을 수 없다 (Codex · 09-01). fb = (r or {}).get(FACET_BINDING) or {} # ★★★**뒷문을 없앴다** (09-01). 앞서는 `expected_parent_final_id` 를 # 직접 실어도 받았는데, 그 자리가 곧 시험이 값을 **지어 넣는** 문이다 # — 실제로 그렇게 통과했고 production 경로에서는 결속 0이었다. # 이제 좌표의 출처는 **facet 좌표 덩어리 하나**뿐이다. want_parent = str(fb.get("parent_final_id") or "").strip() if fb: base[FACET_BINDING] = copy.deepcopy(fb) base["expected_parent_final_id"] = want_parent or None if not want_parent: # ★★기대 부모가 없으면 **씬만으로 짐작하지 않는다** out.append({**base, "status": UNRESOLVED, "why": WHY_NO_PARENT}) continue segs = _segments_of(r) in_scene: List[Dict[str, Any]] = [] for seg in sorted(segs): in_scene.extend(idx.get(seg) or ()) # ★그 씬의 배정 중 **기대 부모의 것만** hits = [h for h in in_scene if h["character_id"] == want_parent] if in_scene and not hits: out.append({**base, "status": UNRESOLVED, "why": WHY_PARENT_NOT_IN_SCENE, "candidates": sorted({h["character_id"] for h in in_scene})}) continue oids = sorted({h["outlook_id"] for h in hits}) if not oids: out.append({**base, "status": UNRESOLVED, "why": WHY_NO_MATCH}) continue if len(oids) > 1: out.append({**base, "status": UNRESOLVED, "why": WHY_AMBIGUOUS, "candidates": oids}) continue oid = oids[0] if oid not in cat: out.append({**base, "status": UNRESOLVED, "why": WHY_NOT_IN_CATALOG, "candidates": [oid]}) continue # ★여기 온 것은 **전부 기대 부모의 배정**이다(위에서 걸렀다) — # 「같은 씬에 다른 인물이 있다」는 이제 결속을 흐리지 못한다. # ★★공통 입구 모양으로 낸다 — `final_id`/`parent_final_id` 가 # 중앙 계약의 이름이다. outlook 전용 이름은 **감사용으로 같이** 둔다 # (Codex BLOCK · 09-01: resolver 는 갈래별이되 나온 모양은 하나다). out.append({**base, "status": BOUND, "final_id": oid, "parent_final_id": want_parent, "outlook_id": oid, "character_id": want_parent}) counts = {s: sum(1 for x in out if x["status"] == s) for s in (BOUND, UNRESOLVED)} return {"contract_version": CONTRACT_VERSION, "rows": out, "counts": counts}