"""반복 출제 / 유사 유형 분류 service (PR-12-A).

학습 의미가 회차 기반이라 응답이 곧 의미. 회차 조건은 백엔드가 강제한다.
단건 endpoint(`/related-types`) + 배치 endpoint(`/related-types-bulk`) 가
같은 service 함수를 공유해 threshold/round_count 정의가 drift 나지 않게 한다.

분류:
  🔥 반복 출제 — similarity >= 0.95, 다른 exam_round
  🧩 유사 유형 — 0.88 <= similarity < 0.95, 다른 exam_round

round_count 정의:
  관련 후보가 0 이면 round_count 도 0 (현재 회차만 1 로 채우지 않는다).
  관련 후보가 있으면 distinct(candidate.exam_round) ∪ {source.exam_round} 의 크기.
"""

from __future__ import annotations

from dataclasses import dataclass

from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from models.study_question import StudyQuestion

# 임계값 / 응답 cap (튜닝 후보 — 한 곳에서만 정의)
# 5 source 분포 측정 결과 (2026-04-30, 2,100문항):
#   0.97+ (3건) ↔ 0.85~0.92 (4건) 사이에 0.92~0.97 자연 갭. 0.95 임계값 유지가 안전.
#   0.85~0.92 = 진짜 "유사 유형" (같은 개념 다른 표현). 0.88 → 0.85 로 살짝 낮춤.
REPEAT_THRESHOLD = 0.95
SIMILAR_THRESHOLD = 0.85
REPEAT_K = 10
SIMILAR_K = 8
RELATED_QUERY_LIMIT = 50  # cosine top-K 1차 fetch (회차 필터 + threshold split 후 K cap)

# 임계값 fingerprint — 임계값 변경 시 일괄 캐시 invalidate 키. 상수 갱신 후
# `UPDATE study_questions SET related_computed_at=NULL WHERE related_threshold_version != '<신>'`
# 한 번 돌리면 cron 이 자동 재계산.
THRESHOLD_VERSION = f"v1_{REPEAT_THRESHOLD}_{SIMILAR_THRESHOLD}"

# PR-12-B: 출제 단계 spacing (한 세션 안 같은 유형 과밀 방지). 회차 무관.
SPACING_THRESHOLD = SIMILAR_THRESHOLD  # 0.85
PER_TYPE_CAP = 2  # local neighbor cap (transitive cluster 보장 X)
SPACING_BUFFER_RATIO = 2.0  # 후보 buffer = target × 2.0

# 라벨 등급 — round_count 기반 (반복 출제 빈도). source 회차 + candidate 회차 distinct 합집합.
GRADE_DANGOL = "단골"  # ≥ 7개 회차 — 거의 매 회차 등장
GRADE_FREQUENT = "잘 나오는 반복 출제"  # 5~6
GRADE_REPEAT = "반복 출제"  # 3~4
GRADE_NEW = "신출"  # 2 + 최근 회차 (>= 2024) 포함
GRADE_OLD = "빈출"  # 2 + 모두 옛 회차 (< 2024) — 가끔 나오던 옛 유형

NEW_YEAR_CUTOFF = 2024  # 이 연도 이상 = "신출" (최근 도입)


@dataclass
class RelatedCandidate:
    """단건 endpoint 응답 item 직렬화 직전 단계."""
    id: int
    study_topic_id: int
    question_text: str
    subject: str | None
    scope: str | None
    exam_round: str | None
    exam_question_number: int | None
    similarity: float


@dataclass
class RelatedClassification:
    """단건 endpoint 응답에 그대로 매핑되는 결과 묶음."""
    repeat: list[RelatedCandidate]
    similar: list[RelatedCandidate]
    repeat_related_count: int
    repeat_round_count: int
    repeat_grade: str | None  # 단골/잘 나오는 반복 출제/반복 출제/신출/빈출 (또는 None)
    similar_related_count: int
    similar_round_count: int


