"""요소 리스팅 — 씬 묶음 체이닝 방식.

씬을 3000자 이하로 묶어 LLM에 순차 요청하고,
이전 결과를 체이닝하여 중복 없이 누적.
앞쪽 씬 2000자를 참조로 제공하여 문맥 유지.
"""
import logging
from typing import Any, Dict, List, Optional

from app.modules.llm.llm_client import call_structured
from app.modules.prompt_loader import load_prompt, load_schema

logger = logging.getLogger(__name__)
_MODULE = "entity_all"

# 씬 묶음 / 참조 크기 (env로 오버라이드 가능하도록 모듈 상수)
BUNDLE_TARGET = 3000   # 분석 대상 씬 묶음 최대 글자 수
REF_MAX = 2000         # 앞쪽 참조 씬 최대 글자 수


def list_entities_by_type(
    fulltext: str,
    entity_type: str,
    visual_rules: str = "",
    project_config: Optional[Dict] = None,
    opik_metadata: Optional[Dict] = None,
    scenes: Optional[List[Dict]] = None,
    a0_candidates: Optional[List[Dict]] = None,
    binding_out: Optional[Dict] = None,
    prior_roster: Optional[tuple] = None,
    carry_out: Optional[Dict] = None,
) -> List[Dict]:
    """씬 묶음 체이닝으로 요소 리스팅.

    ★``a0_candidates`` — shot 경로와 **같은 결속 계약**을 이 갈래에도 준다.
    한쪽만 주면 fallback 을 탄 프로젝트에서 결속이 통째로 없다.

    ★``prior_roster`` 도 **같은 이유로** 여기 있다 — 앞 화 명부를 shot 경로에만
    주면 이 갈래를 탄 프로젝트는 화마다 새 번호를 받는다.

    scenes가 있으면 씬 단위로 묶어서 체이닝.
    없으면 기존 방식(전문 1회 호출) fallback.
    """
    system = load_prompt(_MODULE, "system")
    type_prompt = load_prompt(_MODULE, entity_type)
    schema = load_schema(_MODULE, f"{entity_type}_schema")
    key = f"{entity_type}s" if entity_type != "prop" else "props"
    step_name = f"entity_all_{entity_type}"

    # 씬 데이터 없으면 기존 방식 fallback
    if not scenes:
        return _list_single_call(
            fulltext, system, type_prompt, schema, key,
            step_name, visual_rules, project_config, opik_metadata,
            a0_candidates=a0_candidates, entity_type=entity_type,
            binding_out=binding_out,
            prior_roster=prior_roster, carry_out=carry_out,
        )

    # 씬별 텍스트 준비
    scene_texts = []
    for seg in scenes:
        text = seg.get("text", "")
        scene_texts.append({
            "text": text,
            "len": seg.get("length", len(text)),
            "idx": seg.get("scene_index", 0),
        })

    # ★★결속 블록 (3b) — shot 경로와 **같은 함수**로 만든다.
    binding_block, _allowed = "", []
    _binding_out = binding_out if binding_out is not None else {}
    if a0_candidates:
        from app.modules.pipeline.grounding_binding import (
            build_binding_block, patch_schema_with_candidate_ids)

        binding_block, _allowed = build_binding_block(a0_candidates, entity_type)
        schema = patch_schema_with_candidate_ids(schema, _allowed)

    # ★★앞 화 명부 — shot 경로와 같은 계약. 비면 바이트 불변.
    roster_block, _prior_allowed = (prior_roster or ("", []))
    if _prior_allowed:
        from app.modules.pipeline.episode_carry import patch_schema_with_prior_ids

        schema = patch_schema_with_prior_ids(schema, _prior_allowed)

    all_entities: List[Dict] = []
    existing_names: set = set()
    calls = 0
    failed_calls = 0
    i = 0

    while i < len(scene_texts):
        # 씬 묶음 (BUNDLE_TARGET 이하)
        bundle = []
        bundle_len = 0
        while i < len(scene_texts) and bundle_len + scene_texts[i]["len"] <= BUNDLE_TARGET:
            bundle.append(scene_texts[i])
            bundle_len += scene_texts[i]["len"]
            i += 1
        # 씬 하나가 BUNDLE_TARGET보다 크면 그냥 포함
        if not bundle and i < len(scene_texts):
            bundle.append(scene_texts[i])
            bundle_len = scene_texts[i]["len"]
            i += 1

        # 앞쪽 씬 참조 (REF_MAX 이하)
        ref_text = ""
        ref_len = 0
        j = bundle[0]["idx"] - 2  # 번들 첫 씬 직전 인덱스 (0-based)
        ref_parts = []
        while j >= 0 and j < len(scene_texts) and ref_len + scene_texts[j]["len"] <= REF_MAX:
            ref_parts.insert(0, scene_texts[j]["text"])
            ref_len += scene_texts[j]["len"]
            j -= 1
        if ref_parts:
            ref_text = "".join(ref_parts)

        bundle_text = "".join(s["text"] for s in bundle)
        calls += 1

        # 체이닝 — 이전 결과 포함
        prev_list = ""
        if all_entities:
            prev_list = (
                "지금까지 찾은 요소:\n"
                + "\n".join(f"- {e['name']}" for e in all_entities)
                + "\n\n위 목록에 없는 새로운 요소만 추가하세요.\n\n"
            )

        user_prompt = prev_list
        if visual_rules:
            user_prompt += f"시각적 규칙:\n{visual_rules}\n\n"
        if ref_text:
            user_prompt += f"[앞쪽 씬 — 참조만]\n{ref_text}\n\n"
        user_prompt += (f"[분석 대상 씬]\n{bundle_text}\n\n"
                        f"{type_prompt}{binding_block}{roster_block}")

        try:
            result = call_structured(
                step=step_name,
                system_prompt=system,
                user_prompt=user_prompt,
                response_schema=schema,
                project_config=project_config,
                schema_name=f"{step_name}_{calls}",
                opik_metadata=opik_metadata,
            )
        except Exception as exc:
            logger.warning(
                "%s call %d failed (scenes %d~%d): %s",
                step_name, calls, bundle[0]["idx"], bundle[-1]["idx"], exc,
            )
            failed_calls += 1
            continue

        new_entities = result.get(key, [])
        added = 0
        for e in new_entities:
            name = e.get("name", "")
            if not name:
                continue
            if name in existing_names:
                # scene_count 누적 업데이트
                for existing in all_entities:
                    if existing["name"] == name:
                        old_sc = existing.get("scene_count", 0)
                        new_sc = e.get("scene_count", 0)
                        if new_sc > 0:
                            existing["scene_count"] = old_sc + new_sc
                        # ★★**후보 ID 를 합친다** (3b). 안 합치면 뒤 묶음이
                        #  적은 결속이 이 자리에서 통째로 사라진다.
                        if _allowed:
                            from app.modules.pipeline.grounding_binding import (
                                carry_into)

                            carry_into(existing, e)
                        break
            else:
                all_entities.append(e)
                existing_names.add(name)
                added += 1

        if added > 0:
            logger.info(
                "%s call %d (scenes %d~%d, %d chars): +%d (total %d)",
                step_name, calls, bundle[0]["idx"], bundle[-1]["idx"],
                bundle_len, added, len(all_entities),
            )

    if _allowed:
        # ★모든 묶음이 끝난 **뒤에** 가른다. 묶음마다 가르면 다음 묶음이
        #  같은 후보를 또 집어도 못 본다.
        from app.modules.pipeline.grounding_binding import normalize_binding

        _b = normalize_binding(all_entities, _allowed)
        logger.info("entity_all_%s 결속(체이닝): %s", entity_type, _b["counts"])
        _binding_out.update(_b)
    if _prior_allowed:
        # ★여기도 **묶음이 다 끝난 뒤**다. 묶음마다 대조하면 뒤 묶음이 같은
        #  앞 화 ID 를 또 주장해도 「다퉜다」를 못 본다.
        from app.modules.pipeline.episode_carry import apply_prior_ids

        _c = apply_prior_ids(all_entities, _prior_allowed)
        logger.info("entity_all_%s 앞 화 이관(체이닝): %s", entity_type, _c["counts"])
        if carry_out is not None:
            carry_out.update(_c)
    if failed_calls:
        logger.warning("entity_all_%s: %d entities (%d calls, %d failed)", entity_type, len(all_entities), calls, failed_calls)
    else:
        logger.info("entity_all_%s: %d entities (%d calls)", entity_type, len(all_entities), calls)
    return all_entities


