"""G4.6 Wave A1 RC-C — entity reference 보호 룰.

low_freq_skip 의 deterministic protection cascade (모두 generic, 시나리오
의존 0건):
- variant pole / variant_self (RO-1) — character-character + identity/transformation
  관계의 second-participant 만 (possession prop / location-location 은 제외)
- is_base_for_variant — RelationFact 기반 base 측 보호
- pipeline manifest cascade — scene_director / shot_validator / shot_director
  + scene_detail rescue (force/retry 시 보강) — production data 만 읽음
- count > 1 — frequency 자체

이전 iter 의 keyword-based protection 은 제거됨 — 보호 판정은 graph (RelationFact)
+ pipeline manifest signal 만으로 derive.
"""
from __future__ import annotations

import json
import logging
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)


def _parse_traits(raw: Any) -> list[str]:
    """`EntityCanon.stable_traits` 는 JSON 문자열 (또는 이미 list). 양쪽 지원."""
    if not raw:
        return []
    if isinstance(raw, list):
        return [str(t) for t in raw]
    if isinstance(raw, str):
        try:
            parsed = json.loads(raw)
        except (json.JSONDecodeError, TypeError):
            return [raw]
        if isinstance(parsed, list):
            return [str(t) for t in parsed]
        if isinstance(parsed, dict):
            return [str(v) for v in parsed.values()]
        return [str(parsed)]
    return []


def _load_cp(project_id: str, episode_id: str, step_id: str) -> dict | None:
    """checkpoint manifest loader. 부재 시 None (caller 가 union 시 무시).

    PRO-5: settings.projects_dir 절대경로 사용 (process cwd 비의존).
    """
    from app.core.config import settings

    cp_path = (
        Path(settings.projects_dir)
        / project_id / "checkpoints" / "episodes" / episode_id / step_id
        / "manifest.json"
    )
    if not cp_path.exists():
        return None
    try:
        return json.loads(cp_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError) as exc:
        logger.warning(
            "entity_protection._load_cp(%s): manifest 파싱 실패 — %s — None 반환",
            step_id, exc,
        )
        return None


def _short_id_base(short_id: str) -> str:
    """C##O## composite → C## base. C## / L## / P## / O## bare 는 그대로."""
    return short_id.split("O")[0] if short_id and "O" in short_id else (short_id or "")


def _collect_required_entity_ids(project_id: str, episode_id: str) -> set[str]:
    """pipeline manifest cascade — short_id 집합 union.

    sources (모두 manifest-driven, union):
    - scene_director.data.scenes[].present_entity_ids
    - shot_validator.data.scenes[].shots[].character_ids (RO-1)
    - shot_director.data.scenes[].shots[].visible_entity_ids (production schema)
    - scene_detail rescue — visible_entities + render_prompt_card.asset_requirements.required_refs[].id
      (force/retry 시 보강)

    DB EntityEpisodeLink.t2i_appearance_count 기반 source 는 의도적으로 제외
    (Codex iter2 BLOCKING — count >= 1 보호는 low_freq_skip 정책 자체를 무력화).
    """
    short_ids: set[str] = set()

    director_cp = _load_cp(project_id, episode_id, "scene_director")
    if director_cp:
        for sc in director_cp.get("data", {}).get("scenes", []) or []:
            for sid in sc.get("present_entity_ids", []) or []:
                short_ids.add(_short_id_base(sid))

    sv_cp = _load_cp(project_id, episode_id, "shot_validator")
    if sv_cp:
        for sc in sv_cp.get("data", {}).get("scenes", []) or []:
            for sh in sc.get("shots", []) or []:
                for sid in sh.get("character_ids", []) or []:
                    short_ids.add(_short_id_base(sid))

    sdir_cp = _load_cp(project_id, episode_id, "shot_director")
    if sdir_cp:
        for sc in sdir_cp.get("data", {}).get("scenes", []) or []:
            for sh in sc.get("shots", []) or []:
                for sid in sh.get("visible_entity_ids", []) or []:
                    short_ids.add(_short_id_base(sid))

    sd_cp = _load_cp(project_id, episode_id, "scene_detail")
    if sd_cp:
        for sh in sd_cp.get("data", {}).get("scenes", []) or []:
            for sid in sh.get("visible_entities", []) or []:
                short_ids.add(_short_id_base(sid))
            rpc = sh.get("render_prompt_card") or {}
            asset_req = rpc.get("asset_requirements") or {}
            for ref in asset_req.get("required_refs", []) or []:
                if not isinstance(ref, dict):
                    continue
                rid = ref.get("id")
                if rid:
                    short_ids.add(_short_id_base(rid))

    return short_ids


