"""재사용하는 canon 의 **시각 계약 칸**을 다음 화가 덮지 않는다 (2026-09-04).

## 왜 신원을 고쳐도 이것이 남나 — Codex 가 잡은 자리

번호를 프로젝트 장부에서 발급하고 앞 화 명부로 이어 붙여도,
`entity_sync_service.py:251-264` 가 재사용 canon 의
`description`·`t2i_prompt`·`stable_traits`·`metadata_json` 을 **새 화 것으로
무조건 덮었다.**

참조 이미지는 **거기 적힌 `t2i_prompt` 로** 만들어졌다. 덮는 순간 그림이
무엇을 그린 것인지 알 수 없게 된다 — 실측 `da049582` 의 O01 이 정확히 그
모양이다(1화가 만든 남색작업복 그림 + 3화의 감색차장제복 이름).

## 규칙

- 기존 canon 의 **비어 있지 않은** 안정 칸은 **자산 유무와 무관하게** 보존한다.
  (앞 판은 「참조 이미지가 있을 때만」 얼렸다. 그러면 이미지 만들기 전에 2화가
  sync 되면 1화 canon 이 2화 외형으로 바뀌고, 나중에 1화 참조를 만들 때 이미
  틀린 프롬프트를 쓴다 — Codex 재지적 2026-09-04.)
- 새 이름은 지우지 않고 **별명**(`EntityAlias`)으로 쌓는다 — 화마다 부르는
  말이 다를 수 있고, 그 다름이 다음 화의 대조 재료다.
- 이번 화의 묘사는 **버리지도 덮지도 않고** 링크(`episode_notes_json`)에 둔다.
- **빈 칸만** 채운다. 처음 채우는 것은 막지 않는다.

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, ep1: str, ep2: str) -> None:
    uid = f"stable-{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, 'stable', :uid, '2026-01-01', '2026-01-01')"
    ), {"pid": pid, "uid": uid})
    for n, eid in enumerate((ep1, ep2), 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 _canon(session, pid: str, short_id: str, name: str, etype: str,
           t2i: str, traits: list) -> str:
    cid = str(uuid.uuid4())
    session.execute(sql_text(
        "INSERT INTO entity_canon (id, project_id, short_id, name, entity_type, "
        "description, stable_traits, metadata_json, t2i_prompt, status, "
        "created_at, updated_at) VALUES "
        "(:cid, :pid, :s, :n, :e, '1화 설명', :tr, '{}', :t2i, 'active', "
        "'2026-01-01', '2026-01-01')"
    ), {"cid": cid, "pid": pid, "s": short_id, "n": name, "e": etype,
        "t2i": t2i, "tr": json.dumps(traits, ensure_ascii=False)})
    return cid


def _reference_image(session, pid: str, canon_id: str) -> None:
    session.execute(sql_text(
        "INSERT INTO image_asset (id, project_id, asset_type, entity_id, "
        "file_path, is_primary, created_at) VALUES "
        "(:i, :p, 'reference', :e, 'assets/x.png', 1, '2026-01-01')"
    ), {"i": str(uuid.uuid4()), "p": pid, "e": canon_id})


def _write_cp(root: Path, pid: str, eid: str, payload: dict) -> None:
    d = root / pid / "checkpoints" / "episodes" / eid / "entity_t2i"
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(json.dumps(payload, ensure_ascii=False),
                                     encoding="utf-8")


def _cp(name: str, desc: str, t2i: str, traits: list) -> dict:
    return {
        "status": "completed",
        "data": {"characters": [{
            "short_id": "C01", "name": name, "description": desc,
            "t2i_prompt": t2i, "visual_traits": traits,
            # ★character 의 정본 모양 — 키 둘 다 있고 값은 None
            #  (`validate_entity_metadata_shape`). 손으로 `{}` 를 넣으면
            #  프로덕션이 안 받는 모양을 시험이 만드는 것이다.
            "metadata_json": {"location": None, "visual_identity": None},
        }], "locations": [], "props": []},
    }


@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)
    cid = _canon(pg_session, pid, "C01", "최씨", "character",
                 "a man in a navy work jacket", ["navy jacket", "short hair"])
    pg_session.execute(sql_text(
        "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
        "VALUES (:l, :c, :p, :e)"
    ), {"l": str(uuid.uuid4()), "c": cid, "p": pid, "e": ep1})
    pg_session.commit()
    return pid, ep1, ep2, cid, tmp_path


def _sync(session, pid: str, eid: str):
    from app.services.checkpoint_sync import EntitySyncService

    EntitySyncService(session, pid, eid).sync_from_checkpoint()
    session.commit()


def test_참조_이미지가_있으면_t2i_프롬프트를_안_덮는다(pg_session, scene):
    """★핵심 — 그 그림이 이 문장으로 만들어졌다."""
    pid, _ep1, ep2, cid, root = scene
    _reference_image(pg_session, pid, cid)
    pg_session.commit()

    _write_cp(root, pid, ep2, _cp("기사 최씨", "2화 설명",
                                  "a man in a conductor uniform", ["uniform"]))
    _sync(pg_session, pid, ep2)

    row = pg_session.execute(sql_text(
        "SELECT t2i_prompt, stable_traits, name FROM entity_canon WHERE id = :c"
    ), {"c": cid}).fetchone()
    assert row[0] == "a man in a navy work jacket", (
        f"그림을 만든 t2i_prompt 가 2화 것으로 덮였다: {row[0]}")
    assert "navy jacket" in row[1], f"시각 특징이 덮였다: {row[1]}"
    assert row[2] == "최씨", f"이름이 덮였다: {row[2]}"


def test_새_이름은_별명으로_쌓인다(pg_session, scene):
    pid, _ep1, ep2, cid, root = scene
    _reference_image(pg_session, pid, cid)
    pg_session.commit()

    _write_cp(root, pid, ep2, _cp("기사 최씨", "2화 설명", "x", ["y"]))
    _sync(pg_session, pid, ep2)

    aliases = {r[0] for r in pg_session.execute(sql_text(
        "SELECT alias FROM entity_alias WHERE canon_id = :c"), {"c": cid}).fetchall()}
    assert "기사 최씨" in aliases, f"새 이름이 안 남았다: {aliases}"


def test_이번_화_묘사는_링크에_남는다(pg_session, scene):
    """버리지도 덮지도 않는다."""
    pid, _ep1, ep2, cid, root = scene
    _reference_image(pg_session, pid, cid)
    pg_session.commit()

    _write_cp(root, pid, ep2, _cp("기사 최씨", "2화 설명",
                                  "conductor uniform", ["uniform"]))
    _sync(pg_session, pid, ep2)

    row = pg_session.execute(sql_text(
        "SELECT episode_notes_json FROM entity_episode_link "
        "WHERE canon_id = :c AND episode_id = :e"), {"c": cid, "e": ep2}).fetchone()
    assert row is not None and row[0], "이번 화 묘사가 안 남았다"
    notes = json.loads(row[0])
    assert notes["name"] == "기사 최씨"
    assert notes["description"] == "2화 설명"
    assert notes["t2i_prompt"] == "conductor uniform"


def test_참조_이미지가_없어도_안정_칸을_안_덮는다(pg_session, scene):
    """★Codex 재지적 — 이미지 존재는 **신원 안정성의 조건이 아니다**.

    이미지 만들기 전에 2화가 sync 되면 1화 canon 이 2화 외형으로 바뀌고,
    나중에 1화 참조를 만들 때 이미 틀린 프롬프트를 쓴다.
    """
    pid, _ep1, ep2, cid, root = scene
    # 참조 이미지 없음 — 그래도 덮으면 안 된다.
    _write_cp(root, pid, ep2, _cp("기사 최씨", "2화 설명",
                                  "conductor uniform", ["uniform"]))
    _sync(pg_session, pid, ep2)

    row = pg_session.execute(sql_text(
        "SELECT t2i_prompt, name FROM entity_canon WHERE id = :c"
    ), {"c": cid}).fetchone()
    assert row[0] == "a man in a navy work jacket", (
        f"참조가 없다는 이유로 1화의 t2i_prompt 를 덮었다: {row[0]}")
    assert row[1] == "최씨", f"이름이 덮였다: {row[1]}"


def test_비어_있는_칸은_이번_화가_채운다(pg_session, scene):
    """과보호도 결함이다 — 처음 채우는 것은 막지 않는다."""
    pid, _ep1, ep2, cid, root = scene
    pg_session.execute(sql_text(
        "UPDATE entity_canon SET t2i_prompt = '', description = '' "
        "WHERE id = :c"), {"c": cid})
    pg_session.commit()

    _write_cp(root, pid, ep2, _cp("최씨", "2화 설명", "conductor uniform", ["u"]))
    _sync(pg_session, pid, ep2)

    row = pg_session.execute(sql_text(
        "SELECT t2i_prompt, description FROM entity_canon WHERE id = :c"
    ), {"c": cid}).fetchone()
    assert row[0] == "conductor uniform", f"빈 칸을 안 채웠다: {row[0]}"
    assert row[1] == "2화 설명"


def test_1화_링크는_2화_sync_뒤에도_그대로(pg_session, scene):
    """C/L/P sync 는 원래 링크만 지운다 — 그 계약이 유지되는지 같이 본다."""
    pid, ep1, ep2, cid, root = scene
    _write_cp(root, pid, ep2, _cp("기사 최씨", "d", "t", []))
    _sync(pg_session, pid, ep2)

    n = pg_session.execute(sql_text(
        "SELECT count(*) FROM entity_episode_link "
        "WHERE canon_id = :c AND episode_id = :e"), {"c": cid, "e": ep1}).fetchone()[0]
    assert n == 1, "1화 링크가 2화 sync 로 사라졌다"