def _extract_year(round_str: str | None) -> int | None:
    """'2024년 1회' 같은 회차 문자열에서 연도 추출."""
    if not round_str:
        return None
    import re as _re
    m = _re.search(r"(\d{4})", round_str)
    return int(m.group(1)) if m else None


def compute_repeat_grade(
    round_count: int, source_round: str | None, candidate_rounds: list[str]
) -> str | None:
    """반복 출제 등급. round_count = source 회차 포함 distinct 회차 수.

    - >= 7: 단골
    - 5~6: 잘 나오는 반복 출제
    - 3~4: 반복 출제
    - == 2: 신출 (candidate 중 NEW_YEAR_CUTOFF 이상 있으면) / 빈출 (모두 그 이전)
    - < 2: None (배지 표시 안 됨)
    """
    if round_count >= 7:
        return GRADE_DANGOL
    if round_count >= 5:
        return GRADE_FREQUENT
    if round_count >= 3:
        return GRADE_REPEAT
    if round_count == 2:
        # source + candidate 회차 중 NEW_YEAR_CUTOFF 이상이 하나라도 있으면 신출
        years: list[int] = []
        sy = _extract_year(source_round)
        if sy is not None:
            years.append(sy)
        for cr in candidate_rounds:
            cy = _extract_year(cr)
            if cy is not None:
                years.append(cy)
        if years and max(years) >= NEW_YEAR_CUTOFF:
            return GRADE_NEW
        return GRADE_OLD
    return None


def serialize_candidates(cands: list[RelatedCandidate]) -> list[dict]:
    """RelatedCandidate → JSONB 저장용 dict. study_topic_id 는 부모 row 와 동일하므로 생략."""
    return [
        {
            "id": c.id,
            "sim": round(c.similarity, 4),
            "exam_round": c.exam_round,
            "exam_question_number": c.exam_question_number,
            "subject": c.subject,
            "scope": c.scope,
            "qtext": c.question_text,  # 이미 80자 truncate 된 상태
        }
        for c in cands
    ]


def deserialize_candidates(rows: list[dict] | None, study_topic_id: int) -> list[RelatedCandidate]:
    """JSONB → RelatedCandidate. NULL 이면 빈 리스트."""
    if not rows:
        return []
    return [
        RelatedCandidate(
            id=int(r["id"]),
            study_topic_id=study_topic_id,
            question_text=r.get("qtext") or "",
            subject=r.get("subject"),
            scope=r.get("scope"),
            exam_round=r.get("exam_round"),
            exam_question_number=r.get("exam_question_number"),
            similarity=float(r.get("sim") or 0.0),
        )
        for r in rows
    ]


def _norm_round(s: str | None) -> str:
    return (s or "").strip()


def _truncate(text: str, n: int = 80) -> str:
    if not text:
        return ""
    s = text.strip()
    return s if len(s) <= n else s[:n].rstrip() + "…"


def _round_count(related: list[RelatedCandidate], source_round: str) -> int:
    """관련 후보가 없으면 0. 있으면 distinct(candidate.exam_round) ∪ {source} 크기.

    candidate.exam_round 는 이미 trim 후 source 와 다른 값으로 필터된 상태.
    """
    if not related:
        return 0
    rounds: set[str] = {_norm_round(source_round)}
    for c in related:
        nr = _norm_round(c.exam_round)
        if nr:
            rounds.add(nr)
    return len(rounds)


