"""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 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