hyungi_document_server/app/services/study/session_finalize.py

"""세션 종료 시 progress upsert + pattern 갱신 + 복습 stage 갱신 (Phase 1).

호출 시점: 마지막 attempt 가 박혀 status='done' 으로 전이된 직후.
한 트랜잭션 안에서 처리 — caller 가 commit 책임.

책임:
  1. 이 세션의 attempts qid 별 그룹
  2. 각 qid 별 progress upsert:
     - last_outcome / last_attempted_at / last_attempt_id 갱신
     - 최근 3회 attempts (이번 세션 + 과거) 로 pattern_state 계산
     - pattern_window_attempts 갱신
  3. 만약 progress.due_at IS NOT NULL (이미 복습 큐에 있던 문제):
     - correct → review_stage += 1, due_at 다음 단계 (1→3→7→14일)
     - stage ≥ 4 → due_at = NULL (학습완료)
     - wrong/unsure → review_stage = 0, due_at = now()+1일
  4. progress.due_at IS NULL 인 일반 풀이 결과는 due_at/stage 건드리지 않음
  5. 세션 결과 요약 dict 반환

attempts 원본 수정 0건. progress upsert 만.
"""

from __future__ import annotations

from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from models.study_question import StudyQuestion, StudyQuestionAttempt
from models.study_question_progress import StudyQuestionProgress
from models.study_quiz_session import StudyQuizSession
from services.study.learning_pattern import (
    PATTERN_CHRONIC_WRONG,
    PATTERN_RECOVERED,
    PATTERN_REGRESSED,
    PATTERN_STABLE,
    attempts_from_rows,
    compute_pattern_state,
)

# SR 산술은 sr_schedule.py 단일 source (문제 SR + 카드 SR 공용). 상수는 재-export 유지.
from services.study.sr_schedule import (  # noqa: E402
    DEFAULT_FIRST_DUE_DAYS,  # noqa: F401
    REVIEW_INTERVAL_DAYS,  # noqa: F401
    REVIEW_STAGE_MASTERED,  # noqa: F401
    advance as sr_advance,
)


@dataclass
class SessionSummary:
    """세션 결과 요약 — 응답 payload 직전 단계."""
    correct: int
    wrong: int
    unsure: int
    skipped: int
    # 상태 변화 카운트
    newly_correct: int       # 처음 푸는데 맞힘 (이전 attempts 0건)
    relapsed: int            # 이전 정답이었으나 이번 wrong
    recovered: int           # pattern_state = recovered 새로 박힘
    chronic_remaining: int   # 이번 세션 후 chronic_wrong 으로 박힌 행 수
    # 바로 할 일 후보
    pending_review_count: int  # wrong/unsure + 미확인 (last_reviewed_at < last_attempted_at)
    chronic_count: int         # pattern_state = chronic_wrong
    regressed_count: int       # pattern_state = regressed