async def classify_related_for_question(
    session: AsyncSession,
    *,
    user_id: int,
    source: StudyQuestion,
) -> RelatedClassification:
    """단건 분류. source 의 임베딩으로 같은 토픽 안 다른 회차 candidate 끌어와 임계값 split.

    호출자는 source 의 ownership 을 미리 검증해야 한다. 본 함수는 user_id 비교만 강제.
    """
    src_round = _norm_round(source.exam_round)

    # 가드 1: 임베딩 미준비
    if source.embedding_status != "ready" or source.embedding is None:
        return RelatedClassification([], [], 0, 0, None, 0, 0)
    # 가드 2: source 회차 미지정 → 회차 간 반복성 분석 불가
    if not src_round:
        return RelatedClassification([], [], 0, 0, None, 0, 0)

    distance_expr = StudyQuestion.embedding.cosine_distance(source.embedding)
    stmt = (
        select(StudyQuestion, distance_expr.label("distance"))
        .where(
            StudyQuestion.id != source.id,
            StudyQuestion.user_id == user_id,
            StudyQuestion.study_topic_id == source.study_topic_id,
            StudyQuestion.deleted_at.is_(None),
            StudyQuestion.embedding_status == "ready",
            StudyQuestion.embedding.is_not(None),
            StudyQuestion.exam_round.is_not(None),
            func.btrim(StudyQuestion.exam_round) != "",
            func.btrim(StudyQuestion.exam_round) != src_round,
        )
        .order_by(distance_expr.asc())
        .limit(RELATED_QUERY_LIMIT)
    )
    rows = (await session.execute(stmt)).all()

    repeat: list[RelatedCandidate] = []
    similar: list[RelatedCandidate] = []
    for q, distance in rows:
        sim = 1.0 - float(distance)
        if sim >= REPEAT_THRESHOLD:
            target = repeat
        elif sim >= SIMILAR_THRESHOLD:
            target = similar
        else:
            # cosine asc 정렬이라 한 번 임계값 미만이면 이후도 다 미만.
            break
        target.append(RelatedCandidate(
            id=q.id,
            study_topic_id=q.study_topic_id,
            question_text=_truncate(q.question_text, 80),
            subject=q.subject,
            scope=q.scope,
            exam_round=q.exam_round,
            exam_question_number=q.exam_question_number,
            similarity=sim,
        ))

    # round_count 는 K cap 전 전체 후보 기준 (배지 일관성 유지용).
    repeat_round = _round_count(repeat, src_round)
    similar_round = _round_count(similar, src_round)

    # 응답 리스트만 K cap (count 는 cap 전 기준 유지).
    repeat_display = repeat[:REPEAT_K]
    similar_display = similar[:SIMILAR_K]

    # 라벨 등급 — repeat 카테고리 round_count 기반.
    repeat_grade = compute_repeat_grade(
        round_count=repeat_round,
        source_round=src_round,
        candidate_rounds=[c.exam_round or "" for c in repeat],
    )

    return RelatedClassification(
        repeat=repeat_display,
        similar=similar_display,
        repeat_related_count=len(repeat_display),
        repeat_round_count=repeat_round,
        repeat_grade=repeat_grade,
        similar_related_count=len(similar_display),
        similar_round_count=similar_round,
    )


@dataclass
class BulkCounts:
    repeat_related_count: int
    repeat_round_count: int
    repeat_grade: str | None
    similar_related_count: int
    similar_round_count: int


_ZERO_COUNTS = BulkCounts(0, 0, None, 0, 0)