def list_characters_from_shots(
    shots_data: List[Dict],
    visual_rules: str = "",
    project_config: Optional[Dict] = None,
    opik_metadata: Optional[Dict] = None,
    a0_candidates: Optional[List[Dict]] = None,
    binding_out: Optional[Dict] = None,
    prior_roster: Optional[tuple] = None,
    carry_out: Optional[Dict] = None,
    provenance_out: Optional[Dict] = None,
) -> List[Dict]:
    """shot descriptions를 모아 1회 호출로 인물 추출.

    ★얇은 wrapper 다 — 새 칸이 생기면 **여기도** 실어야 한다. 안 그러면
    인물 갈래만 장부가 없다.
    """
    return list_entities_from_shots(
        shots_data, "character", visual_rules, project_config, opik_metadata,
        a0_candidates=a0_candidates, binding_out=binding_out,
        prior_roster=prior_roster, carry_out=carry_out,
        provenance_out=provenance_out)


def list_entities_from_shots(
    shots_data: List[Dict],
    entity_type: str,
    visual_rules: str = "",
    project_config: Optional[Dict] = None,
    opik_metadata: Optional[Dict] = None,
    a0_candidates: Optional[List[Dict]] = None,
    binding_out: Optional[Dict] = None,
    prior_roster: Optional[tuple] = None,
    carry_out: Optional[Dict] = None,
    provenance_out: Optional[Dict] = None,
) -> List[Dict]:
    """shot descriptions를 모아 1회 호출로 엔티티 추출 (인물/배경/소품 공통).

    ★``binding_out`` — 결속 장부를 받아 갈 자리. 호출부가 체크포인트에
    남긴다. 안 주면 장부가 로그로만 남고 **다음 소비자가 못 읽는다**.

    전체 shot의 description을 이어붙여 LLM에게 1회 호출.

    ★``a0_candidates`` 는 GROUNDING-V2 §2-3.5 의 **후보 승격 통로**다.
    이 함수는 shot description 만 읽어서, 샷 구조에 안 들어온 대상은 **볼 기회조차
    없다**. A0 가 원문에서 건진 후보를 여기 넣어야 결속할 엔티티가 생긴다.

    ★**기본값 None 이면 legacy 와 한 글자도 안 달라진다** — 아래 overlay 블록이
    통째로 안 붙는다. 그것을 시험으로 잠갔다.

    ★``prior_roster`` — 앞 화들에서 확정된 신원 명부 `(블록, 허용 ID)`.
    같은 대상이 화마다 새 번호를 받는 것을 막는다. **첫 화는 비어 있어서
    한 바이트도 안 달라진다.** ``carry_out`` 은 그 이관 장부가 앉을 자리.
    """
    system = load_prompt(_MODULE, "system")
    type_prompt = load_prompt(_MODULE, entity_type)
    key = f"{entity_type}s" if entity_type != "prop" else "props"
    schema = load_schema(_MODULE, f"{entity_type}_schema")
    step_name = f"entity_all_{entity_type}"

    # 전체 shot descriptions 이어붙이기
    lines = []
    # ★★근거로 쓸 글은 **묘사문만** 모은다 (2026-09-19).
    #  아래 한 줄 끝의 `[{chars}]` 는 앞 단계(shot_validator)가 채운 **인물 이름
    #  칸**이다 — 모델이 앞 단계에서 지어낸 말이 그대로 들어 있다. 그것을
    #  근거 확인에 쓰면 모델이 **자기가 지어낸 말을 인용해** 문을 지난다.
    #  실측(09-19): 묘사문은 「젊은 여자」인데 인물 칸이 「젊은 아시아계 여자
    #  로봇」이었고, 모델이 근거로 댄 구절이 바로 `[현우, 젊은 아시아계 여자
    #  로봇]` 이었다 — 63개 전부 확인으로 통과했다.
    desc_lines = []
    for s in shots_data:
        for sh in s.get("shots", []):
            chars = ", ".join(sh.get("characters", []))
            lines.append(
                f"Scene {s['scene_index']} ({s.get('scene_heading', '')}) "
                f"Shot {sh['shot_index']}: {sh['description']} [{chars}]"
            )
            desc_lines.append(sh.get("description") or "")
    all_shots_text = "\n".join(lines)
    #: 이름 근거 확인에만 쓰는 글 — 모델이 채운 이름 칸은 **빠져 있다**.
    evidence_text = "\n".join(desc_lines)

    # shot_count 기반 프롬프트/스키마
    type_prompt_shot = type_prompt.replace("scene_count", "shot_count").replace("등장하는 씬 수", "등장하는 샷(스틸컷) 수")
    schema_shot = _patch_schema_shot_count(schema)

    # ★A0 후보 overlay — 후보가 없으면 **빈 목록**이라 legacy 는 바이트 불변이다.
    overlay = ""
    # ★★**결속 블록** (3b) — 이 호출은 후보 목록과 추출 대상을 **동시에 보는**
    #  유일한 자리다. 그래서 「이 행은 저 후보다」를 여기서 받아야 하고,
    #  그 뒤로는 코드가 기계적으로 나른다. 이름 대조가 아니다.
    binding_block, _allowed = "", []
    _binding_out = binding_out if binding_out is not None else {}
    if a0_candidates:
        from app.modules.pipeline.grounding_binding import (
            build_binding_block, patch_schema_with_candidate_ids)
        from app.modules.pipeline.grounding_overlay import build_overlay_lines

        _lines = build_overlay_lines(a0_candidates, entity_type)
        if _lines:
            overlay = "\n\n" + "\n".join(_lines)
        binding_block, _allowed = build_binding_block(a0_candidates, entity_type)
        # ★후보가 없으면 schema 를 **안 건드린다** — 빈 enum 은 모델을 세운다.
        schema_shot = patch_schema_with_candidate_ids(schema_shot, _allowed)

    # ★★**앞 화 명부** — 이 호출은 앞 화 신원과 이번 화 대상을 **동시에 보는**
    #  자리다. 「이 행은 앞 화의 저것이다」를 여기서 받고, 그 뒤로는 코드가
    #  대조해서 나른다. 이름 대조가 아니다.
    #  ★명부가 비면(첫 화) 블록도 스키마 패치도 **통째로 안 붙는다.**
    roster_block, _prior_allowed = (prior_roster or ("", []))
    if _prior_allowed:
        from app.modules.pipeline.episode_carry import patch_schema_with_prior_ids

        schema_shot = patch_schema_with_prior_ids(schema_shot, _prior_allowed)

    user_prompt = ""
    if visual_rules:
        user_prompt += f"시각적 규칙:\n{visual_rules}\n\n"
    # ★overlay 는 **제외 기준 뒤**에 온다. 앞에 두면 「빼지 마라 … 빼라」가 되어
    #  나중에 오는 제외 목록이 이긴다 (Codex 지적).
    user_prompt += (f"[전체 샷 목록]\n{all_shots_text}\n\n"
                    f"{type_prompt_shot}{overlay}{binding_block}{roster_block}")

    result = call_structured(
        step=step_name,
        system_prompt=system,
        user_prompt=user_prompt,
        response_schema=schema_shot,
        project_config=project_config,
        schema_name=step_name,
        opik_metadata=opik_metadata,
    )
    entities = result.get(key, [])
    # ★이름이 **본문에서 왔는지** 확인하고, 어긋나면 한 번 짚어서 다시 묻는다
    #  (2026-09-18). 금지문만으로는 지어낸 인종이 이름에 계속 붙었다.
    if entity_type == "character":
        from app.modules.pipeline.name_provenance import (
            check_name_provenance, correction_block, log_summary)

        # ★근거의 출처는 **모델이 실제로 본 글**이다 — 이 갈래는 대본 전문이
        #  아니라 **샷 목록**을 보여 준다. 전문으로 견주면 모델이 못 본 글을
        #  인용하라는 뜻이 되어 전부 어긋난다.
        # ★★다만 **묘사문만** 쓴다. 샷 줄 끝의 인물 이름 칸은 앞 단계 모델이
        #  채운 것이라, 그것까지 근거로 치면 지어낸 말이 자기 자신을 증명한다.
        _ok, _bad = check_name_provenance(entities, evidence_text)
        log_summary(step_name, _ok, _bad)
        if _bad:
            _retry = call_structured(
                step=step_name,
                system_prompt=system,
                user_prompt=user_prompt + correction_block(_bad),
                response_schema=schema_shot,
                project_config=project_config,
                schema_name=step_name,
                opik_metadata=opik_metadata,
            )
            _re_ents = _retry.get(key, [])
            # ★재질문도 **같은 출처**로 본다 — 여기만 `fulltext` 로 두어
            #  이 함수에 없는 이름을 참조했다(Codex BLOCK). 재질문이 한 번이라도
            #  돌면 **유료 호출을 마친 뒤** NameError 로 죽는다.
            _ok2, _bad2 = check_name_provenance(_re_ents, evidence_text)
            log_summary(step_name + "(재질문)", _ok2, _bad2)
            # ★다시 물어도 어긋나면 **그대로 둔다** — 여기서 인물을 버리면
            #  하류가 통째로 빈다. 기록만 남기고 사람이 본다.
            if len(_ok2) >= len(_ok):
                entities = _re_ents
        # ★★**남은 어긋남을 밖으로 낸다** (2026-09-19, Codex BLOCK 후속).
        #  「그대로 둔다」는 결정은 그대로지만, 로그로만 흘리면 체크포인트에
        #  아무 흔적이 없어 **사람이 볼 자리가 없다**. 여기서 낸 장부를
        #  호출부가 체크포인트에 남기고 갤러리가 읽는다.
        #  ★이것은 문이 아니다 — 막지 않고 **보이게** 할 뿐이다.
        if provenance_out is not None:
            _final_ok, _final_bad = check_name_provenance(entities, evidence_text)
            provenance_out.update({
                "checked": len(_final_ok) + len(_final_bad),
                "verified": len(_final_ok),
                "unverified": [
                    # ★`quote_preview` — 자른 것을 「인용문」이라 부르면
                    #  감사하는 쪽이 원문으로 읽는다(Codex 지적). 원문은
                    #  이 스텝의 provider 응답에 그대로 있다.
                    {"name": e.get("name"),
                     "quote_preview": (e.get("name_source_quote") or "")[:200],
                     "reason": e.get("_provenance_reason", "")}
                    for e in _final_bad
                ],
                "evidence_source": "shot_descriptions",
            })
    if _allowed:
        # ★모델이 낸 결속을 다듬고 **갈린 것을 가른다** — 한 후보를 두 행이
        #  가져갔으면 임의로 고르지 않는다.
        from app.modules.pipeline.grounding_binding import normalize_binding

        _b = normalize_binding(entities, _allowed)
        logger.info("entity_all_%s 결속: %s", entity_type, _b["counts"])
        # ★★**장부를 밖으로 낸다.** 로그만 찍으면 체크포인트에는 빈 배열만
        #  남아 「다퉜다」와 「아무것도 아니다」가 구별이 안 된다.
        _binding_out.update(_b)
    if _prior_allowed:
        # ★★모델이 고른 앞 화 신원을 **대조한다.** 명부 밖 값은 안 받고,
        #  둘이 같은 것을 주장하면 합치지 않는다(fail-closed).
        from app.modules.pipeline.episode_carry import apply_prior_ids

        _c = apply_prior_ids(entities, _prior_allowed)
        logger.info("entity_all_%s 앞 화 이관: %s", entity_type, _c["counts"])
        if carry_out is not None:
            carry_out.update(_c)
    logger.info("entity_all_%s (shot-based, 1 call): %d entities from %d shots", entity_type, len(entities), len(lines))
    return entities


