"""저빈도로 걸러낸 요소를 **지우지 말고 이 화에서만 뺀다** (2026-09-04).

## 무엇이 결함이었나 — 사용자가 지적한 바로 그 자리

`entity_filter` 가 걸러낸 요소는 `entity_detail`·`entity_t2i` 를 안 타고,
`EntitySyncService` 는 그 체크포인트만 읽는다. 그래서 걸러진 요소는 DB 에
**아무 흔적도 안 남았다.**

    실측 da049582 — 1화에서 공구상자·렌치·지팡이·담요 4개가 걸러졌고,
    체크포인트 `decisions` 에 이름과 사유 한 줄만 남았다.
    2화·3화는 **같은 소품**을 각각 따로 판단했고 이름도 서로 달랐다.
    5화에서 담요가 중요해져도 1화의 그것과 이을 근거가 없다.

## 여기서 재는 것

1. 걸러진 행이 **통째로** 산출에 남는가 (이름 한 줄이 아니라)
2. 제거 판단을 **`short_id`** 로 하는가 (이름 글자 대조가 아니라)
3. 보내지 않은 대상을 지우라고 해도 **안 지우는가**
4. 보류 행이 canon + `shelved` 링크로 남는가 · `t2i_prompt` 는 비어 있는가
5. 다음 화에 다시 나오면 **같은 번호로 되살아나는가**
6. 하류 게이트가 보류 행을 **대상에서 빼는가** (안 빼면 유료 호출이 는다)

Lane: ``-m pg``.
"""
from __future__ import annotations

import json
import uuid
from pathlib import Path

import pytest
from sqlalchemy import text as sql_text

pytestmark = pytest.mark.pg


def _seed(session, pid: str, *eids: str) -> None:
    uid = f"shelf-{uuid.uuid4()}"
    session.execute(sql_text(
        "INSERT INTO user_account (id, username, display_name, password_hash, "
        "role, is_active, created_at, updated_at) VALUES "
        "(:uid, :un, 't', 'x', 'creator', 1, '2026-01-01', '2026-01-01')"
    ), {"uid": uid, "un": f"u_{uid}"})
    session.execute(sql_text(
        "INSERT INTO project_registry (id, name, created_by, created_at, updated_at) "
        "VALUES (:pid, 'shelf', :uid, '2026-01-01', '2026-01-01')"
    ), {"pid": pid, "uid": uid})
    for n, eid in enumerate(eids, start=1):
        session.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, created_at, updated_at) VALUES "
            "(:eid, :pid, :t, :n, 'x.txt', 'x/x.txt', '2026-01-01', '2026-01-01')"
        ), {"eid": eid, "pid": pid, "t": f"{n}화", "n": n})
    session.commit()


def _write_filter_cp(root: Path, pid: str, eid: str, kept: list, removed: list) -> None:
    d = root / pid / "checkpoints" / "episodes" / eid / "entity_filter"
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(json.dumps({
        "status": "completed",
        "data": {
            "decisions": [],
            "filtered_entities": {"characters": [], "locations": [],
                                  "props": kept},
            "removed_entities": removed,
            "removed_count": len(removed),
            "kept_count": len(kept),
        },
    }, ensure_ascii=False), encoding="utf-8")


def _shelf(session, pid: str, eid: str):
    from app.services.checkpoint_sync import ShelfSyncService

    out = ShelfSyncService(session, pid, eid).sync_from_checkpoint()
    session.commit()
    return out


# ── 필터 자체 ────────────────────────────────────────────────────────


def _run_filter(monkeypatch, entities, decisions):
    from app.modules.pipeline import entity_filter as ef

    monkeypatch.setattr(ef, "load_prompt", lambda *a, **k: "sys")
    monkeypatch.setattr(ef, "load_schema", lambda *a, **k: {})
    monkeypatch.setattr(ef, "call_structured",
                        lambda **kw: {"decisions": decisions})
    return ef.filter_low_frequency_entities(
        entities=entities, segments=[], fulltext="", max_scenes=3)


