"""SceneStillWriter 단위 테스트 — W4 P3-1d.

Base.metadata.create_all로 in-memory SQLite에 전체 모델 생성. SQLite는
기본적으로 FK를 enforce하지 않으므로 project/episode row 없이도 insert 가능.
Writer가 ORM query/setattr 패턴을 유지하므로 실제 모델 스키마가 필요.
"""
from __future__ import annotations

import json
import uuid
from typing import Dict, List, Optional

import pytest
from sqlalchemy import create_engine, text as sql_text
from sqlalchemy.orm import Session


@pytest.fixture
def engine_db():
    from app.core.database import Base
    # import all models so Base.metadata includes every table
    import app.models.project  # noqa: F401
    import app.models.catalog  # noqa: F401

    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    with Session(engine) as db:
        yield engine, db


@pytest.fixture
def writer(engine_db):
    from app.services.checkpoint_sync.scene_still_writer import SceneStillWriter

    _, db = engine_db
    return SceneStillWriter(db, "proj-1", "ep-1", now="2026-04-22T00:00:00+00:00")


def _bundle(**kw):
    from app.services.checkpoint_sync._scene_still_contracts import CheckpointBundle

    b = CheckpointBundle()
    for k, v in kw.items():
        setattr(b, k, v)
    return b


def _planned(key=(1, 1), still_index=1, scene_index=1, version="scene_extractor_v4", columns=None):
    from app.services.checkpoint_sync._scene_still_contracts import PlannedStill

    return PlannedStill(
        key=key, still_index=still_index, scene_index=scene_index,
        t2i_composer_version=version,
        columns=columns or {
            "screenplay_scene_heading": "S#1",
            "beat_title": "beat",
            "still_frame_prompt": "prompt",
            "visible_entities_json": "[]",
            "t2i_prompt_cinematic": "",
            "t2i_variations_json": None,
            "scene_type": "normal",
            "shot_index": 1,
            "shot_description": "desc",
            "based_on_beat": None,
            "audio_entity_ids": "[]",
            "hallucination_entity_ids": "[]",
            "is_selected": True,
        },
    )


def _count_rows(db) -> int:
    return db.execute(sql_text("SELECT COUNT(*) FROM scene_still")).scalar()


# ── Insert 경로 ──


def test_write_insert_creates_new_rows(writer, engine_db):
    _, db = engine_db
    result = writer.write([_planned()], _bundle())
    db.commit()
    assert result == 1
    assert _count_rows(db) == 1

    row = db.execute(sql_text("SELECT scene_index, shot_index, still_index, t2i_composer_version FROM scene_still")).fetchone()
    assert (row[0], row[1], row[2], row[3]) == (1, 1, 1, "scene_extractor_v4")


# ── Update 경로 — 기존 ID 보존 ──


def test_write_update_preserves_existing_id(writer, engine_db):
    _, db = engine_db
    existing_id = str(uuid.uuid4())
    db.execute(sql_text(
        "INSERT INTO scene_still (id, project_id, episode_id, scene_index, shot_index, still_index, still_frame_prompt, created_at) "
        "VALUES (:id, :pid, :eid, 1, 1, 999, 'old', '2025-01-01')"
    ), {"id": existing_id, "pid": "proj-1", "eid": "ep-1"})
    db.commit()

    writer.write([_planned(columns={
        "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "new",
        "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
        "scene_type": "normal", "shot_index": 1, "shot_description": "",
        "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
        "is_selected": True,
    })], _bundle())
    db.commit()

    rows = db.execute(sql_text("SELECT id, still_frame_prompt, still_index FROM scene_still")).fetchall()
    assert len(rows) == 1
    assert rows[0][0] == existing_id  # ID 보존
    assert rows[0][1] == "new"
    assert rows[0][2] == 1  # 새 still_index


# ── Stale 마킹 ──


def test_write_marks_stale_for_rows_not_in_plan(writer, engine_db):
    _, db = engine_db
    stale_id = str(uuid.uuid4())
    db.execute(sql_text(
        "INSERT INTO scene_still (id, project_id, episode_id, scene_index, shot_index, still_index, created_at) "
        "VALUES (:id, 'proj-1', 'ep-1', 9, 9, 5, '2025')"
    ), {"id": stale_id})
    db.commit()

    writer.write([_planned(key=(1, 1))], _bundle())
    db.commit()

    stale_row = db.execute(sql_text("SELECT still_index, status FROM scene_still WHERE id = :id"),
                           {"id": stale_id}).fetchone()
    assert stale_row[0] == -1
    assert stale_row[1] == "stale"


# ── dependent_scene_id (shot-based) ──