def list_entities_with_shots(
    fulltext: str,
    entity_type: str,
    visual_rules: str = "",
    scenes: Optional[List[Dict]] = None,
    shots_by_scene: Optional[Dict[int, List[Dict]]] = None,
    project_config: Optional[Dict] = None,
    opik_metadata: Optional[Dict] = None,
) -> List[Dict]:
    """씬 원본(3000자) + 앞쪽 씬(2000자) + 해당 씬 shots로 체이닝 추출."""
    system = load_prompt(_MODULE, "system")
    type_prompt = load_prompt(_MODULE, entity_type)
    schema = load_schema(_MODULE, f"{entity_type}_schema")
    key = f"{entity_type}s" if entity_type != "prop" else "props"
    step_name = f"entity_all_{entity_type}"

    # shot_count 기반 프롬프트/스키마 패치
    type_prompt_shot = type_prompt.replace("scene_count", "shot_count").replace("등장하는 씬 수", "등장하는 샷(스틸컷) 수")
    if "shot_count" not in type_prompt_shot:
        type_prompt_shot += "\n- shot_count: 해당 요소가 등장하는 샷(스틸컷) 수를 함께 기재하세요."
    schema_shot = _patch_schema_shot_count(schema)

    if not scenes:
        return []

    scene_texts = []
    for seg in scenes:
        text = seg.get("text", "")
        scene_texts.append({
            "text": text,
            "len": seg.get("length", len(text)),
            "idx": seg.get("scene_index", 0),
        })

    # ★★결속 블록 (3b) — shot 경로와 **같은 함수**로 만든다.
    binding_block, _allowed = "", []
    _binding_out = binding_out if binding_out is not None else {}
    if a0_candidates:
        from app.modules.pipeline.grounding_binding import (
            build_binding_block, patch_schema_with_candidate_ids)

        binding_block, _allowed = build_binding_block(a0_candidates, entity_type)
        schema = patch_schema_with_candidate_ids(schema, _allowed)

    all_entities: List[Dict] = []
    existing_names: set = set()
    calls = 0
    i = 0

    while i < len(scene_texts):
        bundle = []
        bundle_len = 0
        while i < len(scene_texts) and bundle_len + scene_texts[i]["len"] <= BUNDLE_TARGET:
            bundle.append(scene_texts[i])
            bundle_len += scene_texts[i]["len"]
            i += 1
        if not bundle and i < len(scene_texts):
            bundle.append(scene_texts[i])
            bundle_len = scene_texts[i]["len"]
            i += 1

        # 앞쪽 참조
        ref_text = ""
        ref_len = 0
        j = bundle[0]["idx"] - 2
        ref_parts = []
        while j >= 0 and j < len(scene_texts) and ref_len + scene_texts[j]["len"] <= REF_MAX:
            ref_parts.insert(0, scene_texts[j]["text"])
            ref_len += scene_texts[j]["len"]
            j -= 1
        if ref_parts:
            ref_text = "".join(ref_parts)

        bundle_text = "".join(s["text"] for s in bundle)
        calls += 1

        # 해당 씬의 shots 포함
        shot_lines = []
        if shots_by_scene:
            for s in bundle:
                shots = shots_by_scene.get(s["idx"], [])
                for sh in shots:
                    chars = ", ".join(sh.get("characters", []))
                    shot_lines.append(f"  Scene {s['idx']} Shot {sh['shot_index']}: {sh['description']} [{chars}]")

        # 체이닝
        prev_list = ""
        if all_entities:
            prev_list = (
                "지금까지 찾은 요소:\n"
                + "\n".join(f"- {e['name']} (shot_count: {e.get('shot_count', 0)})" for e in all_entities)
                + "\n\n위 목록에 없는 새로운 요소만 추가하세요.\n\n"
            )

        user_prompt = prev_list
        if visual_rules:
            user_prompt += f"시각적 규칙:\n{visual_rules}\n\n"
        if ref_text:
            user_prompt += f"[앞쪽 씬 — 참조만]\n{ref_text}\n\n"
        user_prompt += f"[분석 대상 씬]\n{bundle_text}\n\n"
        if shot_lines:
            user_prompt += f"[해당 씬의 샷]\n" + "\n".join(shot_lines) + "\n\n"
        user_prompt += type_prompt_shot

        try:
            result = call_structured(
                step=step_name,
                system_prompt=system,
                user_prompt=user_prompt,
                response_schema=schema_shot,
                project_config=project_config,
                schema_name=f"{step_name}_{calls}",
                opik_metadata=opik_metadata,
            )
        except Exception as exc:
            logger.warning("%s call %d failed: %s", step_name, calls, exc)
            continue

        new_entities = result.get(key, [])
        added = 0
        for e in new_entities:
            name = e.get("name", "")
            if not name:
                continue
            if name in existing_names:
                for existing in all_entities:
                    if existing["name"] == name:
                        new_sc = e.get("shot_count", 0)
                        if new_sc > 0:
                            existing["shot_count"] = existing.get("shot_count", 0) + new_sc
                        break
            else:
                all_entities.append(e)
                existing_names.add(name)
                added += 1

        if added > 0:
            logger.info("%s call %d: +%d (total %d)", step_name, calls, added, len(all_entities))

    logger.info("%s (shot-chained): %d entities (%d calls)", step_name, len(all_entities), calls)
    return all_entities


