"""조사 결과 → **묶음 멤버**. ★여기서 의무를 만들지 않는다. ## 무엇을 하나 / 안 하나 Codex BLOCK (2026-09-01) — 「bundle projection 이 acquisition 완료 뒤 context 멤버를 **발명하면 안 됩니다**. 중앙 참조 조사 **전에** reference obligation plan 을 만드십시오.」 앞 판 조사 끝난 뒤 맥락 멤버를 지어냈다 → 같은 조사 결과·같은 사진에 쓰임만 다시 적었다. 부모가 비대상이면 복구도 못 했다. 지금 의무는 `grounding_reference_obligations.plan` 이 **조사 앞에서** 세운다. 여기는 그 결과를 **옮기기만** 한다. ## 못 구한 것도 남긴다 `reference_unavailable` 인 의무는 이미지 required 를 안 만들지만, **논리 의무와 처지는 durable 하게 남는다** — 나중 UI 가 「무엇이 없었는지」를 찾을 수 있어야 한다 (Codex · 09-01). """ from __future__ import annotations import copy from typing import Any, Dict, List, Optional, Sequence from app.modules.pipeline import grounding_entity_contract as _gc from app.modules.pipeline.grounding_host_context import PURPOSE_DETAIL #: ★2026-09-01 판 3 — 아웃룩 갈래가 **인물 아웃룩 자리**로 간다. #: ★2026-09-02: **고증 축**이 소비 문에 들어왔다 — 판을 올린다. 옛 판이 #: 섞이면 확인 안 된 사진이 sidecar 로 붙는다. PROJECTION_CONTRACT_VERSION = "4.202609020800" #: ★이 모듈이 **배경 자리**를 집행한다 (장소·장소부분). ENFORCES_GATE = _gc.ENFORCE_BY_BACKGROUND_SIDECAR class ProjectionError(RuntimeError): """묶음으로 옮길 수 없다. ★조용히 건너뛰지 않는다.""" def _row_field(row: Dict[str, Any], name: str) -> str: return str((row.get("ledger_row") or {}).get(name) or "") def _purpose(row: Dict[str, Any]): return (row.get("ledger_row") or {}).get("purpose") def members_from_rows(rows: Sequence[Dict[str, Any]], *, content_sha_of, coordinate_of) -> List[Dict[str, Any]]: """중앙 조사 결과 줄들 → `plan_bundle` 멤버들. ★쓰임은 **의무가 정했다**. Args: rows: `grounding_central_acquisition.run(...)["rows"]`. 그 `ledger_row` 는 `grounding_reference_obligations.plan` 이 낸 의무다. content_sha_of: `(row) -> str|None`. 고른 사진의 내용 해시. ★이 모듈이 파일을 안 읽는다 — 읽는 자리는 부르는 쪽이다. coordinate_of: `(row) -> {"source": …, "asset_id"|"path": …}`. Returns: 멤버 목록. ★`selected` 가 아닌 것도 **논리 의무로 남는다** — `plan_bundle` 이 이미지 required 를 안 만들 뿐이다. """ from app.modules.pipeline import grounding_reference_bundle as rb out: List[Dict[str, Any]] = [] for r in rows: purpose = _purpose(r) if not purpose: continue # ★배경 자리가 아닌 갈래다 subject = _row_field(r, "final_id") if not subject: raise ProjectionError( f"{r.get('research_subject_id')!r} 의무에 `final_id` 가 없다") covers = list((r.get("ledger_row") or {}).get("covers") or [subject]) outcome = str(r.get("outcome") or "") base = { "subject_final_id": subject, "purpose": str(purpose), "covers": covers, "outcome": outcome, # ★못 구한 까닭을 그대로 — 나중 UI 가 이것을 읽는다 "why": str(r.get("why") or ""), "why_unbought": r.get("why_unbought"), "obligation_subject_id": str(r.get("research_subject_id") or ""), "context_owner": (r.get("ledger_row") or {}).get("context_owner"), } if not _usable(r): # ★붙일 사진이 없다 — **신원도 해시도 안 지어낸다** out.append({**base, "member_identity": None, "content_sha256": None, **_why_not(r)}) continue out.append({**base, **_selected(r, subject, str(purpose), content_sha_of, coordinate_of)}) return out def _usable(row: Dict[str, Any]) -> bool: """이 줄을 **붙여도 되나**. ★뜻을 여기서 다시 적지 않는다. ★★★실측 (Codex 재리뷰 2026-09-02): 집행자가 **둘**인데 정책 lane 하나만 막아 두고 「붙는 것 0」이라고 보고했다. 배경 sidecar 는 `outcome == selected` 만 보고 그대로 붙이고 있었다 — 아무도 안 본 사진이 그림의 근거가 된다. 판단은 `reference_acquisition` **한 곳**이 한다. """ from app.modules.pipeline import reference_acquisition as ra return ra.usable_as_reference(row) def _why_not(row: Dict[str, Any]) -> Dict[str, Any]: """왜 안 붙었나 — **못 구한 것**과 **아직 안 본 것**을 가른다.""" from app.modules.pipeline import reference_acquisition as ra if str(row.get("outcome") or "") != ra.STATUS_SELECTED: return {"not_attached": None} return {"not_attached": ra.FIDELITY_UNVERIFIED, "fidelity": ra.fidelity_of(row), "★means": ("거친 종류로는 골랐지만 **그 시대·그 지역 것인지는 " "아무도 안 봤다**. 논리 의무로만 남는다")} def _selected(row: Dict[str, Any], subject: str, purpose: str, content_sha_of, coordinate_of) -> Dict[str, Any]: from app.modules.pipeline.grounding_host_context import member_identity ident = str(row.get("identity") or "") if not ident: raise ProjectionError( f"{subject}/{purpose} 에 획득 신원이 없다 — 무엇으로 조사했는지가 " "값에 안 실린다") sha = content_sha_of(row) if not sha: raise ProjectionError( f"{subject}/{purpose} 가 selected 인데 내용 해시가 없다") return { "member_identity": member_identity( subject_final_id=subject, purpose=purpose, acquisition_identity=ident), "content_sha256": sha, **coordinate_of(row), "parent_final_id": (_row_field(row, "parent_final_id") or None), } def outlook_ref_id(row: Dict[str, Any]) -> Optional[str]: """아웃룩 줄 → 기존 **인물 아웃룩 참조 신원** `C##O##`. 못 만들면 `None`. ★★★**검증된 결속에서만** 만든다 (Codex 2026-09-01) — 이름·부분문자열 0. `grounding_outlook_binding` 이 이미 phase3 로 확인한 `parent_final_id` (인물)와 `final_id`(아웃룩)를 **그대로 잇는다**. 둘 중 하나라도 없거나 갈래가 안 맞으면 **안 지어낸다**. """ from app.modules.pipeline.grounding_entity_contract import ( owner_of_final_id) led = row.get("ledger_row") or {} if str(led.get("owner_type") or "") != "outlook": return None oid = str(row.get("final_id") or led.get("final_id") or "") cid = str(row.get("parent_final_id") or led.get("parent_final_id") or "") if not oid or not cid: return None if owner_of_final_id(oid) != "outlook": return None if owner_of_final_id(cid) != "character": return None return f"{cid}{oid}" def outlook_members_from_rows(rows: Sequence[Dict[str, Any]], *, required_ref_ids: Sequence[str], content_sha_of, coordinate_of) -> List[Dict[str, Any]]: """아웃룩 조사 결과 → **인물 아웃룩 자리**에 붙을 멤버들. Args: required_ref_ids: 이 컷의 카드가 **이미 요구하는** `C##O##` 들. ★★이것이 「그 복장이 이 컷에서 실제로 쓰이나」의 정본이다 — 조사 결과가 **새 짝을 만들지 않는다** (Codex 2026-09-01). 카드에 없는 짝이면 그 사진은 이 컷에 안 붙는다. ★새 슬롯도 새 구매자도 만들지 않는다 — 기존 `character_outlook` 자리에 **덧붙이고**, 값이 `grounding:…` 앞머리라 기존 합성본과 안 겹친다. 그래서 합성본이 붙어 있다고 고증 참조가 채워진 것이 되지 않는다. """ from app.modules.pipeline import grounding_reference_bundle as rb want = {str(x) for x in (required_ref_ids or ()) if str(x or "").strip()} out: List[Dict[str, Any]] = [] for r in rows: ref_id = outlook_ref_id(r) if not ref_id or ref_id not in want: continue # ★이 컷이 안 쓰는 복장이다 outcome = str(r.get("outcome") or "") base = { "subject_final_id": ref_id, # ★그 옷 **자체의 사진**이다 — 장소부분 상세와 같은 뜻이라 # 쓰임 어휘를 새로 만들지 않는다. "purpose": PURPOSE_DETAIL, "covers": [ref_id], "outcome": outcome, "why": str(r.get("why") or ""), "why_unbought": r.get("why_unbought"), "obligation_subject_id": str(r.get("research_subject_id") or ""), "context_owner": None, rb.MEMBER_KIND_KEY: rb.OUTLOOK_KIND, rb.MEMBER_ROLE_KEY: rb.OUTLOOK_ROLE, } if not _usable(r): out.append({**base, "member_identity": None, "content_sha256": None, **_why_not(r)}) continue out.append({**base, **_selected(r, ref_id, PURPOSE_DETAIL, content_sha_of, coordinate_of)}) return out def unavailable_members(members: Sequence[Dict[str, Any]] ) -> List[Dict[str, Any]]: """★**공개 끝점** — 붙을 사진이 없는 논리 의무들. ★2026-09-02: **「골랐지만 아직 아무도 안 본 것」은 여기 안 든다** — 갈래가 다르다(`unverified_members`). 못 구한 것과 섞으면 사람 검토 화면이 무엇을 봐야 하는지 못 고른다. """ from app.modules.pipeline import grounding_reference_bundle as rb return [copy.deepcopy(m) for m in members if str(m.get("outcome") or "") != rb.OUTCOME_SELECTED] def unverified_members(members: Sequence[Dict[str, Any]] ) -> List[Dict[str, Any]]: """★**공개 끝점** — 골랐지만 **고증을 아무도 안 본** 것들. 못 구한 것과 **섞지 않는다** — 사람 검토 화면이 볼 것은 이쪽이다. """ from app.modules.pipeline import reference_acquisition as ra return [copy.deepcopy(m) for m in members if m.get("not_attached") == ra.FIDELITY_UNVERIFIED]