async def classify_related_bulk(
    session: AsyncSession,
    *,
    user_id: int,
    study_topic_id: int,
    question_ids: list[int],
) -> dict[int, BulkCounts]:
    """배치 카운트. 입력 question_ids 끼리 비교가 아니라 토픽 전체 ready pool 과 비교.

    응답 dict 는 입력한 모든 qid 를 키로 보존 (권한 없음/임베딩 미준비/회차 미지정 시 0,0,0,0).
    """
    if not question_ids:
        return {}

    # 입력 qid 의 임베딩 + exam_round 모두 fetch (권한 + 토픽 일치 강제).
    input_rows = (
        await session.execute(
            select(
                StudyQuestion.id,
                StudyQuestion.embedding,
                StudyQuestion.embedding_status,
                StudyQuestion.exam_round,
            )
            .where(
                StudyQuestion.id.in_(question_ids),
                StudyQuestion.user_id == user_id,
                StudyQuestion.study_topic_id == study_topic_id,
                StudyQuestion.deleted_at.is_(None),
            )
        )
    ).all()
    input_map = {r.id: r for r in input_rows}

    # 토픽 전체 ready pool 한 번에 (회차 trim 후 non-empty).
    pool_rows = (
        await session.execute(
            select(
                StudyQuestion.id,
                StudyQuestion.embedding,
                StudyQuestion.exam_round,
            )
            .where(
                StudyQuestion.user_id == user_id,
                StudyQuestion.study_topic_id == study_topic_id,
                StudyQuestion.deleted_at.is_(None),
                StudyQuestion.embedding_status == "ready",
                StudyQuestion.embedding.is_not(None),
                StudyQuestion.exam_round.is_not(None),
                func.btrim(StudyQuestion.exam_round) != "",
            )
        )
    ).all()
    pool = [(r.id, list(r.embedding), _norm_round(r.exam_round)) for r in pool_rows]

    out: dict[int, BulkCounts] = {qid: _ZERO_COUNTS for qid in question_ids}

    if not pool:
        return out

    # cosine: a·b / (|a||b|). pool 벡터 norm 미리 계산.
    import math
    pool_norms = [math.sqrt(sum(v * v for v in vec)) or 1e-9 for _, vec, _ in pool]

    for qid in question_ids:
        row = input_map.get(qid)
        if row is None:
            continue
        if row.embedding_status != "ready" or row.embedding is None:
            continue
        src_round = _norm_round(row.exam_round)
        if not src_round:
            continue

        src_vec = list(row.embedding)
        src_norm = math.sqrt(sum(v * v for v in src_vec)) or 1e-9

        repeat_rounds: set[str] = set()
        similar_rounds: set[str] = set()
        repeat_n = 0
        similar_n = 0

        for (cid, cvec, cround), cnorm in zip(pool, pool_norms, strict=True):
            if cid == qid:
                continue
            if cround == src_round:
                continue
            dot = 0.0
            for a, b in zip(src_vec, cvec, strict=True):
                dot += a * b
            sim = dot / (src_norm * cnorm)
            if sim >= REPEAT_THRESHOLD:
                repeat_n += 1
                repeat_rounds.add(cround)
            elif sim >= SIMILAR_THRESHOLD:
                similar_n += 1
                similar_rounds.add(cround)

        repeat_round_count = (len(repeat_rounds) + 1) if repeat_n > 0 else 0
        repeat_grade = compute_repeat_grade(
            round_count=repeat_round_count,
            source_round=src_round,
            candidate_rounds=list(repeat_rounds),
        )
        out[qid] = BulkCounts(
            repeat_related_count=repeat_n,
            repeat_round_count=repeat_round_count,
            repeat_grade=repeat_grade,
            similar_related_count=similar_n,
            similar_round_count=(len(similar_rounds) + 1) if similar_n > 0 else 0,
        )

    return out


# ─── PR-12-B: 출제 단계 spacing ───


import logging
import math
from collections.abc import Sequence

logger = logging.getLogger(__name__)


@dataclass
class _SpacingCandidate:
    qid: int
    embedding: list | None  # ready 면 list, 아니면 None
    is_ready: bool


@dataclass
class SpacingResult:
    """apply_type_spacing 결과 + 운영 관찰용 메타."""
    selected_ids: list[int]  # 채택 순서 (target 까지 채워지면 거기까지)
    selected_from_ready: int  # ready spacing 으로 채택
    selected_from_pending: int  # 임베딩 미준비 보충
    selected_from_hold: int  # cap 위반 fallback
    held: int  # 최종까지 hold 인 채로 남은 개수