def _collect_reference_required_ids(project_id: str, episode_id: str) -> set[str]:
    """Phase 1 — reference 생성 보호의 단일 SOT.

    `scene_detail.render_prompt_card.asset_requirements.required_refs[].id` 만
    수집한다 (`_collect_required_entity_ids` 의 4-source broad union 과 달리
    scene_director.present / shot_validator.character_ids / shot_director.
    visible_entity_ids / scene_detail.visible_entities 전부 **제외**).

    설계 근거: docs/reference-necessity/index.html §6.1 — broad union 은
    catalog/audit 용도로만 남기고, reference generation 보호는 실제 attach
    계약(required_refs)에 가장 가까운 단일 source 로 좁힌다.
    """
    short_ids: set[str] = set()
    sd_cp = _load_cp(project_id, episode_id, "scene_detail")
    if sd_cp:
        for sh in sd_cp.get("data", {}).get("scenes", []) or []:
            rpc = sh.get("render_prompt_card") or {}
            asset_req = rpc.get("asset_requirements") or {}
            for ref in asset_req.get("required_refs", []) or []:
                if not isinstance(ref, dict):
                    continue
                rid = ref.get("id")
                if rid:
                    short_ids.add(_short_id_base(rid))
    return short_ids


def collect_reference_required_canon_ids(
    db: Any, project_id: str, episode_id: str,
) -> dict[str, str]:
    """참조 생성 보호 대상 — canon id → 보호 사유.

    세 출처다.

    - ``scene_detail.required_refs`` (`_collect_reference_required_ids`) —
      종전의 단일 SOT.
    - **선택된 샷의 VE 에 있는 캐릭터** (2026-09-19 사용자 지시). 샷 이미지
      (still_recipe)는 VE 를 보고 캐릭터 참조를 붙인다 — 그것이 실제 첨부
      기준이다. 둘을 따로 보면 「샷에는 나오는데 참조는 안 만든」 캐릭터가
      생긴다(컨트리로드 2판 6명·6샷 — 여섯 다 `t2i_count=0 · required=False`
      로 건너뛰었다. 찰리 홀로그램은 scene_detail 이 visible 에는 넣고
      required_refs 에는 안 넣었다).
    - **선택된 샷의 촬영 계획이 화면 안에 세운 캐릭터** (`_staged_in_frame_sids`
      — 스틸 조립이 VE 밖이어도 참조를 붙이는 인물). 이 출처는 배경 전용 샷을
      안 가르므로 붙지 않을 참조를 만들 수 있다 — **샷 그림은 안 망가지지만
      참조 생성 요금은 든다**(해가 없다 ≠ 무료).

    ★캐릭터만 넓힌다 — 보호를 좁힌 설계 취지(쓸데없는 참조 생성을 막는다)는
     그대로이고, 선택 샷 VE 는 그 샷이 **실제로 붙이는** 것이라 넓게 잡는
     출처가 아니다.
    ★short_id 는 프로젝트마다 따로다 — project 로 거른다.
    """
    from app.models.project import EntityCanon
    from app.modules.pipeline.shot_ref_classify import (
        load_character_sids_by_tag,
    )

    sd_sids = _collect_reference_required_ids(project_id, episode_id)
    by_tag = load_character_sids_by_tag(
        db, project_id, episode_id, selected_only=True)
    ve_sids: set[str] = set().union(*by_tag.values()) if by_tag else set()
    staged_sids = _staged_in_frame_sids(db, project_id, episode_id,
                                        set(by_tag))
    want = sd_sids | ve_sids | staged_sids
    if not want:
        return {}
    out: dict[str, str] = {}
    rows = (
        db.query(EntityCanon)
        .filter(
            EntityCanon.project_id == project_id,
            EntityCanon.short_id.in_(want),
        )
        .all()
    )
    for e in rows:
        sid = e.short_id or ""
        if sid in sd_sids:
            out[e.id] = "scene_detail.required_refs"
        elif sid in ve_sids:
            out[e.id] = "selected shot VE"
        elif sid in staged_sids:
            out[e.id] = "selected shot staging"
    return out