def test_write_dependent_scene_id_from_shot_deps(writer, engine_db):
    _, db = engine_db
    planned = [
        _planned(key=(1, 1), still_index=1, scene_index=1),
        _planned(key=(2, 1), still_index=2, scene_index=2, columns={
            "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "",
            "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
            "scene_type": "normal", "shot_index": 1, "shot_description": "",
            "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
            "is_selected": True,
        }),
    ]
    bundle = _bundle(
        shot_dep_completed=True,
        shot_deps=[{
            "scene_index": 2, "shot_index": 1,
            "location_refs": [{"scene_index": 1, "shot_index": 1}],
        }],
    )
    writer.write(planned, bundle)
    db.commit()

    rows = db.execute(sql_text(
        "SELECT scene_index, shot_index, dependent_scene_id FROM scene_still ORDER BY scene_index"
    )).fetchall()
    target_id = rows[0][2] is None, rows[1][2] is not None
    # Scene 1 has no dep, Scene 2 points to Scene 1's still
    scene1_id = db.execute(sql_text(
        "SELECT id FROM scene_still WHERE scene_index = 1 AND shot_index = 1"
    )).scalar()
    assert rows[1][2] == scene1_id


def test_write_clears_dependent_scene_id_when_shot_dep_completed(writer, engine_db):
    """shot_dep_completed면 기존 dependent_scene_id를 먼저 NULL로 초기화."""
    _, db = engine_db
    db.execute(sql_text(
        "INSERT INTO scene_still (id, project_id, episode_id, scene_index, shot_index, still_index, dependent_scene_id, created_at) "
        "VALUES ('s1', 'proj-1', 'ep-1', 1, 1, 1, 'old-dep', '2025')"
    ))
    db.commit()

    writer.write([_planned(key=(1, 1))], _bundle(shot_dep_completed=True, shot_deps=[]))
    db.commit()

    dep = db.execute(sql_text("SELECT dependent_scene_id FROM scene_still WHERE id = 's1'")).scalar()
    assert dep is None


# ── dependent_scene_id (legacy scene_dependency fallback) ──


def test_write_legacy_dep_map_fallback(writer, engine_db):
    """shot_deps 없고 dep_map 있으면 legacy 경로 사용 (scene당 1 still)."""
    _, db = engine_db
    planned = [
        _planned(key=(1, None), still_index=1, scene_index=1, version="scene_extractor_v3", columns={
            "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "",
            "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
            "scene_type": "normal", "shot_index": None, "shot_description": "",
            "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
            "is_selected": True,
        }),
        _planned(key=(2, None), still_index=2, scene_index=2, version="scene_extractor_v3", columns={
            "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "",
            "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
            "scene_type": "normal", "shot_index": None, "shot_description": "",
            "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
            "is_selected": True,
        }),
    ]
    bundle = _bundle(
        scenes=[{"scene_index": 1}, {"scene_index": 2}],
        dep_map={"2": {"location_refs": [{"scene_index": 1}]}},
    )
    writer.write(planned, bundle)
    db.commit()

    scene1_id = db.execute(sql_text(
        "SELECT id FROM scene_still WHERE scene_index = 1 AND shot_index IS NULL"
    )).scalar()
    scene2_dep = db.execute(sql_text(
        "SELECT dependent_scene_id FROM scene_still WHERE scene_index = 2 AND shot_index IS NULL"
    )).scalar()
    assert scene2_dep == scene1_id


# ── scene_summary 전파 ──


def test_write_applies_scene_summary(writer, engine_db):
    _, db = engine_db
    planned = [
        _planned(key=(1, 1), still_index=1, scene_index=1),
        _planned(key=(1, 2), still_index=2, scene_index=1, columns={
            "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "",
            "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
            "scene_type": "normal", "shot_index": 2, "shot_description": "",
            "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
            "is_selected": True,
        }),
    ]
    bundle = _bundle(scene_summaries=[{"scene_index": 1, "scene_summary": "S1 요약"}])
    writer.write(planned, bundle)
    db.commit()

    rows = db.execute(sql_text(
        "SELECT scene_summary FROM scene_still WHERE scene_index = 1 ORDER BY shot_index"
    )).fetchall()
    assert [r[0] for r in rows] == ["S1 요약", "S1 요약"]


# ── shot_type ──


def test_write_shot_type_from_shot_cine(writer, engine_db):
    _, db = engine_db
    planned = [_planned(key=(1, 1))]
    bundle = _bundle(shot_cine_shots=[{
        "scene_index": 1, "shot_index": 1,
        "technique_1": {"name": "wide"}, "technique_2": {"name": "close"},
    }])
    writer.write(planned, bundle)
    db.commit()

    row = db.execute(sql_text(
        "SELECT shot_type_1, shot_type_2 FROM scene_still WHERE scene_index = 1 AND shot_index = 1"
    )).fetchone()
    assert row == ("wide", "close")