async def apply_type_spacing(
    candidates: Sequence[_SpacingCandidate],
    *,
    target: int,
    subject_label: str = "",
) -> SpacingResult:
    """한 subject bucket 안 후보 리스트에 type spacing 적용.

    - 회차 무관 (PR-12-A 의 회차 필터 로직 재사용 안 함, 자기 자신만 제외).
    - cluster ID 안 만듦 — 새 후보와 이미 채택된 ready 문제와의 cosine ≥ 0.88
      개수가 PER_TYPE_CAP 이상이면 hold (local neighbor cap, 단순화).
    - 3단계 fallback: ready spacing → 임베딩 미준비 보충 → hold cap 위반 fallback.
    - 임계값 미만이거나 매칭 0 인 후보는 즉시 채택.
    """
    if target <= 0 or not candidates:
        return SpacingResult([], 0, 0, 0, 0)

    selected_ids: list[int] = []
    selected_ready_vecs: list[tuple[int, list, float]] = []  # (qid, vec, norm)
    pending_pool: list[int] = []
    hold_pool: list[int] = []

    selected_from_ready = 0
    selected_from_pending = 0
    selected_from_hold = 0

    # 1단계: ready 후보 type spacing.
    for c in candidates:
        if len(selected_ids) >= target:
            break
        if not c.is_ready or c.embedding is None:
            pending_pool.append(c.qid)
            continue
        cand_vec = list(c.embedding)
        cand_norm = math.sqrt(sum(v * v for v in cand_vec)) or 1e-9

        similar_count = 0
        for _, sel_vec, sel_norm in selected_ready_vecs:
            dot = 0.0
            for a, b in zip(cand_vec, sel_vec, strict=True):
                dot += a * b
            sim = dot / (cand_norm * sel_norm)
            if sim >= SPACING_THRESHOLD:
                similar_count += 1
                if similar_count >= PER_TYPE_CAP:
                    break
        if similar_count >= PER_TYPE_CAP:
            hold_pool.append(c.qid)
        else:
            selected_ids.append(c.qid)
            selected_ready_vecs.append((c.qid, cand_vec, cand_norm))
            selected_from_ready += 1

    # 2단계: 임베딩 미준비 후보 보충 (cap 카운트 안 함).
    for qid in pending_pool:
        if len(selected_ids) >= target:
            break
        selected_ids.append(qid)
        selected_from_pending += 1

    # 3단계: hold fallback (cap 위반 허용 — 풀이 가능한 세션 우선).
    for qid in hold_pool:
        if len(selected_ids) >= target:
            break
        selected_ids.append(qid)
        selected_from_hold += 1

    # hold_pool 중 끝까지 안 들어간 것
    held_remaining = max(0, len(hold_pool) - selected_from_hold)

    logger.info(
        "type_spacing_applied subject=%s target=%d ready=%d selected=%d "
        "pending_fill=%d hold_fill=%d held=%d",
        subject_label or "(전체)",
        target,
        sum(1 for c in candidates if c.is_ready and c.embedding is not None),
        len(selected_ids),
        selected_from_pending,
        selected_from_hold,
        held_remaining,
    )

    return SpacingResult(
        selected_ids=selected_ids,
        selected_from_ready=selected_from_ready,
        selected_from_pending=selected_from_pending,
        selected_from_hold=selected_from_hold,
        held=held_remaining,
    )


def make_spacing_candidates_from_rows(rows: list) -> list[_SpacingCandidate]:
    """SQL row (id, embedding, embedding_status) → _SpacingCandidate 변환 helper.

    호출자는 이미 random shuffle / subject 필터를 마친 row 만 넘긴다.
    """
    out: list[_SpacingCandidate] = []
    for r in rows:
        is_ready = (
            getattr(r, "embedding_status", None) == "ready"
            and getattr(r, "embedding", None) is not None
        )
        emb = list(r.embedding) if is_ready else None
        out.append(_SpacingCandidate(qid=r.id, embedding=emb, is_ready=is_ready))
    return out