def _staged_in_frame_sids(
    db: Any, project_id: str, episode_id: str, selected_tags: set,
) -> set[str]:
    """선택 샷의 촬영 계획(shot_staging)이 **화면 안에 세운** 등록 인물 short_id
    (2026-09-19, Codex 지적 — 참조를 붙이는 명단과 만드는 명단이 같아야 한다).

    스틸 조립은 이 인물에게 참조를 붙인다(`still_recipe_service._staged_in_frame`).
    실측: 셰퍼드·재판관이 촬영 계획엔 있고 VE 엔 없어 **참조가 아예 없었다** —
    붙일 것이 없으면 이름만 실린다.

    ★이름 조인은 스틸 조립과 **같은 함수**다(`staged_in_frame_character_ids` —
     POV 제외 · 동명 제외). 배경 전용 여부는 가르지 않는다 — 넓게 잡아도 샷
     그림은 안 망가지지만(그 샷에 안 붙을 뿐) **참조 생성 요금은 든다**.
     실측(컨트리로드 2판)에서 이 출처로 새로 생기는 참조는 2장(셰퍼드·재판관).
    """
    from app.core.config import settings
    from app.models.project import EntityCanon
    from app.modules.pipeline.shot_ref_classify import tag_of
    from app.modules.pipeline.shot_visibility import (
        staged_in_frame_character_ids,
        unique_character_ids_by_name,
    )

    path = (Path(settings.projects_dir) / project_id / "checkpoints"
            / "episodes" / episode_id / "shot_staging" / "manifest.json")
    if not path.exists():
        return set()
    try:
        shots = (json.loads(path.read_text(encoding="utf-8")).get("data")
                 or {}).get("shots") or []
    except Exception as exc:  # noqa: BLE001 — 보호를 넓히는 출처일 뿐
        logger.warning("entity_protection: shot_staging 을 못 읽었다: %s", exc)
        return set()
    # ★이름을 잇는 범위도 스틸 조립과 같다 — **이 화에서 살아 있는** 인물만
    #  (`load_episode_entity_dicts` 가 쓰는 `active_episode_canon_ids`). 프로젝트
    #  전체로 이으면 이 화에 연결이 없는 인물(실측: 재판관 C50)까지 세고,
    #  동명 판정도 스틸 조립과 갈린다.
    from app.core.entity_identity import active_episode_canon_ids

    active = set(active_episode_canon_ids(db, project_id, episode_id))
    chars = [
        e for e in (
            db.query(EntityCanon)
            .filter(EntityCanon.project_id == project_id,
                    EntityCanon.entity_type == "character")
            .all()
        )
        if e.id in active
    ]
    id_by_name = unique_character_ids_by_name(
        (e.id, e.name or "") for e in chars)
    sid_by_id = {e.id: e.short_id for e in chars if e.short_id}
    out: set[str] = set()
    for s in shots:
        if not isinstance(s, dict):
            continue
        si, shi = s.get("scene_index"), s.get("shot_index")
        if si is None or shi is None or tag_of(int(si), int(shi)) not in selected_tags:
            continue
        for eid in staged_in_frame_character_ids(
                s.get("character_angles") or [], id_by_name,
                s.get("pov_character")):
            if eid in sid_by_id:
                out.add(str(sid_by_id[eid]))
    return out


def compute_variant_pole_ids(
    entities: list[dict], relations: list[dict], participants: list[dict],
) -> set[str]:
    """RO-1 narrowed — character-character + relation_family in
    {identity, transformation} 의 second participant 집합 (variant 측 canon_id).

    `build_visual_dependency_graph` 의 generic deps 는 possession (character→prop)
    + location-location 도 dependency 로 만들어 variant_self 가 너무 넓어짐
    (Codex iter2 IMPORTANT). 이 helper 는 RO-1 의 의도인 'variant pole 자체
    보호' 만 정확히 반영.

    relations 는 [{id, relation_family}], participants 는 [{relation_id, canon_id}].
    """
    entity_types = {e["id"]: e.get("entity_type") for e in entities}
    rel_parts: dict[str, list] = {}
    for p in participants:
        rel_parts.setdefault(p["relation_id"], []).append(p)

    variant_poles: set[str] = set()
    for rel in relations:
        if rel.get("relation_family") not in ("identity", "transformation"):
            continue
        parts = rel_parts.get(rel["id"], [])
        if len(parts) < 2:
            continue
        for i, p1 in enumerate(parts):
            for p2 in parts[i + 1:]:
                # `entity_dependency.py:76` 와 동일: first participant 가 base,
                # second 가 variant. character-character 만 RO-1 적용 대상.
                if entity_types.get(p1["canon_id"]) != "character":
                    continue
                if entity_types.get(p2["canon_id"]) != "character":
                    continue
                variant_poles.add(p2["canon_id"])
    return variant_poles


def should_skip_low_freq(
    e: dict,
    count: int,
    is_base_for_variant: bool,
    is_variant_self: bool,
    required_by_pipeline: bool,
) -> bool:
    """deterministic protection cascade — skip 조건 모두 만족 시 True.

    보호 발동 (return False) 우선순위:
    - location/outlook entity_type — helper 진입 시 False (caller 가 이미 skip)
    - count > 1 — frequency 자체로 충분
    - is_base_for_variant — variant pole 의 base 측
    - is_variant_self — character-character identity/transformation variant 자체 (RO-1)
    - required_by_pipeline — manifest 4-source cascade union 안

    위 조건 모두 안 걸리면 skip (return True).
    """
    etype = e.get("entity_type", "")
    if etype in ("location", "outlook"):
        return False
    if count > 1:
        return False
    if is_base_for_variant:
        return False
    if is_variant_self:
        return False
    if required_by_pipeline:
        return False
    return True