"""★끝점 — 실제 사슬: 같은 이름·같은 갈래(location_part)의 두 실체(다른 short_id · 다른 부모)가
EntityDetailStep._execute → EntityT2iStep._execute → EntitySyncService → RelationSyncService 를 지나
DB 에 canon 둘과 part_of 둘로 남는다 (Codex 계약 2026-09-03 06:40 · 실측 f7cc45c576c0 는 하나가 접혀 pre-sync 가 섰다).
이름·시나리오 문자열을 계약에 넣지 않는다 — fixture 낱말은 일반어."""
from __future__ import annotations

import json
import re
import uuid
from datetime import datetime, timezone
from pathlib import Path

import pytest

import app.core.steps.entity_steps as es
from app.core.steps.entity_steps import EntityDetailStep, EntityT2iStep
from app.modules.pipeline.grounding_entity_sync_ext import CHUNK_SCHEMA_MARKER, CHUNK_SCHEMA_VERSION

NEUTRAL = {"location": None, "visual_identity": None}
LOC_META = {"location": {"space_profile": {"kind": "single_space", "allowed_space_keys": ["main"], "default_space_key": None}},
            "visual_identity": None}


def _t2i_answer(kw):
    """장소는 location 모양, 부분은 중립 모양 — production 의 post-validate 가 요구하는 대로."""
    text = str(kw)
    meta = NEUTRAL if "location_part" in text else LOC_META
    return {"t2i_prompt": "p", "metadata_json": meta}


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


def _marked(data):
    return {"status": "completed", CHUNK_SCHEMA_MARKER: CHUNK_SCHEMA_VERSION, "data": data}


@pytest.fixture
def seeded(tmp_path, monkeypatch):
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    from app.core.database import SessionLocal, init_db
    from app.models.catalog import ProjectRegistry, UserAccount
    from app.models.project import Episode
    init_db()
    session = SessionLocal()
    try:
        now = datetime.now(timezone.utc).isoformat()
        uid = f"test-user-{uuid.uuid4()}"; pid = f"test-idc-{uuid.uuid4()}"; eid = f"test-idc-ep-{uuid.uuid4()}"
        session.add(UserAccount(id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1,
                                created_at=now, updated_at=now)); session.flush()
        session.add(ProjectRegistry(id=pid, name="idc", description="", created_by=uid, created_at=now, updated_at=now)); session.flush()
        session.add(Episode(id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt",
                            language="ko", status="uploaded", created_at=now, updated_at=now)); session.flush()
        yield session, pid, eid, tmp_path
    finally:
        session.rollback(); session.close()


def _echo_detail(kw):
    ents = []
    text = str(kw.get("user_prompt") or kw.get("prompt") or "") + str(kw)
    for m in re.finditer(r"- \[(\S+)\] (.+?) \((\S+)\)", text):
        ents.append({"entity_id": m.group(1), "name": m.group(2), "entity_type": m.group(3),
                     "description": f"d-{m.group(1)}", "visual_traits": ["v"], "distinctive_visual_traits": []})
    return {"entities": ents}


def _planning(monkeypatch):
    class _P:
        has_planning_doc = False
        is_first_episode = False
        def inject_if_available(self, section, header=""):
            return ""
    monkeypatch.setattr("app.core.planning_doc_context.get_planning_context", lambda pid, eid, db=None: _P())


