"""Create two live P1 sessions and verify DB-backed openness curves.

This is an evidence smoke for the backlog item "저항엔진 openness 곡선 DB 실증".
It uses the public API surface for login/onboarding/session turns, then queries
Postgres for the stored deterministic state and fast-loop client-state labels.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from http.cookiejar import CookieJar
from pathlib import Path
from typing import Any

import asyncpg


EMPATHIC_UTTERANCES = [
    "얼마나 힘들었는지 마음이 느껴져요. 어떤 순간이 제일 버거웠나요?",
    "그런 마음을 꺼내는 것 자체가 쉽지 않았을 것 같아요. 더 말해줘도 괜찮아요.",
    "잠도 잘 못 자고 학교도 버거웠다면 하루가 길게 느껴졌겠어요.",
    "지금은 해결책보다 그 마음을 천천히 이해하는 게 먼저인 것 같아요.",
    "그 시간을 버텨온 마음을 함께 살펴보고 싶어요. 무엇부터 이야기해볼까요?",
]

ADVICE_JUMP_UTTERANCES = [
    "그냥 학교는 가야 해요. 노력하면 하면 돼요. 왜 안 하죠?",
    "그건 잘못 생각하는 거예요. 원래 다 힘들어요.",
    "당연히 엄마 말을 들어야죠. 하지 마세요.",
    "내 생각엔 그냥 계획표를 만들면 돼요.",
    "그러니까 더 노력해야 해요. 왜 안 바꾸나요?",
]


class SmokeError(RuntimeError):
    pass


class ApiClient:
    def __init__(self, base_url: str, timeout: float) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self._opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(CookieJar()))

    def request(self, method: str, path: str, payload: dict[str, Any] | None = None) -> Any:
        data = None
        headers = {"Accept": "application/json"}
        if payload is not None:
            data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
            headers["Content-Type"] = "application/json"
        req = urllib.request.Request(
            f"{self.base_url}{path}",
            data=data,
            headers=headers,
            method=method,
        )
        try:
            with self._opener.open(req, timeout=self.timeout) as resp:
                body = resp.read().decode("utf-8")
                return json.loads(body) if body else {}
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode("utf-8", errors="replace")
            raise SmokeError(f"{method} {path} failed with HTTP {exc.code}: {detail}") from exc
        except urllib.error.URLError as exc:
            raise SmokeError(f"{method} {path} transport failed: {exc}") from exc


def _load_api_env() -> None:
    env_path = Path("apps/api/.env")
    if not env_path.exists():
        return
    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        os.environ.setdefault(key, value)


def _onboarding_payload(display_name: str) -> dict[str, Any]:
    return {
        "legal_name": display_name,
        "affiliation": "한신대학교",
        "department": "상담심리학과",
        "grade_level": "3학년",
        "phone": "010-2222-2222",
        "contact_address": "경기도 오산시 한신대학교",
        "nickname": display_name,
        "self_introduction": "저항엔진 DB 실증용 스모크 사용자입니다.",
        "avatar_url": "",
        "terms_accepted": True,
        "privacy_accepted": True,
    }


def _start_session(client: ApiClient, email: str, display_name: str) -> str:
    client.request(
        "POST",
        "/auth/dev-login",
        {"email": email, "role": "learner", "display_name": display_name},
    )
    client.request("POST", "/users/me/onboarding", _onboarding_payload(display_name))
    started = client.request(
        "POST",
        "/sessions",
        {"persona_code": "P1", "theory_mode": "humanistic"},
    )
    session_id = str(started.get("session_id") or "")
    if not session_id:
        raise SmokeError(f"session start returned no session_id: {started}")
    if started.get("degraded"):
        raise SmokeError(f"session start was degraded, refusing to use it as DB proof: {started}")
    return session_id


def _run_turns(client: ApiClient, session_id: str, utterances: list[str]) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    for text in utterances:
        results.append(client.request("POST", f"/sessions/{session_id}/turn", {"text": text}))
    return results


async def _fetch_curve(dsn: str, session_id: str) -> dict[str, Any]:
    conn = await asyncpg.connect(dsn)
    try:
        states = await conn.fetch(
            """
            SELECT session_id::text AS session_id, stage, turn_seq, effective_openness,
                   rapport_credit, resistance
            FROM app.session_state
            WHERE session_id = $1::uuid
            """,
            session_id,
        )
        turns = await conn.fetch(
            """
            SELECT t.id::text AS turn_id, t.seq AS turn_seq, t.actor_kind, t.text_masked,
                   array_remove(array_agg(cs.code ORDER BY cs.code), NULL) AS client_states
            FROM app.turns t
            LEFT JOIN app.turn_client_state tcs ON tcs.turn_id = t.id
            LEFT JOIN app.client_state_def cs ON cs.label_id = tcs.label_id
            WHERE t.session_id = $1::uuid
            GROUP BY t.id, t.seq, t.actor_kind, t.text_masked
            ORDER BY t.seq, t.actor_kind
            """,
            session_id,
        )
        state = dict(states[0]) if states else {}
        return {
            "session_state": state,
            "turns": [dict(row) for row in turns],
        }
    finally:
        await conn.close()


def _summarize(
    label: str,
    session_id: str,
    api_results: list[dict[str, Any]],
    db_result: dict[str, Any],
) -> dict[str, Any]:
    return {
        "label": label,
        "session_id": session_id,
        "api_curve": [
            {
                "turn_seq": item.get("turn_seq"),
                "stage": item.get("stage"),
                "effective_openness": item.get("effective_openness"),
                "safety_flagged": item.get("safety_flagged"),
            }
            for item in api_results
        ],
        "db_state": db_result.get("session_state"),
        "db_turn_count": len(db_result.get("turns") or []),
        "db_client_states": [
            {
                "turn_seq": row.get("turn_seq"),
                "actor_kind": row.get("actor_kind"),
                "client_states": row.get("client_states") or [],
            }
            for row in db_result.get("turns") or []
            if row.get("client_states")
        ],
    }


def _openness_curve(api_results: list[dict[str, Any]]) -> list[float]:
    return [float(item.get("effective_openness") or 0.0) for item in api_results]


def _stage_curve(api_results: list[dict[str, Any]]) -> list[str]:
    return [str(item.get("stage") or "") for item in api_results]


def _assert_turn_sequence(label: str, api_results: list[dict[str, Any]], expected_count: int) -> None:
    if len(api_results) != expected_count:
        raise SmokeError(f"{label} API returned {len(api_results)} turns, expected {expected_count}")
    expected_turns = list(range(1, expected_count + 1))
    actual_turns = [int(item.get("turn_seq") or 0) for item in api_results]
    if actual_turns != expected_turns:
        raise SmokeError(f"{label} API turn sequence mismatch: {actual_turns} != {expected_turns}")
    flagged_turns = [item.get("turn_seq") for item in api_results if item.get("safety_flagged")]
    if flagged_turns:
        raise SmokeError(f"{label} API unexpectedly triggered safety flags on turns {flagged_turns}")


def _assert_db_storage(label: str, db_result: dict[str, Any], expected_state_turn_seq: int) -> None:
    state = db_result.get("session_state") or {}
    if int(state.get("turn_seq") or 0) != expected_state_turn_seq:
        raise SmokeError(f"{label} DB session_state turn_seq mismatch: {state}")

    turns = db_result.get("turns") or []
    expected_turn_rows = expected_state_turn_seq * 2
    if len(turns) != expected_turn_rows:
        raise SmokeError(f"{label} DB stored {len(turns)} turns, expected {expected_turn_rows}")

    state_rows = [row for row in turns if row.get("client_states")]
    if len(state_rows) < expected_state_turn_seq:
        raise SmokeError(
            f"{label} DB stored {len(state_rows)} turn_client_state rows, expected at least {expected_state_turn_seq}"
        )


async def run(args: argparse.Namespace) -> dict[str, Any]:
    _load_api_env()
    dsn = args.database_url or os.environ.get("DATABASE_URL")
    if not dsn:
        raise SmokeError("DATABASE_URL is required via --database-url or apps/api/.env")

    suffix = str(int(time.time()))
    empathy_client = ApiClient(args.api_base_url, args.timeout)
    advice_client = ApiClient(args.api_base_url, args.timeout)

    empathy_session = _start_session(
        empathy_client,
        f"resistance.empathy.{suffix}@hs.ac.kr",
        "Resistance Empathy",
    )
    advice_session = _start_session(
        advice_client,
        f"resistance.advice.{suffix}@hs.ac.kr",
        "Resistance Advice",
    )

    empathy_results = _run_turns(empathy_client, empathy_session, EMPATHIC_UTTERANCES)
    advice_results = _run_turns(advice_client, advice_session, ADVICE_JUMP_UTTERANCES)
    empathy_db = await _fetch_curve(dsn, empathy_session)
    advice_db = await _fetch_curve(dsn, advice_session)

    expected_turns = len(EMPATHIC_UTTERANCES)
    if len(ADVICE_JUMP_UTTERANCES) != expected_turns:
        raise SmokeError("empathy/advice utterance sets must have the same turn count")

    _assert_turn_sequence("empathy", empathy_results, expected_turns)
    _assert_turn_sequence("advice_jump", advice_results, expected_turns)
    _assert_db_storage("empathy", empathy_db, expected_turns)
    _assert_db_storage("advice_jump", advice_db, expected_turns)

    empathy_curve = _openness_curve(empathy_results)
    advice_curve = _openness_curve(advice_results)
    empathy_stages = _stage_curve(empathy_results)
    advice_stages = _stage_curve(advice_results)

    empathy_final = float((empathy_db.get("session_state") or {}).get("effective_openness") or 0.0)
    advice_final = float((advice_db.get("session_state") or {}).get("effective_openness") or 0.0)
    empathy_stage = str((empathy_db.get("session_state") or {}).get("stage") or "")
    advice_stage = str((advice_db.get("session_state") or {}).get("stage") or "")

    if empathy_stages[2:] != ["탐색"] * (expected_turns - 2):
        raise SmokeError(f"expected empathy stage 탐색 from turn 3, got {empathy_stages}")
    if empathy_curve[2] <= 0 or empathy_curve[-1] <= empathy_curve[2]:
        raise SmokeError(f"expected empathy openness to open from turn 3 and keep rising, got {empathy_curve}")
    if advice_stages != ["라포"] * expected_turns:
        raise SmokeError(f"expected advice_jump stage 라포 throughout, got {advice_stages}")
    if any(value != 0.0 for value in advice_curve):
        raise SmokeError(f"expected advice_jump openness to stay at 0.0, got {advice_curve}")
    if empathy_final <= advice_final:
        raise SmokeError(f"expected empathy openness > advice openness, got {empathy_final} <= {advice_final}")
    if empathy_stage != "탐색" or empathy_final < 0.1:
        raise SmokeError(f"empathy DB curve did not open enough: stage={empathy_stage}, openness={empathy_final}")
    if advice_stage != "라포" or advice_final != 0.0:
        raise SmokeError(f"advice_jump DB curve did not stay closed: stage={advice_stage}, openness={advice_final}")

    return {
        "ok": True,
        "api_base_url": args.api_base_url,
        "persona_code": "P1",
        "empathy": _summarize("empathy", empathy_session, empathy_results, empathy_db),
        "advice_jump": _summarize("advice_jump", advice_session, advice_results, advice_db),
        "assertion": {
            "empathy_final_openness": empathy_final,
            "advice_final_openness": advice_final,
            "empathy_stage": empathy_stage,
            "advice_stage": advice_stage,
            "empathy_api_curve": empathy_curve,
            "advice_jump_api_curve": advice_curve,
            "db_turn_rows_per_session": expected_turns * 2,
        },
    }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--api-base-url", default="http://127.0.0.1:8000")
    parser.add_argument("--database-url", default="")
    parser.add_argument("--timeout", type=float, default=180.0)
    parser.add_argument("--out", default="")
    args = parser.parse_args()

    result = asyncio.run(run(args))
    text = json.dumps(result, ensure_ascii=False, indent=2, default=str)
    if args.out:
        out_path = Path(args.out)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(text + "\n", encoding="utf-8")
    print(text)


if __name__ == "__main__":
    main()