def test_걸러진_행이_통째로_남는다(monkeypatch):
    """★종전에는 이름과 사유 한 줄만 남았다."""
    ents = {"characters": [], "locations": [],
            "props": [{"short_id": "P06", "name": "담요", "shot_count": 1,
                       "description": "낡은 회색 담요",
                       "visual_traits": ["회색", "낡음"]}]}
    out = _run_filter(monkeypatch, ents,
                      [{"short_id": "P06", "name": "담요", "entity_type": "prop",
                        "decision": "remove", "reason": "1회뿐"}])

    assert out["removed_count"] == 1
    row = out["removed_entities"][0]
    assert row["name"] == "담요"
    assert row["description"] == "낡은 회색 담요", "설명이 사라졌다"
    assert row["visual_traits"] == ["회색", "낡음"], "특징이 사라졌다"
    assert row["entity_type"] == "prop", "갈래를 안 찍었다"
    assert row["shelved_reason"] == "1회뿐"


def test_제거는_short_id_로_정한다(monkeypatch):
    """이름이 달라도 `short_id` 가 맞으면 지운다 — 글자 대조가 아니다."""
    ents = {"characters": [], "locations": [],
            "props": [{"short_id": "P06", "name": "담요", "shot_count": 1}]}
    out = _run_filter(monkeypatch, ents,
                      [{"short_id": "P06", "name": "P06 낡은 담요",
                        "entity_type": "prop", "decision": "remove",
                        "reason": "r"}])
    assert out["removed_count"] == 1, "이름이 달라 못 찾았다"


def test_보내지_않은_대상은_안_지운다(monkeypatch):
    """★지우는 쪽으로 틀리면 되돌릴 수 없다."""
    ents = {"characters": [], "locations": [],
            "props": [{"short_id": "P06", "name": "담요", "shot_count": 1}]}
    out = _run_filter(monkeypatch, ents,
                      [{"short_id": "P99", "name": "없는것",
                        "entity_type": "prop", "decision": "remove",
                        "reason": "r"}])
    assert out["removed_count"] == 0, "안 보낸 대상을 지웠다"
    assert len(out["filtered_entities"]["props"]) == 1


def test_short_id_없는_판단은_살린다(monkeypatch):
    """옛 팩 산출. 이름으로 짐작하지 않는다."""
    ents = {"characters": [], "locations": [],
            "props": [{"short_id": "P06", "name": "담요", "shot_count": 1}]}
    out = _run_filter(monkeypatch, ents,
                      [{"name": "담요", "entity_type": "prop",
                        "decision": "remove", "reason": "r"}])
    assert out["removed_count"] == 0, "short_id 없는 판단으로 지웠다"


# ── DB 에 남는가 ─────────────────────────────────────────────────────


@pytest.fixture
def scene(pg_session, tmp_path: Path, monkeypatch):
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    pid = f"p-{uuid.uuid4()}"
    ep1, ep2 = str(uuid.uuid4()), str(uuid.uuid4())
    _seed(pg_session, pid, ep1, ep2)
    return pid, ep1, ep2, tmp_path


_BLANKET = {"short_id": "P06", "name": "담요", "entity_type": "prop",
            "description": "낡은 회색 담요", "visual_traits": ["회색"],
            "shot_count": 1, "shelved_reason": "1회뿐"}


def test_보류_행이_canon_과_shelved_링크로_남는다(pg_session, scene):
    from app.core.entity_identity import PRESENCE_SHELVED

    pid, ep1, _ep2, root = scene
    _write_filter_cp(root, pid, ep1, kept=[], removed=[dict(_BLANKET)])
    out = _shelf(pg_session, pid, ep1)
    assert out["shelved"] == 1

    row = pg_session.execute(sql_text(
        "SELECT c.name, c.t2i_prompt, l.presence_status, l.episode_notes_json "
        "FROM entity_canon c JOIN entity_episode_link l ON l.canon_id = c.id "
        "WHERE c.project_id = :p AND c.short_id = 'P06' AND l.episode_id = :e"
    ), {"p": pid, "e": ep1}).fetchone()
    assert row is not None, "보류 행이 DB 에 안 남았다"
    assert row[0] == "담요"
    assert row[1] == "", f"t2i_prompt 가 채워졌다 — 하류가 만들 것으로 읽는다: {row[1]}"
    assert row[2] == PRESENCE_SHELVED
    assert json.loads(row[3])["shelved_reason"] == "1회뿐"