def _patch_schema_shot_count(schema: Dict) -> Dict:
    """scene_count → shot_count로 스키마 패치. 없으면 shot_count 추가."""
    import copy
    s = copy.deepcopy(schema)
    for key in list(s.get("properties", {}).keys()):
        arr = s["properties"][key]
        if arr.get("type") == "array" and "items" in arr:
            props = arr["items"].get("properties", {})
            req = arr["items"].get("required", [])
            if "scene_count" in props:
                props["shot_count"] = props.pop("scene_count")
                if "scene_count" in req:
                    req[req.index("scene_count")] = "shot_count"
            elif "shot_count" not in props:
                props["shot_count"] = {"type": "integer"}
                if "shot_count" not in req:
                    req.append("shot_count")
    return s


def _list_single_call(
    fulltext: str,
    system: str,
    type_prompt: str,
    schema: Dict,
    key: str,
    step_name: str,
    visual_rules: str,
    project_config: Optional[Dict],
    opik_metadata: Optional[Dict],
    a0_candidates: Optional[List[Dict]] = None,
    entity_type: str = "",
    binding_out: Optional[Dict] = None,
    prior_roster: Optional[tuple] = None,
    carry_out: Optional[Dict] = None,
) -> List[Dict]:
    """기존 방식 — 전문 1회 호출 (fallback).

    ★여기에도 **같은 결속 계약**을 준다. 한 갈래만 빼 두면 그 갈래를 탄
    프로젝트에서 결속이 통째로 없고, 아무도 모른다.
    """
    binding_block, _allowed = "", []
    _binding_out = binding_out if binding_out is not None else {}
    if a0_candidates and entity_type:
        from app.modules.pipeline.grounding_binding import (
            build_binding_block, patch_schema_with_candidate_ids)

        binding_block, _allowed = build_binding_block(a0_candidates, entity_type)
        schema = patch_schema_with_candidate_ids(schema, _allowed)
    roster_block, _prior_allowed = (prior_roster or ("", []))
    if _prior_allowed:
        from app.modules.pipeline.episode_carry import patch_schema_with_prior_ids

        schema = patch_schema_with_prior_ids(schema, _prior_allowed)
    user_prompt = (
        f"시각적 규칙:\n{visual_rules}\n\n"
        f"시나리오 전문:\n{fulltext}\n\n"
        f"{type_prompt}{binding_block}{roster_block}"
    )
    result = call_structured(
        step=step_name,
        system_prompt=system,
        user_prompt=user_prompt,
        response_schema=schema,
        project_config=project_config,
        schema_name=step_name,
        opik_metadata=opik_metadata,
    )
    entities = result.get(key, [])
    if _allowed:
        from app.modules.pipeline.grounding_binding import normalize_binding

        _binding_out.update(normalize_binding(entities, _allowed))
    if _prior_allowed:
        from app.modules.pipeline.episode_carry import apply_prior_ids

        _c = apply_prior_ids(entities, _prior_allowed)
        if carry_out is not None:
            carry_out.update(_c)
    logger.info("entity_all_%s (single call): %d entities", step_name, len(entities))
    return entities