def test_write_shot_type_from_legacy_scene_cine(writer, engine_db):
    """shot_cine가 비어 있고 scene_cine만 있으면 legacy 경로 (scene 단위 업데이트)."""
    _, db = engine_db
    planned = [_planned(key=(1, None), still_index=1, scene_index=1, version="scene_extractor_v3", columns={
        "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "",
        "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
        "scene_type": "normal", "shot_index": None, "shot_description": "",
        "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
        "is_selected": True,
    })]
    bundle = _bundle(scene_cine_scenes=[{
        "scene_index": 1,
        "shots": [{"name": "master"}, {"name": "over"}],
    }])
    writer.write(planned, bundle)
    db.commit()

    row = db.execute(sql_text(
        "SELECT shot_type_1, shot_type_2 FROM scene_still WHERE scene_index = 1"
    )).fetchone()
    assert row == ("master", "over")


# ── Empty plan ──


def test_write_update_preserves_t2i_composer_version(writer, engine_db):
    """Codex P3-1 High: baseline은 UPDATE 시 t2i_composer_version을 덮지 않음.

    기존 v3 row가 있을 때 normalizer가 scene_extractor_v4로 계획을 내놓아도
    composer_version은 원래 값이 유지돼야 함 (선택↔미선택 토글 회귀 방지).
    """
    _, db = engine_db
    db.execute(sql_text(
        "INSERT INTO scene_still (id, project_id, episode_id, scene_index, shot_index, "
        "still_index, t2i_composer_version, created_at) "
        "VALUES ('fixed-id', 'proj-1', 'ep-1', 1, 1, 1, 'baseline_v7', '2025-01-01')"
    ))
    db.commit()

    writer.write([_planned()], _bundle())  # _planned()는 scene_extractor_v4 기본
    db.commit()

    ver = db.execute(sql_text(
        "SELECT t2i_composer_version FROM scene_still WHERE id = 'fixed-id'"
    )).scalar()
    assert ver == "baseline_v7"


def test_write_legacy_dep_prev_ref_form(writer, engine_db):
    """Codex P3-1 Low: scene_dependency의 prev_ref(int) 형식 처리."""
    _, db = engine_db
    planned = [
        _planned(key=(1, None), still_index=1, scene_index=1, version="scene_extractor_v3", columns={
            "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "",
            "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
            "scene_type": "normal", "shot_index": None, "shot_description": "",
            "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
            "is_selected": True,
        }),
        _planned(key=(2, None), still_index=2, scene_index=2, version="scene_extractor_v3", columns={
            "screenplay_scene_heading": "", "beat_title": "", "still_frame_prompt": "",
            "visible_entities_json": "[]", "t2i_prompt_cinematic": "", "t2i_variations_json": None,
            "scene_type": "normal", "shot_index": None, "shot_description": "",
            "based_on_beat": None, "audio_entity_ids": "[]", "hallucination_entity_ids": "[]",
            "is_selected": True,
        }),
    ]
    # loc_refs 없이 prev_ref=1 (int) 단독 — legacy 포맷
    bundle = _bundle(
        scenes=[{"scene_index": 1}, {"scene_index": 2}],
        dep_map={"2": {"prev_ref": 1}},
    )
    writer.write(planned, bundle)
    db.commit()

    scene1_id = db.execute(sql_text(
        "SELECT id FROM scene_still WHERE scene_index = 1 AND shot_index IS NULL"
    )).scalar()
    scene2_dep = db.execute(sql_text(
        "SELECT dependent_scene_id FROM scene_still WHERE scene_index = 2 AND shot_index IS NULL"
    )).scalar()
    assert scene2_dep == scene1_id


def test_write_shot_cine_wins_when_both_present(writer, engine_db):
    """Codex P3-1 Low: shot_cine_shots + scene_cine_scenes 모두 있을 때 shot_cine 우선."""
    _, db = engine_db
    writer.write([_planned()], _bundle(
        shot_cine_shots=[{
            "scene_index": 1, "shot_index": 1,
            "technique_1": {"name": "new_t1"}, "technique_2": {"name": "new_t2"},
        }],
        scene_cine_scenes=[{
            "scene_index": 1,
            "shots": [{"name": "legacy_t1"}, {"name": "legacy_t2"}],
        }],
    ))
    db.commit()
    row = db.execute(sql_text(
        "SELECT shot_type_1, shot_type_2 FROM scene_still WHERE scene_index = 1 AND shot_index = 1"
    )).fetchone()
    assert row == ("new_t1", "new_t2")


def test_write_empty_plan_still_applies_summary(writer, engine_db):
    """planned=[] 이어도 scene_summary/shot_type 갱신은 수행."""
    _, db = engine_db
    db.execute(sql_text(
        "INSERT INTO scene_still (id, project_id, episode_id, scene_index, shot_index, still_index, created_at) "
        "VALUES ('s1', 'proj-1', 'ep-1', 1, 1, 1, '2025')"
    ))
    db.commit()

    bundle = _bundle(scene_summaries=[{"scene_index": 1, "scene_summary": "만"}])
    count = writer.write([], bundle)
    db.commit()
    assert count == 0
    sv = db.execute(sql_text("SELECT scene_summary FROM scene_still WHERE id = 's1'")).scalar()
    assert sv == "만"