class TestTheChainKeepsBothInstancesToTheDatabase:
    def test_two_same_named_parts_under_different_wholes_reach_canon_and_part_of(self, seeded, monkeypatch):
        db, pid, eid, tmp = seeded
        # 앞쪽 산출: 장소 둘 · 같은 이름의 부분 둘(다른 short_id · 다른 부모) · part_of 관계 둘
        locs = [{"name": "가게", "short_id": "L03"}, {"name": "길", "short_id": "L02"}]
        parts = [{"name": "바닥", "short_id": "LP05"}, {"name": "바닥", "short_id": "LP07"}]
        _cp(tmp, pid, eid, "entity_merge", _marked({"removed": [], "locations": locs, "location_parts": parts}))
        _cp(tmp, pid, eid, "entity_filter", _marked({"filtered_entities": {"locations": locs, "location_parts": parts}}))
        _cp(tmp, pid, eid, "entity_relation", _marked({"relations": [
            {"relation_type": "part_of", "participants": [{"short_id": "LP05", "role": "part", "order": 1}, {"short_id": "L03", "role": "whole", "order": 2}]},
            {"relation_type": "part_of", "participants": [{"short_id": "LP07", "role": "part", "order": 1}, {"short_id": "L02", "role": "whole", "order": 2}]}]}))
        _planning(monkeypatch)
        monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_prompt", lambda name, **k: f"{name}:{k.get('entity_list','')}")
        monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_schema", lambda name: {"type": "object"})

        # ① entity_detail (진짜 스텝 · 모델 대역은 표식을 echo)
        calls = []
        monkeypatch.setattr(es, "call_structured", lambda **kw: (calls.append(kw) or _echo_detail(kw)), raising=False)
        det = EntityDetailStep(pid, eid, db, {"grounding_mode": "v2_chunk"}) if _accepts_ctor() else None
        if det is None:
            det = EntityDetailStep.__new__(EntityDetailStep)
            det.project_id, det.episode_id, det.db, det.project_config = pid, eid, db, {"grounding_mode": "v2_chunk"}
            det.build_opik_metadata = lambda *a, **k: {}
        det._load_cleaned_text = lambda: "본문"
        det._journal = lambda *a, **k: None
        out_d = det._execute()
        assert [x[2] for x in out_d["data"]["entity_queue"] if x[1] == "location_part"] == ["LP05", "LP07"]
        _cp(tmp, pid, eid, "entity_detail", {**_marked(out_d["data"]), "config_hash": out_d.get("config_hash")})

        # ② entity_t2i (진짜 스텝 · 모델 대역)
        monkeypatch.setattr(es, "call_structured", lambda **kw: _t2i_answer(kw), raising=False)
        monkeypatch.setattr(es, "_load_system", lambda: "sys", raising=False)
        monkeypatch.setattr(es, "time", type("T", (), {"sleep": staticmethod(lambda s: None)})(), raising=False)
        t2i = EntityT2iStep.__new__(EntityT2iStep)
        t2i.project_id, t2i.episode_id, t2i.db, t2i.project_config = pid, eid, db, {"grounding_mode": "v2_chunk"}
        t2i.build_opik_metadata = lambda *a, **k: {}
        t2i.load_checkpoint = lambda: None
        t2i.save_checkpoint = lambda *a, **k: None
        t2i.update_progress = lambda *a, **k: None
        out_t = t2i._execute()
        done = out_t["data"]["completed"]
        assert {"LP05", "LP07", "L02", "L03"} <= set(done), sorted(done)
        _cp(tmp, pid, eid, "entity_t2i", {**_marked(out_t["data"]), "config_hash": out_t.get("config_hash")})

        # ③ sync — canon 넷 · part_of 둘 · pre-sync(관계 동기화) 가 서지 않는다
        from app.services.checkpoint_sync.entity_sync_service import EntitySyncService
        from app.services.checkpoint_sync.relation_sync_service import RelationSyncService
        from app.models.project import EntityCanon, RelationFact
        EntitySyncService(db, pid, eid).sync_from_checkpoint(); db.flush()
        sids = {c.short_id for c in db.query(EntityCanon).filter(EntityCanon.project_id == pid).all()}
        assert {"LP05", "LP07", "L02", "L03"} <= sids, sids
        RelationSyncService(db, pid, eid).sync_from_checkpoint(); db.flush()
        facts = db.query(RelationFact).filter(RelationFact.project_id == pid, RelationFact.relation_type == "part_of").all()
        assert len(facts) == 2, len(facts)


def _accepts_ctor() -> bool:
    return False