def test_다음_화에_다시_나오면_같은_번호로_되살아난다(pg_session, scene):
    """★핵심 — 이것이 「뒤 화에서 앞과 이어진다」의 실체다."""
    from app.core.entity_identity import PRESENCE_ACTIVE

    pid, ep1, ep2, root = scene
    _write_filter_cp(root, pid, ep1, kept=[], removed=[dict(_BLANKET)])
    _shelf(pg_session, pid, ep1)

    # 2화에서는 여러 번 나와서 살아남았다 — **같은 short_id** 로.
    _write_filter_cp(root, pid, ep2,
                     kept=[{"short_id": "P06", "name": "담요", "shot_count": 5}],
                     removed=[])
    # 2화 링크는 EntitySyncService 가 만든다 — 여기서는 그 상태를 흉내 낸다.
    cid = pg_session.execute(sql_text(
        "SELECT id FROM entity_canon WHERE project_id = :p AND short_id = 'P06'"
    ), {"p": pid}).fetchone()[0]
    pg_session.execute(sql_text(
        "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id, "
        "presence_status) VALUES (:l, :c, :p, :e, 'shelved')"
    ), {"l": str(uuid.uuid4()), "c": cid, "p": pid, "e": ep2})
    pg_session.commit()

    out = _shelf(pg_session, pid, ep2)
    assert out["revived"] == 1, "되살아나지 않았다"

    st = pg_session.execute(sql_text(
        "SELECT presence_status FROM entity_episode_link "
        "WHERE canon_id = :c AND episode_id = :e"), {"c": cid, "e": ep2}).fetchone()[0]
    assert st == PRESENCE_ACTIVE

    # 1화의 보류 기록은 그대로 남아 있어야 한다 — 그것이 계보다.
    st1 = pg_session.execute(sql_text(
        "SELECT presence_status FROM entity_episode_link "
        "WHERE canon_id = :c AND episode_id = :e"), {"c": cid, "e": ep1}).fetchone()[0]
    assert st1 == "shelved", "1화의 보류 기록이 지워졌다"


def test_보류_행은_다음_화_명부에_실린다(pg_session, scene):
    """1화에서 보류된 담요가 5화의 모델에게 보여야 이어 붙일 수 있다."""
    from app.modules.pipeline.episode_carry import build_prior_roster

    pid, ep1, _ep2, root = scene
    _write_filter_cp(root, pid, ep1, kept=[], removed=[dict(_BLANKET)])
    _shelf(pg_session, pid, ep1)

    _lines, allowed = build_prior_roster(pg_session, pid, "prop")
    assert "P06" in allowed, f"보류 행이 명부에 없다: {allowed}"


def test_하류_게이트가_보류_행을_대상에서_뺀다(pg_session, scene):
    """★안 빼면 참조 대상으로 세어져 게이트가 안 풀리고 유료 호출이 는다."""
    from app.core.pipeline_gate import episode_reference_entities

    pid, ep1, _ep2, root = scene
    _write_filter_cp(root, pid, ep1, kept=[], removed=[dict(_BLANKET)])
    _shelf(pg_session, pid, ep1)

    got = episode_reference_entities(pg_session, pid, ep1)
    assert [e.short_id for e in got] == [], (
        f"보류 행이 참조 대상으로 세어졌다: {[e.short_id for e in got]}")


def test_옛_체크포인트는_보류_0_이_아니라_모른다로_지나간다(pg_session, scene):
    """`removed_entities` 칸이 없는 옛 산출을 「없었다」로 읽으면 안 된다."""
    pid, ep1, _ep2, root = scene
    d = root / pid / "checkpoints" / "episodes" / ep1 / "entity_filter"
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(json.dumps({
        "status": "completed",
        "data": {"decisions": [], "filtered_entities": {}, "removed_count": 2},
    }), encoding="utf-8")

    out = _shelf(pg_session, pid, ep1)
    assert out["skipped"] == 1, "옛 산출을 보류 0 으로 읽었다"