async def finalize_session(
    session: AsyncSession,
    *,
    user_id: int,
    study_topic_id: int,
    quiz_session_id: int,
) -> SessionSummary:
    """세션 종료 후처리. caller 가 commit. 멱등성 — 두 번 호출되어도 결과 같음."""
    # 1. 이 세션 attempts 시간순
    session_attempts = (
        await session.execute(
            select(StudyQuestionAttempt)
            .where(
                StudyQuestionAttempt.quiz_session_id == quiz_session_id,
                StudyQuestionAttempt.user_id == user_id,
            )
            .order_by(StudyQuestionAttempt.answered_at.asc(), StudyQuestionAttempt.id.asc())
        )
    ).scalars().all()
    if not session_attempts:
        return SessionSummary(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)

    # qid 그룹: 한 세션 안에 같은 qid 가 두 번 박힐 가능성 거의 없지만 안전하게.
    last_per_qid: dict[int, StudyQuestionAttempt] = {}
    for a in session_attempts:
        last_per_qid[a.study_question_id] = a  # 시간 순 정렬이라 마지막이 최종

    qids = list(last_per_qid.keys())

    # 2. 기존 progress 행 일괄 fetch
    existing = (
        await session.execute(
            select(StudyQuestionProgress)
            .where(
                StudyQuestionProgress.user_id == user_id,
                StudyQuestionProgress.study_topic_id == study_topic_id,
                StudyQuestionProgress.study_question_id.in_(qids),
            )
        )
    ).scalars().all()
    progress_by_qid: dict[int, StudyQuestionProgress] = {p.study_question_id: p for p in existing}

    # 3. qid 별 과거 attempts 일괄 fetch (이 세션 attempts 까지 포함, 시간순)
    history_rows = (
        await session.execute(
            select(StudyQuestionAttempt)
            .where(
                StudyQuestionAttempt.user_id == user_id,
                StudyQuestionAttempt.study_topic_id == study_topic_id,
                StudyQuestionAttempt.study_question_id.in_(qids),
            )
            .order_by(
                StudyQuestionAttempt.study_question_id.asc(),
                StudyQuestionAttempt.answered_at.asc(),
                StudyQuestionAttempt.id.asc(),
            )
        )
    ).scalars().all()
    history_by_qid: dict[int, list[StudyQuestionAttempt]] = defaultdict(list)
    for r in history_rows:
        history_by_qid[r.study_question_id].append(r)

    now = datetime.now(timezone.utc)

    # 카운터 초기화
    correct = wrong = unsure = skipped = 0
    newly_correct = relapsed = recovered_count = 0
    chronic_remaining = 0

    for qid, last_attempt in last_per_qid.items():
        outcome = last_attempt.outcome
        if outcome == "correct":
            correct += 1
        elif outcome == "wrong":
            wrong += 1
        elif outcome == "unsure":
            unsure += 1
        elif outcome == "skipped":
            skipped += 1

        history = history_by_qid.get(qid, [])
        # 이번 세션 이전 attempts 만 (status 변화 판단용)
        prior = [a for a in history if a.id != last_attempt.id]

        # 상태 변화 카운트
        if not prior:
            if outcome == "correct":
                newly_correct += 1
        else:
            prior_last_outcome = prior[-1].outcome
            if prior_last_outcome == "correct" and outcome == "wrong":
                relapsed += 1

        # pattern 계산
        pattern_state, window_size = compute_pattern_state(attempts_from_rows(history))
        if pattern_state == PATTERN_RECOVERED and (
            qid not in progress_by_qid
            or progress_by_qid[qid].pattern_state != PATTERN_RECOVERED
        ):
            recovered_count += 1
        if pattern_state == PATTERN_CHRONIC_WRONG:
            chronic_remaining += 1

        # progress upsert
        progress = progress_by_qid.get(qid)
        if progress is None:
            progress = StudyQuestionProgress(
                user_id=user_id,
                study_topic_id=study_topic_id,
                study_question_id=qid,
            )
            session.add(progress)
            progress_by_qid[qid] = progress

        progress.last_outcome = outcome
        progress.last_attempted_at = last_attempt.answered_at
        progress.last_attempt_id = last_attempt.id
        progress.pattern_state = pattern_state
        progress.pattern_updated_at = now
        progress.pattern_window_attempts = window_size

        # 복습 stage 갱신 — 이미 due_at 박힌 문제만 (산술은 sr_schedule 공용)
        if progress.due_at is not None:
            result = sr_advance(progress.review_stage, outcome, now)
            if result is not None:  # skipped 는 None → due_at/stage 불변
                progress.review_stage, progress.due_at = result
        # progress.due_at IS NULL 일반 풀이 → stage 건드리지 않음

    # 4. 바로 할 일 카운트 (요약 응답용) — finalize 직후 progress 상태 기준 SQL 한 번
    pending_review_count = await _count_pending_review(session, user_id, study_topic_id)
    chronic_count = await _count_pattern(session, user_id, study_topic_id, PATTERN_CHRONIC_WRONG)
    regressed_count = await _count_pattern(session, user_id, study_topic_id, PATTERN_REGRESSED)

    # 5. quiz_session row 에 스냅샷 영속화 (Phase 2-B). 결과 화면 헤더 노출 + 멱등.
    qs = await session.get(StudyQuizSession, quiz_session_id)
    if qs is not None:
        qs.newly_correct_count = newly_correct
        qs.relapsed_count = relapsed
        qs.recovered_count = recovered_count
        qs.chronic_remaining_count = chronic_remaining

    # 6. Phase 4-A: 이 세션의 wrong/unsure qid AI 풀이 prefetch enqueue (best-effort).
    # 실패가 finalize 자체를 깨뜨리지 않도록 try/except 로 격리. 응답 카운트와 무관.
    try:
        from services.study.explanation_enqueue import enqueue_explanation_for_qids
        wrong_unsure_qids = [
            qid for qid, a in last_per_qid.items()
            if a.outcome in ("wrong", "unsure")
        ]
        if wrong_unsure_qids:
            res = await enqueue_explanation_for_qids(
                session, user_id=user_id, qids=wrong_unsure_qids,
            )
            import logging
            logging.getLogger(__name__).info(
                "phase4a_finalize_enqueue session=%s candidates=%s enqueued=%s skipped=%s",
                quiz_session_id, res["candidate_count"], res["enqueue_count"], res["skipped_count"],
            )
    except Exception as e:
        import logging
        logging.getLogger(__name__).warning(
            "phase4a_finalize_enqueue_failed session=%s: %s: %s",
            quiz_session_id, type(e).__name__, e,
        )

    # 7. Phase 4-B v1: 세션 단위 분석 enqueue (best-effort).
    # wrong/unsure < 5 면 worker 가 insufficient_attempts 로 skipped — finalize 는 무관.
    try:
        from services.study.session_analysis_enqueue import enqueue_session_analysis_auto
        res = await enqueue_session_analysis_auto(
            session, user_id=user_id, study_quiz_session_id=quiz_session_id,
        )
        import logging
        logging.getLogger(__name__).info(
            "phase4b_finalize_enqueue session=%s enqueued=%s",
            quiz_session_id, res["enqueued"],
        )
    except Exception as e:
        import logging
        logging.getLogger(__name__).warning(
            "phase4b_finalize_enqueue_failed session=%s: %s: %s",
            quiz_session_id, type(e).__name__, e,
        )

    return SessionSummary(
        correct=correct,
        wrong=wrong,
        unsure=unsure,
        skipped=skipped,
        newly_correct=newly_correct,
        relapsed=relapsed,
        recovered=recovered_count,
        chronic_remaining=chronic_remaining,
        pending_review_count=pending_review_count,
        chronic_count=chronic_count,
        regressed_count=regressed_count,
    )


