"""판별 장부 한 줄 → **중앙 획득이 받는 모양**. ★사지 않는다 · 만들지 않는다. ## 왜 있나 `reference_acquisition_rounds.acquire_one` 은 두 자리를 **따로** 받는다 — target 검색 **저작**용 {subject_id, directive_native, terms_native, language_lock_native} narrow_hint 종류 **판정**용 VLM 에게 「무엇인가」를 묻는 좁힘 문안 C(c) 가 유료로 낸 판정(`grounding_producer_payload`)에는 두 쪽 재료가 **같이** 들어 있다. 그것을 아무렇게나 넘기면 — ★`visual_brief` 가 판정으로 가면 VLM 이 「무엇인가」 대신 「어떻게 생겼나」로 답한다. 그러면 우리가 고른 것은 **종류가 맞는 사진**이 아니라 **글과 닮은 사진**이다. 그래서 계약이 정한 두 표(`JUDGE_FIELDS` · `SEARCH_FIELDS`)로 **갈라서** 만든다. 이 모듈이 그 갈림의 **유일한 자리**다. ★이 모듈은 **검색도 다운로드도 판정도 안 한다.** 모양만 만든다. """ from __future__ import annotations import copy from typing import Any, Dict, List, Optional, Sequence from app.modules.pipeline import grounding_entity_contract as _ec CONTRACT_VERSION = "1.202609011400" class MissingSearchDirective(RuntimeError): """무엇을 찾을지가 없다. ★빈 질의로 안 산다 — 아무거나 온다.""" def _payload(row: Dict[str, Any]) -> Dict[str, Any]: return dict((row or {}).get(_ec.PRODUCER_PAYLOAD) or {}) def search_target(row: Dict[str, Any]) -> Dict[str, Any]: """검색 **저작**으로 나가는 것. ★`SEARCH_FIELDS` 밖은 **한 칸도** 안 넣는다. Raises: MissingSearchDirective: 저작 재료가 없다. 빈 질의로 사지 않는다. """ p = _payload(row) directive = str(p.get("visual_brief") or "").strip() terms = [str(t) for t in (p.get("search_terms_native") or ()) if str(t)] if not directive and not terms: raise MissingSearchDirective( f"{row.get('research_subject_id')!r} 에 검색 저작 재료가 없다 — " "빈 질의로 찾으면 아무거나 온다") got = { "subject_id": str(row.get("research_subject_id") or ""), "directive_native": directive, "terms_native": terms, "language_lock_native": str(p.get("language_lock_native") or ""), # ★★★검색 **지시문 저작기**가 읽는 칸들 (2026-09-02 Codex BLOCK). # 앞 판은 네 칸만 냈고, 그래서 저작기가 첫 대상에서 # `BriefInputsMissing` 으로 **검색 전에 죽었다** — 시험이 손으로 # 만든 target 을 써서 그 자리를 안 지났다. # ★이것은 「판정으로 나가는 것」이 아니다 — 심판에게는 여전히 # `coarse_type_label` 하나만 간다(`judge_hint`). "owner_type": str(row.get("owner_type") or ""), "coarse_type_label": str(p.get("coarse_type_label") or ""), "surface_form": str(p.get("surface_form") or row.get("surface_form") or ""), "visual_brief": str(p.get("visual_brief") or ""), # ★조사 저작기(`grounding_target_research`)가 「그게 무엇인가」를 알아내는 근거 — 원고 # 문장 **전문**(자르지 않는다). 검색어·신원에는 안 실린다(뼈대만이 신원). "source_quotes": _quotes_of(row), } return got def _quotes_of(row: Dict[str, Any]) -> List[str]: """줄의 원고 근거 문장들 — `source_evidence.source_quote` 와 표기. 없으면 빈 목록.""" se = (row or {}).get("source_evidence") or {} out: List[str] = [] for key in ("surface_form", "source_quote"): v = str(se.get(key) or "").strip() if v and v not in out: out.append(v) for occ in (se.get("occurrences") or ()): q = str(((occ or {}).get("source_quote")) or ((occ or {}).get("quote")) or "").strip() if q and q not in out: out.append(q) return out def judge_hint(row: Dict[str, Any]) -> Optional[str]: """종류 **판정**으로 나가는 것. ★`coarse_type_label` **하나뿐**이다. 없으면 `None` — 부르는 쪽이 기본 좁힘 문안을 쓴다. ★없다고 `visual_brief` 로 **대신하지 않는다**. 그것이 정확히 금지된 자리다. """ got = str(_payload(row).get("coarse_type_label") or "").strip() return got or None def reads_only(fn: Any, allowed: Sequence[str]) -> List[str]: """그 함수가 payload 에서 **읽는 칸들** 중 허용 밖. ★AST 로 본다. ★★★값으로 견주면 안 된다 (실측 09-01): producer 가 같은 낱말을 종류 이름과 검색어에 **둘 다** 쓸 수 있다 — 그러면 멀쩡한 것이 「샜다」로 잡힌다. 그것이 바로 「글자로 뜻을 판단」하는 자리다. 그래서 **읽는 자리**를 본다. """ import ast import inspect import textwrap tree = ast.parse(textwrap.dedent(inspect.getsource(fn))) read = set() for n in ast.walk(tree): if (isinstance(n, ast.Call) and getattr(n.func, "attr", "") == "get" and n.args and isinstance(n.args[0], ast.Constant)): k = n.args[0].value if k in _ec.PRODUCER_PAYLOAD_FIELDS: read.add(k) return sorted(read - set(allowed)) class NotBuyable(RuntimeError): """살 자격이 없는 줄이 왔다. ★두 문을 **여기서도** 지킨다.""" def assert_buyable(row: Dict[str, Any]) -> None: """이 줄이 **살 자격**이 있나 — 결속됐고 판별이 **의무**라고 했나. ★★★adapter 가 임의의 줄을 받으면 앞의 두 문을 **우회한다** (Codex NON-BLOCK · 09-01). 그러면 「참조 불필요」로 판정된 것도 모양만 만들어져 나간다. 그래서 **여기서도 fail-closed** 로 본다 — 문은 두 곳에 있어도 되지만 **판정은 한 곳**(`grounding_acquisition_ledger`)에서 온다. ★★★갈래 중립이다 (Codex BLOCK · 09-01). 앞 판은 `grounding_outlook_binding` 을 직접 읽어서 중앙 입구가 **outlook 전용** 이었다 — location·location_part 가 들어올 길이 없었다. """ from app.modules.pipeline import grounding_acquisition_ledger as gl from app.modules.pipeline.grounding_screen import SCREEN_OBLIGATION if row.get("status") != gl.RESOLVED: raise NotBuyable( f"{row.get('research_subject_id')!r} 는 해소가 " f"{row.get('status')!r} 다 — 무엇에 붙일지 모르는 참조는 안 산다") if str(row.get("screen") or "") != SCREEN_OBLIGATION: raise NotBuyable( f"{row.get('research_subject_id')!r} 의 판별은 " f"{row.get('screen')!r} 다 — 살 까닭이 없다") def acquisition_inputs(rows: Sequence[Dict[str, Any]]) -> Dict[str, Any]: """줄들 → 중앙 획득이 그대로 쓸 입력. ★못 만드는 줄은 **남긴다**. ★들어온 줄마다 `assert_buyable` 을 **먼저** 지난다 — 자격 없는 줄은 `refused` 로 남고 **모양조차 안 만든다**. Returns: ``{"targets": [{"target": …, "narrow_hint": …}], "skipped": [...]}`` `skipped` 는 저작 재료가 없어 **안 산 것**이다 — 조용히 사라지지 않는다. ★★세 갈래 모두 **원행(`ledger_row`)을 그대로** 들고 간다. 앞 판은 `research_subject_id` 와 사유만 남겨서, 나중 수동 수정 화면이 screen· producer payload·facet 좌표를 **복구할 수 없었다** (Codex 2026-09-01). """ targets: List[Dict[str, Any]] = [] skipped: List[Dict[str, Any]] = [] refused: List[Dict[str, Any]] = [] for r in rows or (): try: assert_buyable(r) except NotBuyable as exc: refused.append({"research_subject_id": str((r or {}).get("research_subject_id") or ""), "why": str(exc), "ledger_row": copy.deepcopy(r)}) continue try: t = search_target(r) except MissingSearchDirective as exc: skipped.append({"research_subject_id": str((r or {}).get("research_subject_id") or ""), "why": str(exc), "ledger_row": copy.deepcopy(r)}) continue targets.append({"target": t, "narrow_hint": judge_hint(r), # ★증거는 **그대로** 들고 간다 — 사람 검토 화면 몫이다 "source_evidence": copy.deepcopy( (r or {}).get("source_evidence") or {}), "ledger_row": copy.deepcopy(r)}) return {"contract_version": CONTRACT_VERSION, "targets": targets, "skipped": skipped, "refused": refused} def inputs_from_ledger(ledger: Dict[str, Any]) -> Dict[str, Any]: """**장부 하나에서** 중앙 획득 입력까지 — 한 공개 함수. ★★★왜 하나인가 (Codex NON-BLOCK · 09-01) — 부르는 쪽이 `acquisition_targets` 를 건너뛰고 `acquisition_inputs` 에 아무 줄이나 주면 두 구매 문을 **우회한다**. 문은 두 겹으로 두되(여기서도 `assert_buyable`), **정상 입구는 하나**여야 사람이 헷갈리지 않는다. Returns: `acquisition_inputs` 의 산출 + `auto_completed_rows`(사야 하는데 못 붙인 것)와 `not_applicable_rows`(애초에 살 것이 아니었던 것)의 **원행 그대로**, 그리고 그것에서 **파생한** 수. ★들어온 줄이 전부 어디로 갔는지 이 하나로 읽힌다. ★★앞 판은 이 둘을 **수로만** 냈다. 그러면 나중 수동 수정 화면이 어느 줄이었는지 못 찾는다 — 수는 행에서 **파생**한다 (Codex 2026-09-01). """ from app.modules.pipeline import grounding_acquisition_ledger as gl acc = gl.accounting(ledger) # ★합이 안 맞으면 여기서 선다 got = acquisition_inputs(gl.acquisition_targets(ledger)) got["auto_completed_rows"] = copy.deepcopy(acc["lanes"][gl.LANE_AUTO_DONE]) got["not_applicable_rows"] = copy.deepcopy( acc["lanes"][gl.LANE_NOT_APPLICABLE]) # ★어느 갈래가 실제로 들어왔나 — 「받는 쪽만 있고 내는 쪽이 없는」 것을 # 여기서 드러낸다 (Codex · 09-01) got["owners_present"] = list(gl.owners_present(ledger)) got["auto_completed"] = len(got["auto_completed_rows"]) got["not_applicable"] = len(got["not_applicable_rows"]) got["ledger_rows"] = acc["rows"] return got