async def _count_pending_review(session: AsyncSession, user_id: int, topic_id: int) -> int:
    """pending_review 탭 조건 기반 카운트."""
    from sqlalchemy import and_, func, or_
    row = await session.execute(
        select(func.count())
        .select_from(StudyQuestionProgress)
        .where(
            StudyQuestionProgress.user_id == user_id,
            StudyQuestionProgress.study_topic_id == topic_id,
            StudyQuestionProgress.last_outcome.in_(("wrong", "unsure")),
            or_(
                StudyQuestionProgress.last_reviewed_at.is_(None),
                and_(
                    StudyQuestionProgress.last_reviewed_at.is_not(None),
                    StudyQuestionProgress.last_attempted_at.is_not(None),
                    StudyQuestionProgress.last_reviewed_at
                    < StudyQuestionProgress.last_attempted_at,
                ),
            ),
        )
    )
    return int(row.scalar() or 0)


async def _count_pattern(
    session: AsyncSession, user_id: int, topic_id: int, pattern: str
) -> int:
    from sqlalchemy import func
    row = await session.execute(
        select(func.count())
        .select_from(StudyQuestionProgress)
        .where(
            StudyQuestionProgress.user_id == user_id,
            StudyQuestionProgress.study_topic_id == topic_id,
            StudyQuestionProgress.pattern_state == pattern,
        )
    )
    return int(row.scalar() or 0)