Hyungi Ahn
d3bf963a66
Phase 1 finalize 가 계산하던 SessionSummary 가 응답에 포함되지 않고 discard 되던 것을 quiz_session row 4 컬럼으로 영속화. 결과 화면 헤더에 회복/퇴행/ 새로 맞힘/반복 오답 누적 변화 카운트 + "바로 할 일" 콜아웃 (지금 시점 progress 기반 동적 카운트 — pending_review/chronic/regressed). 동적 카운트는 결과 GET 호출 시점에만 계산 (목록 endpoint 비용 회피). 확인완료 통합 — 결과 카드의 [학습완료] 버튼이 attempts.reviewed_at 만 박던 것을 progress.last_reviewed_at + (wrong/unsure 면 due_at 최초 부여) 도 같이 박도록. reviewed=false 토글은 attempts 만 되돌림 (다른 attempt 가 검토 표시 했을 수 있어 progress 의 last_reviewed_at 은 보존). - migrations/230 — quiz_sessions 4 컬럼 ADD (단일 ALTER TABLE) - StudyQuizSession 모델 + finalize_session 가 row 영속화 - QuizSessionSummary 응답에 4 스냅샷 + 3 동적 필드 (default 0) - _build_session_summary include_progress_counts=True 시 SQL 3회 - review-mark 가 reveiwed=true 시 progress 동기화 - 결과 화면: 헤더 변화 카운트 줄 + 바로 할 일 콜아웃 (값 있을 때만) Plan: ~/.claude/plans/crispy-petting-dijkstra.md (Phase 2-B) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1612 lines
59 KiB
Python
1612 lines
59 KiB
Python
"""학습 워크스페이스 문제은행 + 복습모드 API.
|
|
|
|
PR-2 가드레일:
|
|
- study_topic 컨테이너에 자산 타입별 1:N (문제) 추가. polymorphic 단일 테이블 영구 금지.
|
|
- subject/scope 강한 enum 미사용 (자유 텍스트, 자동완성은 후속 PR).
|
|
- 문제 삭제는 soft delete only. attempts 는 RESTRICT FK 로 DB 레벨 보호.
|
|
- correct_choice 변경 시 기존 attempts 재계산 안 함 (기록 = 시점 사실).
|
|
- 복습 출제 시 정답·해설 응답 비공개 → 답 제출 시점에만 노출.
|
|
- wrong_only=true 정의 = 가장 최근 attempt 가 오답인 문제. "한 번이라도 틀린 적 있는" 아님.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import random
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import and_, case, func, select, text as sql_text, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ai.client import AIClient
|
|
from core.auth import get_current_user
|
|
from core.config import settings
|
|
from core.database import get_session
|
|
from models.study_question import StudyQuestion, StudyQuestionAttempt
|
|
from models.study_question_image import StudyQuestionImage
|
|
from models.study_quiz_session import StudyQuizSession
|
|
from models.study_topic import StudyTopic
|
|
from models.user import User
|
|
from services.search.llm_gate import get_mlx_gate
|
|
from services.study.explanation_rag import (
|
|
EvidenceItem,
|
|
gather_explanation_context,
|
|
render_evidence_block,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
|
|
# ─── Helpers ───
|
|
|
|
|
|
def _verify_topic_ownership(topic: StudyTopic | None, user: User) -> StudyTopic:
|
|
"""study_topics.py 와 동일한 패턴 — 정보 누설 방지로 mismatch 도 404."""
|
|
if topic is None or topic.user_id != user.id or topic.deleted_at is not None:
|
|
raise HTTPException(status_code=404, detail="학습 주제를 찾을 수 없습니다")
|
|
return topic
|
|
|
|
|
|
def _verify_question_ownership(q: StudyQuestion | None, user: User) -> StudyQuestion:
|
|
if q is None or q.user_id != user.id or q.deleted_at is not None:
|
|
raise HTTPException(status_code=404, detail="문제를 찾을 수 없습니다")
|
|
return q
|
|
|
|
|
|
# ─── Pydantic 스키마 ───
|
|
|
|
|
|
class StudyQuestionCreate(BaseModel):
|
|
question_text: str = Field(min_length=1)
|
|
choice_1: str = Field(min_length=1)
|
|
choice_2: str = Field(min_length=1)
|
|
choice_3: str = Field(min_length=1)
|
|
choice_4: str = Field(min_length=1)
|
|
correct_choice: int = Field(ge=1, le=4)
|
|
subject: str | None = Field(default=None, max_length=120)
|
|
scope: str | None = Field(default=None, max_length=200)
|
|
exam_name: str | None = Field(default=None, max_length=120)
|
|
exam_round: str | None = Field(default=None, max_length=120)
|
|
# PR-6: 회차 안 문항 번호. 미명시 + exam_round 명시 시 서버가 max+1 자동 채움.
|
|
exam_question_number: int | None = Field(default=None, ge=1)
|
|
explanation: str | None = None
|
|
source_note: str | None = None
|
|
is_active: bool = True
|
|
|
|
|
|
class StudyQuestionUpdate(BaseModel):
|
|
question_text: str | None = Field(default=None, min_length=1)
|
|
choice_1: str | None = Field(default=None, min_length=1)
|
|
choice_2: str | None = Field(default=None, min_length=1)
|
|
choice_3: str | None = Field(default=None, min_length=1)
|
|
choice_4: str | None = Field(default=None, min_length=1)
|
|
correct_choice: int | None = Field(default=None, ge=1, le=4)
|
|
subject: str | None = Field(default=None, max_length=120)
|
|
scope: str | None = Field(default=None, max_length=200)
|
|
exam_name: str | None = Field(default=None, max_length=120)
|
|
exam_round: str | None = Field(default=None, max_length=120)
|
|
exam_question_number: int | None = Field(default=None, ge=1)
|
|
explanation: str | None = None
|
|
source_note: str | None = None
|
|
is_active: bool | None = None
|
|
|
|
|
|
class QuestionAttemptStats(BaseModel):
|
|
attempt_count: int
|
|
correct_count: int
|
|
wrong_count: int
|
|
|
|
|
|
class StudyQuestionImageItem(BaseModel):
|
|
"""문제 첨부 이미지 메타. raw bytes 는 별도 GET endpoint."""
|
|
id: int
|
|
sort_order: int
|
|
mime_type: str
|
|
file_size: int
|
|
served_url: str # 클라이언트 표시용 — /api/study-questions/{qid}/images/{img_id}/raw
|
|
|
|
|
|
class StudyQuestionResponse(BaseModel):
|
|
"""편집·관리용 — 정답·해설 모두 노출."""
|
|
|
|
id: int
|
|
study_topic_id: int
|
|
question_text: str
|
|
choice_1: str
|
|
choice_2: str
|
|
choice_3: str
|
|
choice_4: str
|
|
correct_choice: int
|
|
subject: str | None
|
|
scope: str | None
|
|
exam_name: str | None
|
|
exam_round: str | None
|
|
explanation: str | None
|
|
source_note: str | None
|
|
is_active: bool
|
|
# PR-6: 회차 안 문항 번호
|
|
exam_question_number: int | None = None
|
|
# PR-3: AI 풀이 상태 (편집 화면에서 사용). 본문은 별도 GET /ai-explanation 으로
|
|
ai_explanation_status: str = "none"
|
|
ai_explanation_generated_at: datetime | None = None
|
|
ai_explanation_model: str | None = None
|
|
# PR-8: 첨부 이미지
|
|
images: list[StudyQuestionImageItem] = []
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
stats: QuestionAttemptStats
|
|
|
|
|
|
class StudyQuestionSummary(BaseModel):
|
|
"""통합 뷰·목록용 — 본문 truncate, 정답 비공개."""
|
|
|
|
id: int
|
|
question_text: str # 앞 80자 truncate (서버 측에서 처리)
|
|
subject: str | None
|
|
scope: str | None
|
|
exam_name: str | None
|
|
exam_round: str | None
|
|
exam_question_number: int | None = None # PR-11/12 보기 페이지 prev/next 용
|
|
is_active: bool
|
|
has_images: bool = False
|
|
attempt_count: int
|
|
last_correct: bool | None
|
|
created_at: datetime
|
|
|
|
|
|
class StudyQuestionListResponse(BaseModel):
|
|
items: list[StudyQuestionSummary]
|
|
total: int
|
|
page: int
|
|
page_size: int
|
|
|
|
|
|
class ReviewChoice(BaseModel):
|
|
number: int
|
|
text: str
|
|
|
|
|
|
class ReviewQuestionItem(BaseModel):
|
|
"""복습 출제 응답 — 정답·해설 비공개."""
|
|
|
|
id: int
|
|
question_text: str
|
|
choices: list[ReviewChoice]
|
|
subject: str | None
|
|
scope: str | None
|
|
stats: QuestionAttemptStats
|
|
images: list[StudyQuestionImageItem] = []
|
|
|
|
|
|
class ReviewQuestionListResponse(BaseModel):
|
|
items: list[ReviewQuestionItem]
|
|
total: int
|
|
target_per_subject: int
|
|
subject_distribution: dict[str, int] # {"연소공학": 20, ...}
|
|
|
|
|
|
class AttemptCreate(BaseModel):
|
|
"""PR-9: selected_choice (1~4) 또는 is_unsure 둘 중 하나 필수.
|
|
is_unsure=true 면 selected_choice 무시 + outcome='unsure' 로 박힘.
|
|
PR-10: quiz_session_id 전달 시 세션의 cursor + count 도 함께 갱신.
|
|
"""
|
|
selected_choice: int | None = Field(default=None, ge=1, le=4)
|
|
is_unsure: bool = False
|
|
quiz_session_id: int | None = None
|
|
|
|
|
|
class AttemptResponse(BaseModel):
|
|
id: int # PR-10: 학습완료 토글 시 attempt 단위로 PATCH 하므로 id 노출 필요.
|
|
is_correct: bool
|
|
selected_choice: int | None
|
|
correct_choice: int
|
|
outcome: str # PR-9: correct | wrong | unsure
|
|
explanation: str | None
|
|
stats: QuestionAttemptStats
|
|
quiz_session_id: int | None = None
|
|
quiz_session_status: str | None = None # in_progress | done (마지막 문제면 done 으로 박힘)
|
|
quiz_session_cursor: int | None = None
|
|
reviewed_at: datetime | None = None
|
|
|
|
|
|
# ─── PR-5: 비슷한 문제 검색 (embedding cosine) ───
|
|
|
|
|
|
class SimilarQuestionItem(BaseModel):
|
|
id: int
|
|
study_topic_id: int
|
|
question_text: str # 80자 truncate
|
|
subject: str | None
|
|
scope: str | None
|
|
exam_name: str | None
|
|
exam_round: str | None
|
|
similarity: float # 1 - cosine_distance, 1.0=동일, 0.0=무관
|
|
attempt_count: int
|
|
last_correct: bool | None
|
|
|
|
|
|
class SimilarQuestionsResponse(BaseModel):
|
|
items: list[SimilarQuestionItem]
|
|
source_status: str # 현재 문제의 embedding_status. 'ready' 가 아니면 빈 결과.
|
|
source_id: int
|
|
|
|
|
|
# ─── PR-3: AI 풀이 생성 ───
|
|
|
|
|
|
class AIExplanationCreate(BaseModel):
|
|
"""수동 트리거. regenerate=true 면 캐시 무시하고 새 본문 생성."""
|
|
regenerate: bool = False
|
|
|
|
|
|
class AIExplanationEvidence(BaseModel):
|
|
source_type: str # 'document' | 'question'
|
|
source_id: int
|
|
title: str
|
|
snippet: str
|
|
|
|
|
|
class AIExplanationResponse(BaseModel):
|
|
ai_explanation: str | None
|
|
ai_explanation_status: str # ready | stale | failed | none
|
|
ai_explanation_generated_at: datetime | None
|
|
ai_explanation_model: str | None
|
|
evidence: list[AIExplanationEvidence] = []
|
|
from_cache: bool = False
|
|
is_stale: bool = False
|
|
can_regenerate: bool = True
|
|
|
|
|
|
# ─── 통계 헬퍼 ───
|
|
|
|
|
|
# PR-8: 이미지 저장 root + 헬퍼
|
|
from pathlib import Path as _Path
|
|
|
|
STUDY_IMG_ROOT = _Path(settings.nas_mount_path) / "study_question_images"
|
|
ALLOWED_IMAGE_MIME = {"image/png", "image/jpeg", "image/webp", "image/gif"}
|
|
MAX_IMAGE_BYTES = 10 * 1024 * 1024 # 10MB / 파일
|
|
|
|
|
|
def _image_url(qid: int, img_id: int) -> str:
|
|
return f"/api/study-questions/{qid}/images/{img_id}/raw"
|
|
|
|
|
|
def _image_to_item(img: StudyQuestionImage) -> StudyQuestionImageItem:
|
|
return StudyQuestionImageItem(
|
|
id=img.id,
|
|
sort_order=img.sort_order,
|
|
mime_type=img.mime_type,
|
|
file_size=img.file_size,
|
|
served_url=_image_url(img.study_question_id, img.id),
|
|
)
|
|
|
|
|
|
async def _images_for_question(
|
|
session: AsyncSession, question_id: int
|
|
) -> list[StudyQuestionImageItem]:
|
|
rows = (
|
|
await session.execute(
|
|
select(StudyQuestionImage)
|
|
.where(StudyQuestionImage.study_question_id == question_id)
|
|
.order_by(StudyQuestionImage.sort_order.asc(), StudyQuestionImage.id.asc())
|
|
)
|
|
).scalars().all()
|
|
return [_image_to_item(r) for r in rows]
|
|
|
|
|
|
async def _images_for_questions_batch(
|
|
session: AsyncSession, question_ids: list[int]
|
|
) -> dict[int, list[StudyQuestionImageItem]]:
|
|
"""N+1 회피 — 여러 question 의 이미지 한 번에 조회."""
|
|
if not question_ids:
|
|
return {}
|
|
rows = (
|
|
await session.execute(
|
|
select(StudyQuestionImage)
|
|
.where(StudyQuestionImage.study_question_id.in_(question_ids))
|
|
.order_by(
|
|
StudyQuestionImage.study_question_id,
|
|
StudyQuestionImage.sort_order.asc(),
|
|
StudyQuestionImage.id.asc(),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
out: dict[int, list[StudyQuestionImageItem]] = {qid: [] for qid in question_ids}
|
|
for r in rows:
|
|
out.setdefault(r.study_question_id, []).append(_image_to_item(r))
|
|
return out
|
|
|
|
|
|
async def _attempt_stats(
|
|
session: AsyncSession, user_id: int, question_id: int
|
|
) -> QuestionAttemptStats:
|
|
"""단일 문제의 누적 attempt 통계 (사용자 한정)."""
|
|
row = (
|
|
await session.execute(
|
|
select(
|
|
func.count().label("total"),
|
|
func.coalesce(
|
|
func.sum(case((StudyQuestionAttempt.is_correct.is_(True), 1), else_=0)),
|
|
0,
|
|
).label("correct"),
|
|
).where(
|
|
StudyQuestionAttempt.user_id == user_id,
|
|
StudyQuestionAttempt.study_question_id == question_id,
|
|
)
|
|
)
|
|
).one()
|
|
total = int(row.total or 0)
|
|
correct = int(row.correct or 0)
|
|
return QuestionAttemptStats(
|
|
attempt_count=total, correct_count=correct, wrong_count=total - correct
|
|
)
|
|
|
|
|
|
def _truncate(text: str, n: int = 80) -> str:
|
|
return text if len(text) <= n else text[:n].rstrip() + "…"
|
|
|
|
|
|
# ─── 토픽 단위 엔드포인트 ───
|
|
|
|
|
|
@router.get(
|
|
"/study-topics/{topic_id}/questions",
|
|
response_model=StudyQuestionListResponse,
|
|
)
|
|
async def list_questions_in_topic(
|
|
topic_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
subject: str | None = Query(None),
|
|
scope: str | None = Query(None),
|
|
exam_round: str | None = Query(None, description="회차 필터 — 보기 페이지 prev/next 같은 단일 회차 한정 조회"),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
):
|
|
"""토픽 내 문제 목록. soft-deleted 제외. summary 응답 (정답 비공개)."""
|
|
topic = await session.get(StudyTopic, topic_id)
|
|
_verify_topic_ownership(topic, user)
|
|
|
|
base = select(StudyQuestion).where(
|
|
StudyQuestion.user_id == user.id,
|
|
StudyQuestion.study_topic_id == topic_id,
|
|
StudyQuestion.deleted_at.is_(None),
|
|
)
|
|
if subject is not None:
|
|
base = base.where(StudyQuestion.subject == subject)
|
|
if scope is not None:
|
|
base = base.where(StudyQuestion.scope == scope)
|
|
if exam_round is not None:
|
|
base = base.where(StudyQuestion.exam_round == exam_round)
|
|
|
|
total = (
|
|
await session.execute(select(func.count()).select_from(base.subquery()))
|
|
).scalar() or 0
|
|
|
|
# 정렬: exam_round 필터 시 회차 안 prev/next 안정화 (exam_question_number asc NULLS LAST,
|
|
# created_at asc fallback). 미필터 시 기존 created_at desc 유지.
|
|
if exam_round is not None:
|
|
order_clause = (
|
|
StudyQuestion.exam_question_number.asc().nulls_last(),
|
|
StudyQuestion.created_at.asc(),
|
|
StudyQuestion.id.asc(),
|
|
)
|
|
else:
|
|
order_clause = (StudyQuestion.created_at.desc(), StudyQuestion.id.desc())
|
|
rows = (
|
|
await session.execute(
|
|
base.order_by(*order_clause)
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
)
|
|
).scalars().all()
|
|
|
|
# 한 번의 쿼리로 attempt 통계 + 마지막 정/오답 끌어오기 (N+1 회피)
|
|
qids = [q.id for q in rows]
|
|
stats_map: dict[int, tuple[int, int]] = {}
|
|
last_correct_map: dict[int, bool] = {}
|
|
if qids:
|
|
stats_rows = (
|
|
await session.execute(
|
|
select(
|
|
StudyQuestionAttempt.study_question_id,
|
|
func.count().label("total"),
|
|
func.coalesce(
|
|
func.sum(case((StudyQuestionAttempt.is_correct.is_(True), 1), else_=0)),
|
|
0,
|
|
).label("correct"),
|
|
)
|
|
.where(
|
|
StudyQuestionAttempt.user_id == user.id,
|
|
StudyQuestionAttempt.study_question_id.in_(qids),
|
|
)
|
|
.group_by(StudyQuestionAttempt.study_question_id)
|
|
)
|
|
).all()
|
|
for r in stats_rows:
|
|
stats_map[r.study_question_id] = (int(r.total), int(r.correct))
|
|
|
|
latest_rows = (
|
|
await session.execute(
|
|
select(
|
|
StudyQuestionAttempt.study_question_id,
|
|
StudyQuestionAttempt.is_correct,
|
|
StudyQuestionAttempt.answered_at,
|
|
)
|
|
.where(
|
|
StudyQuestionAttempt.user_id == user.id,
|
|
StudyQuestionAttempt.study_question_id.in_(qids),
|
|
)
|
|
.order_by(
|
|
StudyQuestionAttempt.study_question_id,
|
|
StudyQuestionAttempt.answered_at.desc(),
|
|
)
|
|
.distinct(StudyQuestionAttempt.study_question_id)
|
|
)
|
|
).all()
|
|
for r in latest_rows:
|
|
last_correct_map[r.study_question_id] = bool(r.is_correct)
|
|
|
|
# PR-8: has_images batch
|
|
has_image_set: set[int] = set()
|
|
if qids:
|
|
img_rows = (
|
|
await session.execute(
|
|
select(StudyQuestionImage.study_question_id)
|
|
.where(StudyQuestionImage.study_question_id.in_(qids))
|
|
.distinct()
|
|
)
|
|
).scalars().all()
|
|
has_image_set = set(img_rows)
|
|
|
|
items = [
|
|
StudyQuestionSummary(
|
|
id=q.id,
|
|
question_text=_truncate(q.question_text, 80),
|
|
subject=q.subject,
|
|
scope=q.scope,
|
|
exam_name=q.exam_name,
|
|
exam_round=q.exam_round,
|
|
exam_question_number=q.exam_question_number,
|
|
is_active=q.is_active,
|
|
attempt_count=stats_map.get(q.id, (0, 0))[0],
|
|
last_correct=last_correct_map.get(q.id),
|
|
has_images=q.id in has_image_set,
|
|
created_at=q.created_at,
|
|
)
|
|
for q in rows
|
|
]
|
|
return StudyQuestionListResponse(items=items, total=total, page=page, page_size=page_size)
|
|
|
|
|
|
@router.post(
|
|
"/study-topics/{topic_id}/questions",
|
|
response_model=StudyQuestionResponse,
|
|
status_code=201,
|
|
)
|
|
async def create_question_in_topic(
|
|
topic_id: int,
|
|
body: StudyQuestionCreate,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
topic = await session.get(StudyTopic, topic_id)
|
|
_verify_topic_ownership(topic, user)
|
|
|
|
# PR-6: exam_question_number 미명시 + exam_round 명시 시 서버가 max+1 자동 채움.
|
|
# 개인 사용 환경 race 부담 적어 단순 SELECT max 사용. 동시 입력 빈번해지면 향후
|
|
# SELECT FOR UPDATE 또는 토픽 단위 advisory lock 으로 강화 검토.
|
|
qnum = body.exam_question_number
|
|
if qnum is None and body.exam_round:
|
|
max_row = await session.execute(
|
|
select(func.coalesce(func.max(StudyQuestion.exam_question_number), 0))
|
|
.where(
|
|
StudyQuestion.user_id == user.id,
|
|
StudyQuestion.study_topic_id == topic_id,
|
|
StudyQuestion.exam_round == body.exam_round,
|
|
StudyQuestion.deleted_at.is_(None),
|
|
)
|
|
)
|
|
qnum = int(max_row.scalar() or 0) + 1
|
|
|
|
q = StudyQuestion(
|
|
user_id=user.id,
|
|
study_topic_id=topic_id,
|
|
question_text=body.question_text,
|
|
choice_1=body.choice_1,
|
|
choice_2=body.choice_2,
|
|
choice_3=body.choice_3,
|
|
choice_4=body.choice_4,
|
|
correct_choice=body.correct_choice,
|
|
subject=body.subject,
|
|
scope=body.scope,
|
|
exam_name=body.exam_name,
|
|
exam_round=body.exam_round,
|
|
exam_question_number=qnum,
|
|
explanation=body.explanation,
|
|
source_note=body.source_note,
|
|
is_active=body.is_active,
|
|
)
|
|
session.add(q)
|
|
await session.flush()
|
|
await session.commit()
|
|
|
|
stats = QuestionAttemptStats(attempt_count=0, correct_count=0, wrong_count=0)
|
|
return StudyQuestionResponse(
|
|
id=q.id,
|
|
study_topic_id=q.study_topic_id,
|
|
question_text=q.question_text,
|
|
choice_1=q.choice_1,
|
|
choice_2=q.choice_2,
|
|
choice_3=q.choice_3,
|
|
choice_4=q.choice_4,
|
|
correct_choice=q.correct_choice,
|
|
subject=q.subject,
|
|
scope=q.scope,
|
|
exam_name=q.exam_name,
|
|
exam_round=q.exam_round,
|
|
explanation=q.explanation,
|
|
source_note=q.source_note,
|
|
is_active=q.is_active,
|
|
exam_question_number=q.exam_question_number,
|
|
ai_explanation_status=q.ai_explanation_status,
|
|
ai_explanation_generated_at=q.ai_explanation_generated_at,
|
|
ai_explanation_model=q.ai_explanation_model,
|
|
images=await _images_for_question(session, q.id),
|
|
created_at=q.created_at,
|
|
updated_at=q.updated_at,
|
|
stats=stats,
|
|
)
|
|
|
|
|
|
# ─── 복습 출제 ───
|
|
|
|
|
|
@router.get(
|
|
"/study-topics/{topic_id}/review/questions",
|
|
response_model=ReviewQuestionListResponse,
|
|
)
|
|
async def review_questions_for_topic(
|
|
topic_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
subject: str | None = Query(None, description="단일 과목 집중. 미지정 시 과목별 균등 추출"),
|
|
scope: str | None = Query(None),
|
|
target_per_subject: int = Query(20, ge=1, le=100),
|
|
wrong_only: bool = Query(
|
|
False,
|
|
description="가장 최근 attempt 가 오답인 문제만 (latest-wrong, ever_wrong 아님)",
|
|
),
|
|
):
|
|
"""복습 출제. default = 과목별 target_per_subject 무작위 균등 추출.
|
|
|
|
응답에서 정답·해설은 비공개 — 답 제출(POST /api/study-questions/{id}/attempt) 시점에만 노출.
|
|
"""
|
|
topic = await session.get(StudyTopic, topic_id)
|
|
_verify_topic_ownership(topic, user)
|
|
|
|
# 후보 질문 base — soft delete 제외 + is_active
|
|
base_filter = and_(
|
|
StudyQuestion.user_id == user.id,
|
|
StudyQuestion.study_topic_id == topic_id,
|
|
StudyQuestion.deleted_at.is_(None),
|
|
StudyQuestion.is_active.is_(True),
|
|
)
|
|
|
|
# wrong_only=true: latest attempt 가 오답인 question_id 만 후보로 좁힘.
|
|
# DISTINCT ON (study_question_id) ORDER BY answered_at DESC 패턴.
|
|
wrong_qids: set[int] | None = None
|
|
if wrong_only:
|
|
latest = (
|
|
await session.execute(
|
|
select(
|
|
StudyQuestionAttempt.study_question_id,
|
|
StudyQuestionAttempt.is_correct,
|
|
)
|
|
.where(
|
|
StudyQuestionAttempt.user_id == user.id,
|
|
StudyQuestionAttempt.study_topic_id == topic_id,
|
|
)
|
|
.order_by(
|
|
StudyQuestionAttempt.study_question_id,
|
|
StudyQuestionAttempt.answered_at.desc(),
|
|
)
|
|
.distinct(StudyQuestionAttempt.study_question_id)
|
|
)
|
|
).all()
|
|
wrong_qids = {r.study_question_id for r in latest if r.is_correct is False}
|
|
if not wrong_qids:
|
|
return ReviewQuestionListResponse(
|
|
items=[], total=0, target_per_subject=target_per_subject, subject_distribution={}
|
|
)
|
|
|
|
# 과목 그룹 결정
|
|
if subject is not None:
|
|
# 단일 과목 집중
|
|
subjects: list[str | None] = [subject]
|
|
else:
|
|
subj_query = (
|
|
select(StudyQuestion.subject)
|
|
.where(base_filter)
|
|
.distinct()
|
|
)
|
|
if scope is not None:
|
|
subj_query = subj_query.where(StudyQuestion.scope == scope)
|
|
if wrong_qids is not None:
|
|
subj_query = subj_query.where(StudyQuestion.id.in_(wrong_qids))
|
|
subjects = [r[0] for r in (await session.execute(subj_query)).all()]
|
|
if not subjects:
|
|
return ReviewQuestionListResponse(
|
|
items=[], total=0, target_per_subject=target_per_subject, subject_distribution={}
|
|
)
|
|
|
|
# 각 subject 별로 ORDER BY random() LIMIT target_per_subject
|
|
selected: list[StudyQuestion] = []
|
|
distribution: dict[str, int] = {}
|
|
for subj in subjects:
|
|
sub_q = select(StudyQuestion).where(base_filter)
|
|
if subj is None:
|
|
sub_q = sub_q.where(StudyQuestion.subject.is_(None))
|
|
else:
|
|
sub_q = sub_q.where(StudyQuestion.subject == subj)
|
|
if scope is not None:
|
|
sub_q = sub_q.where(StudyQuestion.scope == scope)
|
|
if wrong_qids is not None:
|
|
sub_q = sub_q.where(StudyQuestion.id.in_(wrong_qids))
|
|
sub_q = sub_q.order_by(func.random()).limit(target_per_subject)
|
|
rows = (await session.execute(sub_q)).scalars().all()
|
|
if rows:
|
|
selected.extend(rows)
|
|
distribution[subj or ""] = len(rows)
|
|
|
|
# 그룹별 추출 후 전체 셔플 (입력 순서 단조 회피)
|
|
random.shuffle(selected)
|
|
|
|
# attempt 통계 batch (N+1 회피)
|
|
qids = [q.id for q in selected]
|
|
stats_map: dict[int, QuestionAttemptStats] = {}
|
|
if qids:
|
|
stats_rows = (
|
|
await session.execute(
|
|
select(
|
|
StudyQuestionAttempt.study_question_id,
|
|
func.count().label("total"),
|
|
func.coalesce(
|
|
func.sum(case((StudyQuestionAttempt.is_correct.is_(True), 1), else_=0)),
|
|
0,
|
|
).label("correct"),
|
|
)
|
|
.where(
|
|
StudyQuestionAttempt.user_id == user.id,
|
|
StudyQuestionAttempt.study_question_id.in_(qids),
|
|
)
|
|
.group_by(StudyQuestionAttempt.study_question_id)
|
|
)
|
|
).all()
|
|
for r in stats_rows:
|
|
t = int(r.total)
|
|
c = int(r.correct)
|
|
stats_map[r.study_question_id] = QuestionAttemptStats(
|
|
attempt_count=t, correct_count=c, wrong_count=t - c
|
|
)
|
|
|
|
# PR-8: 이미지 batch
|
|
images_map = await _images_for_questions_batch(session, qids)
|
|
|
|
items = [
|
|
ReviewQuestionItem(
|
|
id=q.id,
|
|
question_text=q.question_text,
|
|
choices=[
|
|
ReviewChoice(number=1, text=q.choice_1),
|
|
ReviewChoice(number=2, text=q.choice_2),
|
|
ReviewChoice(number=3, text=q.choice_3),
|
|
ReviewChoice(number=4, text=q.choice_4),
|
|
],
|
|
subject=q.subject,
|
|
scope=q.scope,
|
|
stats=stats_map.get(
|
|
q.id, QuestionAttemptStats(attempt_count=0, correct_count=0, wrong_count=0)
|
|
),
|
|
images=images_map.get(q.id, []),
|
|
)
|
|
for q in selected
|
|
]
|
|
return ReviewQuestionListResponse(
|
|
items=items,
|
|
total=len(items),
|
|
target_per_subject=target_per_subject,
|
|
subject_distribution=distribution,
|
|
)
|
|
|
|
|
|
# ─── 단건 엔드포인트 ───
|
|
|
|
|
|
@router.get("/study-questions/{question_id}", response_model=StudyQuestionResponse)
|
|
async def get_question(
|
|
question_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
q = await session.get(StudyQuestion, question_id)
|
|
q = _verify_question_ownership(q, user)
|
|
stats = await _attempt_stats(session, user.id, question_id)
|
|
return StudyQuestionResponse(
|
|
id=q.id,
|
|
study_topic_id=q.study_topic_id,
|
|
question_text=q.question_text,
|
|
choice_1=q.choice_1,
|
|
choice_2=q.choice_2,
|
|
choice_3=q.choice_3,
|
|
choice_4=q.choice_4,
|
|
correct_choice=q.correct_choice,
|
|
subject=q.subject,
|
|
scope=q.scope,
|
|
exam_name=q.exam_name,
|
|
exam_round=q.exam_round,
|
|
explanation=q.explanation,
|
|
source_note=q.source_note,
|
|
is_active=q.is_active,
|
|
exam_question_number=q.exam_question_number,
|
|
ai_explanation_status=q.ai_explanation_status,
|
|
ai_explanation_generated_at=q.ai_explanation_generated_at,
|
|
ai_explanation_model=q.ai_explanation_model,
|
|
images=await _images_for_question(session, q.id),
|
|
created_at=q.created_at,
|
|
updated_at=q.updated_at,
|
|
stats=stats,
|
|
)
|
|
|
|
|
|
@router.patch("/study-questions/{question_id}", response_model=StudyQuestionResponse)
|
|
async def update_question(
|
|
question_id: int,
|
|
body: StudyQuestionUpdate,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""부분 업데이트. correct_choice 변경 시 기존 attempts.is_correct 재계산 안 함 (시점 사실 보존).
|
|
프론트 편집 페이지에 안내 배너 노출."""
|
|
q = await session.get(StudyQuestion, question_id)
|
|
q = _verify_question_ownership(q, user)
|
|
|
|
fields_set = body.model_fields_set
|
|
SIMPLE_FIELDS = {
|
|
"question_text", "choice_1", "choice_2", "choice_3", "choice_4",
|
|
"correct_choice", "subject", "scope", "exam_name", "exam_round",
|
|
"exam_question_number",
|
|
"explanation", "source_note", "is_active",
|
|
}
|
|
for fname in SIMPLE_FIELDS & fields_set:
|
|
setattr(q, fname, getattr(body, fname))
|
|
|
|
# PR-3: 문제 핵심 필드 변경 시 AI 해설 stale 전이 (본문은 보존, UI 배지로 안내).
|
|
# ready 상태에서만 stale 로 전이 — pending/failed/none/stale 은 변경 안 함.
|
|
AI_STALE_TRIGGER = {
|
|
"question_text", "choice_1", "choice_2", "choice_3", "choice_4", "correct_choice",
|
|
}
|
|
if AI_STALE_TRIGGER & fields_set and q.ai_explanation_status == "ready":
|
|
q.ai_explanation_status = "stale"
|
|
|
|
# PR-4: 임베딩 stale 전이. 본문(question_text/choice_*)이 바뀌었을 때만 재계산.
|
|
# correct_choice 변경은 의미 검색에 영향 없으므로 재계산 안 함.
|
|
EMBED_STALE_TRIGGER = {
|
|
"question_text", "choice_1", "choice_2", "choice_3", "choice_4",
|
|
}
|
|
if EMBED_STALE_TRIGGER & fields_set and q.embedding_status == "ready":
|
|
q.embedding_status = "stale"
|
|
|
|
# PR-12-A 후속: related 캐시 stale 전이.
|
|
# 임베딩 본문 변경 (위 EMBED_STALE_TRIGGER) + exam_round 변경 시.
|
|
# exam_round 는 related-types 의 회차 필터 조건이라 본인 round 바뀌면 candidate 재선정 필요.
|
|
RELATED_STALE_TRIGGER = EMBED_STALE_TRIGGER | {"exam_round"}
|
|
if RELATED_STALE_TRIGGER & fields_set and q.related_computed_at is not None:
|
|
q.related_computed_at = None
|
|
|
|
q.updated_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
|
|
stats = await _attempt_stats(session, user.id, question_id)
|
|
return StudyQuestionResponse(
|
|
id=q.id,
|
|
study_topic_id=q.study_topic_id,
|
|
question_text=q.question_text,
|
|
choice_1=q.choice_1,
|
|
choice_2=q.choice_2,
|
|
choice_3=q.choice_3,
|
|
choice_4=q.choice_4,
|
|
correct_choice=q.correct_choice,
|
|
subject=q.subject,
|
|
scope=q.scope,
|
|
exam_name=q.exam_name,
|
|
exam_round=q.exam_round,
|
|
explanation=q.explanation,
|
|
source_note=q.source_note,
|
|
is_active=q.is_active,
|
|
exam_question_number=q.exam_question_number,
|
|
ai_explanation_status=q.ai_explanation_status,
|
|
ai_explanation_generated_at=q.ai_explanation_generated_at,
|
|
ai_explanation_model=q.ai_explanation_model,
|
|
images=await _images_for_question(session, q.id),
|
|
created_at=q.created_at,
|
|
updated_at=q.updated_at,
|
|
stats=stats,
|
|
)
|
|
|
|
|
|
@router.delete("/study-questions/{question_id}", status_code=204)
|
|
async def soft_delete_question(
|
|
question_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""soft delete only. attempts 는 RESTRICT FK 로 보호되어 영구 보존.
|
|
hard delete 호출 경로 자체가 없음 — DB 레벨에서도 attempts 가 있으면 거부.
|
|
|
|
PR-12-A 후속: 같은 토픽의 다른 문제 related 캐시에 이 qid 가 candidate 로
|
|
남아있을 수 있음. 같은 토픽 ready 행들의 related_computed_at 을 NULL 마킹 → cron 재계산.
|
|
"""
|
|
q = await session.get(StudyQuestion, question_id)
|
|
q = _verify_question_ownership(q, user)
|
|
q.deleted_at = datetime.now(timezone.utc)
|
|
q.updated_at = q.deleted_at
|
|
await session.execute(
|
|
update(StudyQuestion)
|
|
.where(
|
|
StudyQuestion.study_topic_id == q.study_topic_id,
|
|
StudyQuestion.id != q.id,
|
|
StudyQuestion.embedding_status == "ready",
|
|
StudyQuestion.deleted_at.is_(None),
|
|
StudyQuestion.related_computed_at.is_not(None),
|
|
)
|
|
.values(related_computed_at=None)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
# ─── attempt ───
|
|
|
|
|
|
@router.post("/study-questions/{question_id}/attempt", response_model=AttemptResponse)
|
|
async def submit_attempt(
|
|
question_id: int,
|
|
body: AttemptCreate,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""답 제출. PR-9: is_unsure=true 면 outcome='unsure' + selected_choice=NULL.
|
|
그 외엔 selected_choice 와 correct_choice 비교 → outcome='correct'/'wrong'.
|
|
PR-10: quiz_session_id 전달 시 같은 트랜잭션에서 세션 cursor + count 도 갱신.
|
|
cursor 가 question_ids 끝까지 도달하면 status='done', finished_at 박힘.
|
|
"""
|
|
q = await session.get(StudyQuestion, question_id)
|
|
q = _verify_question_ownership(q, user)
|
|
|
|
if body.is_unsure:
|
|
selected = None
|
|
is_correct = False
|
|
outcome = "unsure"
|
|
elif body.selected_choice is None:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="selected_choice (1~4) 또는 is_unsure=true 가 필요합니다",
|
|
)
|
|
else:
|
|
selected = body.selected_choice
|
|
is_correct = selected == q.correct_choice
|
|
outcome = "correct" if is_correct else "wrong"
|
|
|
|
# PR-10: 세션 연동. 기본은 None.
|
|
quiz_session: StudyQuizSession | None = None
|
|
if body.quiz_session_id is not None:
|
|
quiz_session = await session.get(StudyQuizSession, body.quiz_session_id)
|
|
if quiz_session is None or quiz_session.user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="quiz_session 을 찾을 수 없습니다")
|
|
if quiz_session.study_topic_id != q.study_topic_id:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="quiz_session 의 토픽과 문제의 토픽이 다릅니다",
|
|
)
|
|
if quiz_session.status != "in_progress":
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"quiz_session 상태가 {quiz_session.status} — 이미 종료된 세션입니다",
|
|
)
|
|
# 현재 cursor 위치의 문제와 일치해야만 진행 (out-of-order 제출 차단).
|
|
qids = quiz_session.question_ids or []
|
|
if quiz_session.cursor >= len(qids):
|
|
raise HTTPException(status_code=409, detail="이미 모든 문제를 풀었습니다")
|
|
if qids[quiz_session.cursor] != q.id:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="현재 cursor 위치의 문제와 question_id 가 일치하지 않습니다 (이중 제출/순서 어긋남)",
|
|
)
|
|
|
|
attempt = StudyQuestionAttempt(
|
|
user_id=user.id,
|
|
study_question_id=q.id,
|
|
study_topic_id=q.study_topic_id,
|
|
selected_choice=selected,
|
|
correct_choice=q.correct_choice,
|
|
is_correct=is_correct,
|
|
outcome=outcome,
|
|
quiz_session_id=quiz_session.id if quiz_session else None,
|
|
)
|
|
session.add(attempt)
|
|
|
|
session_just_finished = False
|
|
if quiz_session is not None:
|
|
quiz_session.cursor = quiz_session.cursor + 1
|
|
if outcome == "correct":
|
|
quiz_session.correct_count += 1
|
|
elif outcome == "wrong":
|
|
quiz_session.wrong_count += 1
|
|
elif outcome == "unsure":
|
|
quiz_session.unsure_count += 1
|
|
if quiz_session.cursor >= len(quiz_session.question_ids or []):
|
|
quiz_session.status = "done"
|
|
quiz_session.finished_at = datetime.now(timezone.utc)
|
|
session_just_finished = True
|
|
quiz_session.updated_at = datetime.now(timezone.utc)
|
|
|
|
# 1차 commit: attempt + quiz_session.status (실패 시 attempt 조차 안 박힘 — 사용자 재시도)
|
|
await session.commit()
|
|
await session.refresh(attempt)
|
|
|
|
# Phase 1: 세션 종료 시 progress upsert + pattern 갱신 + (있는 경우) 복습 stage 갱신.
|
|
# 별도 트랜잭션 — finalize 실패가 attempt 보존을 위협하지 않게 분리. 멱등성 보장 (다음 review/queue
|
|
# 호출 시 stale 진단 가능). 첫 차 commit 의 attempts 가 이미 가시화된 상태에서 시작.
|
|
if session_just_finished and quiz_session is not None:
|
|
from services.study.session_finalize import finalize_session
|
|
try:
|
|
await finalize_session(
|
|
session,
|
|
user_id=user.id,
|
|
study_topic_id=q.study_topic_id,
|
|
quiz_session_id=quiz_session.id,
|
|
)
|
|
await session.commit()
|
|
except Exception:
|
|
import logging
|
|
logging.getLogger(__name__).exception(
|
|
"finalize_session_failed quiz_session_id=%s", quiz_session.id
|
|
)
|
|
await session.rollback()
|
|
|
|
stats = await _attempt_stats(session, user.id, question_id)
|
|
return AttemptResponse(
|
|
id=attempt.id,
|
|
is_correct=is_correct,
|
|
selected_choice=selected,
|
|
correct_choice=q.correct_choice,
|
|
outcome=outcome,
|
|
explanation=q.explanation,
|
|
stats=stats,
|
|
quiz_session_id=quiz_session.id if quiz_session else None,
|
|
quiz_session_status=quiz_session.status if quiz_session else None,
|
|
quiz_session_cursor=quiz_session.cursor if quiz_session else None,
|
|
reviewed_at=attempt.reviewed_at,
|
|
)
|
|
|
|
|
|
# ─── PR-10: 학습완료 토글 ───
|
|
|
|
|
|
class AttemptReviewRequest(BaseModel):
|
|
reviewed: bool
|
|
|
|
|
|
class AttemptReviewResponse(BaseModel):
|
|
id: int
|
|
reviewed_at: datetime | None
|
|
|
|
|
|
@router.patch(
|
|
"/study-question-attempts/{attempt_id}/review-mark",
|
|
response_model=AttemptReviewResponse,
|
|
)
|
|
async def mark_attempt_reviewed(
|
|
attempt_id: int,
|
|
body: AttemptReviewRequest,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""결과 카드에서 [학습완료] 토글. reviewed=true 면 timestamp, false 면 NULL.
|
|
|
|
Phase 2-B: progress 단위 review-complete 도 동시에 박는다 — last_reviewed_at +
|
|
(wrong/unsure 면) due_at 최초 부여. attempt.reviewed_at 과 progress.last_reviewed_at
|
|
은 의미가 다르지만 결과 화면 UX 에서 한 버튼으로 통합. reviewed=false 로 되돌릴 때는
|
|
progress 의 last_reviewed_at 은 건드리지 않는다 (다른 attempt 가 검토 표시했을 수 있음).
|
|
"""
|
|
from models.study_question_progress import StudyQuestionProgress
|
|
|
|
attempt = await session.get(StudyQuestionAttempt, attempt_id)
|
|
if attempt is None or attempt.user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="attempt 를 찾을 수 없습니다")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
attempt.reviewed_at = now if body.reviewed else None
|
|
|
|
# progress 동기화 — reviewed=true 일 때만. false 토글은 attempts 레벨만 되돌림.
|
|
if body.reviewed:
|
|
progress = (
|
|
await session.execute(
|
|
select(StudyQuestionProgress).where(
|
|
StudyQuestionProgress.user_id == user.id,
|
|
StudyQuestionProgress.study_topic_id == attempt.study_topic_id,
|
|
StudyQuestionProgress.study_question_id == attempt.study_question_id,
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if progress is not None:
|
|
progress.last_reviewed_at = now
|
|
# due_at 최초 부여 — wrong/unsure 인 progress 만, due_at 미박힌 경우만.
|
|
if progress.last_outcome in ("wrong", "unsure") and progress.due_at is None:
|
|
progress.review_stage = 0
|
|
progress.due_at = now + timedelta(days=1)
|
|
|
|
await session.commit()
|
|
return AttemptReviewResponse(id=attempt.id, reviewed_at=attempt.reviewed_at)
|
|
|
|
|
|
# ─── PR-5: 비슷한 문제 검색 (embedding cosine) ───
|
|
|
|
|
|
@router.get(
|
|
"/study-questions/{question_id}/similar",
|
|
response_model=SimilarQuestionsResponse,
|
|
)
|
|
async def list_similar_questions(
|
|
question_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
limit: int = Query(5, ge=1, le=20),
|
|
topic_only: bool = Query(True, description="True 면 같은 study_topic 안에서만"),
|
|
):
|
|
"""현재 문제의 embedding 기준 cosine 유사도 top-K. 26B 호출 없이 vector search 만.
|
|
|
|
제외:
|
|
- 자기 자신 (id != current)
|
|
- soft-deleted
|
|
- embedding_status != 'ready' (대기 중·실패·미생성)
|
|
|
|
source_status='ready' 가 아니면 items 빈 배열 반환 (UI 에서 안내).
|
|
"""
|
|
src = await session.get(StudyQuestion, question_id)
|
|
src = _verify_question_ownership(src, user)
|
|
|
|
if src.embedding_status != "ready" or src.embedding is None:
|
|
return SimilarQuestionsResponse(
|
|
items=[], source_status=src.embedding_status, source_id=question_id
|
|
)
|
|
|
|
# cosine 거리 = embedding <=> ref. 유사도 = 1 - distance.
|
|
distance_expr = StudyQuestion.embedding.cosine_distance(src.embedding)
|
|
base = (
|
|
select(StudyQuestion, distance_expr.label("distance"))
|
|
.where(
|
|
StudyQuestion.user_id == user.id,
|
|
StudyQuestion.id != question_id,
|
|
StudyQuestion.deleted_at.is_(None),
|
|
StudyQuestion.embedding_status == "ready",
|
|
StudyQuestion.embedding.is_not(None),
|
|
)
|
|
.order_by(distance_expr.asc())
|
|
.limit(limit)
|
|
)
|
|
if topic_only:
|
|
base = base.where(StudyQuestion.study_topic_id == src.study_topic_id)
|
|
|
|
rows = (await session.execute(base)).all()
|
|
similar_qs = [(r[0], float(r.distance)) for r in rows]
|
|
|
|
# attempt_count + last_correct batch (N+1 회피)
|
|
qids = [q.id for q, _ in similar_qs]
|
|
attempt_count_map: dict[int, int] = {}
|
|
last_correct_map: dict[int, bool] = {}
|
|
if qids:
|
|
cnt_rows = (
|
|
await session.execute(
|
|
select(
|
|
StudyQuestionAttempt.study_question_id,
|
|
func.count().label("total"),
|
|
)
|
|
.where(
|
|
StudyQuestionAttempt.user_id == user.id,
|
|
StudyQuestionAttempt.study_question_id.in_(qids),
|
|
)
|
|
.group_by(StudyQuestionAttempt.study_question_id)
|
|
)
|
|
).all()
|
|
for r in cnt_rows:
|
|
attempt_count_map[r.study_question_id] = int(r.total)
|
|
|
|
latest_rows = (
|
|
await session.execute(
|
|
select(
|
|
StudyQuestionAttempt.study_question_id,
|
|
StudyQuestionAttempt.is_correct,
|
|
)
|
|
.where(
|
|
StudyQuestionAttempt.user_id == user.id,
|
|
StudyQuestionAttempt.study_question_id.in_(qids),
|
|
)
|
|
.order_by(
|
|
StudyQuestionAttempt.study_question_id,
|
|
StudyQuestionAttempt.answered_at.desc(),
|
|
)
|
|
.distinct(StudyQuestionAttempt.study_question_id)
|
|
)
|
|
).all()
|
|
for r in latest_rows:
|
|
last_correct_map[r.study_question_id] = bool(r.is_correct)
|
|
|
|
def _truncate(text: str, n: int = 80) -> str:
|
|
return text if len(text) <= n else text[:n].rstrip() + "…"
|
|
|
|
items = [
|
|
SimilarQuestionItem(
|
|
id=q.id,
|
|
study_topic_id=q.study_topic_id,
|
|
question_text=_truncate(q.question_text, 80),
|
|
subject=q.subject,
|
|
scope=q.scope,
|
|
exam_name=q.exam_name,
|
|
exam_round=q.exam_round,
|
|
similarity=round(1.0 - dist, 4),
|
|
attempt_count=attempt_count_map.get(q.id, 0),
|
|
last_correct=last_correct_map.get(q.id),
|
|
)
|
|
for q, dist in similar_qs
|
|
]
|
|
return SimilarQuestionsResponse(items=items, source_status="ready", source_id=question_id)
|
|
|
|
|
|
# ─── PR-12-A: 반복 출제 / 유사 유형 분류 ───
|
|
|
|
|
|
class RelatedQuestionItem(BaseModel):
|
|
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
|
|
|
|
|
|
class RelatedTypesResponse(BaseModel):
|
|
source_id: int
|
|
source_status: str # ready | pending | failed | stale | none
|
|
source_exam_round: str | None # null/empty 면 두 리스트 모두 빈 + 카운트 0
|
|
|
|
# 다른 user / 같은 회차 / 자기 자신은 백엔드에서 모두 제외됨
|
|
repeat_questions: list[RelatedQuestionItem] # similarity >= 0.95
|
|
similar_questions: list[RelatedQuestionItem] # 0.88 <= similarity < 0.95
|
|
|
|
# 카운트 — round_count 가 1차 표시 기준 (회차 간 반복성).
|
|
repeat_related_count: int
|
|
repeat_round_count: int
|
|
repeat_grade: str | None = None # 단골/잘 나오는 반복 출제/반복 출제/신출/빈출 (round_count<2 면 null)
|
|
similar_related_count: int
|
|
similar_round_count: int
|
|
|
|
|
|
@router.get(
|
|
"/study-questions/{question_id}/related-types",
|
|
response_model=RelatedTypesResponse,
|
|
)
|
|
async def list_related_types(
|
|
question_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""반복 출제(🔥) / 유사 유형(🧩) 분리. 회차 조건 백엔드 강제.
|
|
|
|
학습 의미:
|
|
- 반복 출제 = 거의 같은 형태가 다른 회차에 다시 등장 (암기/패턴 고정 가치)
|
|
- 유사 유형 = 같은 개념·풀이 패턴이 다른 회차에 등장 (개념 복습 가치)
|
|
|
|
공통 service 함수 사용 — bulk endpoint 와 분류 로직 공유 (drift 회피).
|
|
|
|
PR-12-A 후속: study_questions.related_* 컬럼 캐시 우선 조회. cache hit (computed_at IS NOT NULL
|
|
+ threshold version 일치) 면 HNSW 검색 생략. miss 면 즉시 계산 + 캐시 저장 (cron 못 따라온 안전망).
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from services.study.related_types import (
|
|
THRESHOLD_VERSION,
|
|
classify_related_for_question,
|
|
deserialize_candidates,
|
|
serialize_candidates,
|
|
)
|
|
|
|
src = await session.get(StudyQuestion, question_id)
|
|
src = _verify_question_ownership(src, user)
|
|
|
|
src_round = (src.exam_round or "").strip()
|
|
src_round_or_none = src_round if src_round else None
|
|
|
|
# 임베딩 미준비 → 빈 응답.
|
|
if src.embedding_status != "ready" or src.embedding is None:
|
|
return RelatedTypesResponse(
|
|
source_id=question_id,
|
|
source_status=src.embedding_status,
|
|
source_exam_round=src_round_or_none,
|
|
repeat_questions=[],
|
|
similar_questions=[],
|
|
repeat_related_count=0,
|
|
repeat_round_count=0,
|
|
similar_related_count=0,
|
|
similar_round_count=0,
|
|
)
|
|
|
|
def _candidates_to_items(cands):
|
|
return [
|
|
RelatedQuestionItem(
|
|
id=c.id,
|
|
study_topic_id=c.study_topic_id,
|
|
question_text=c.question_text,
|
|
subject=c.subject,
|
|
scope=c.scope,
|
|
exam_round=c.exam_round,
|
|
exam_question_number=c.exam_question_number,
|
|
similarity=round(c.similarity, 4),
|
|
)
|
|
for c in cands
|
|
]
|
|
|
|
# cache hit 조건: computed_at 존재 + 임계값 fingerprint 일치
|
|
if (
|
|
src.related_computed_at is not None
|
|
and src.related_threshold_version == THRESHOLD_VERSION
|
|
):
|
|
repeat_cands = deserialize_candidates(src.related_repeat, src.study_topic_id)
|
|
similar_cands = deserialize_candidates(src.related_similar, src.study_topic_id)
|
|
return RelatedTypesResponse(
|
|
source_id=question_id,
|
|
source_status="ready",
|
|
source_exam_round=src_round_or_none,
|
|
repeat_questions=_candidates_to_items(repeat_cands),
|
|
similar_questions=_candidates_to_items(similar_cands),
|
|
repeat_related_count=len(repeat_cands),
|
|
repeat_round_count=src.related_repeat_round_count or 0,
|
|
repeat_grade=src.related_repeat_grade,
|
|
similar_related_count=len(similar_cands),
|
|
similar_round_count=src.related_similar_round_count or 0,
|
|
)
|
|
|
|
# cache miss → 즉시 계산 + 저장 (cron 다음 틱 기다리지 않음, 빈 응답 회피)
|
|
cls = await classify_related_for_question(session, user_id=user.id, source=src)
|
|
src.related_repeat = serialize_candidates(cls.repeat)
|
|
src.related_similar = serialize_candidates(cls.similar)
|
|
src.related_repeat_round_count = cls.repeat_round_count
|
|
src.related_similar_round_count = cls.similar_round_count
|
|
src.related_repeat_grade = cls.repeat_grade
|
|
src.related_computed_at = datetime.now(timezone.utc)
|
|
src.related_threshold_version = THRESHOLD_VERSION
|
|
await session.commit()
|
|
|
|
return RelatedTypesResponse(
|
|
source_id=question_id,
|
|
source_status="ready",
|
|
source_exam_round=src_round_or_none,
|
|
repeat_questions=_candidates_to_items(cls.repeat),
|
|
similar_questions=_candidates_to_items(cls.similar),
|
|
repeat_related_count=cls.repeat_related_count,
|
|
repeat_round_count=cls.repeat_round_count,
|
|
repeat_grade=cls.repeat_grade,
|
|
similar_related_count=cls.similar_related_count,
|
|
similar_round_count=cls.similar_round_count,
|
|
)
|
|
|
|
|
|
# ─── PR-8: 이미지 업로드/조회/삭제 ───
|
|
|
|
|
|
@router.post(
|
|
"/study-questions/{question_id}/images",
|
|
response_model=StudyQuestionImageItem,
|
|
status_code=201,
|
|
)
|
|
async def upload_question_image(
|
|
question_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
file: UploadFile = File(...),
|
|
):
|
|
"""문제 첨부 이미지 업로드. 1파일 = 1행. 여러 호출로 여러 이미지 추가.
|
|
|
|
경로: /documents/study_question_images/{topic_id}/{qid}/{img_id}.{ext}
|
|
인증: 같은 user 의 question 만. 다른 사용자는 404.
|
|
제한: PNG/JPEG/WEBP/GIF, 10MB/파일.
|
|
"""
|
|
q = await session.get(StudyQuestion, question_id)
|
|
q = _verify_question_ownership(q, user)
|
|
|
|
# MIME 검증
|
|
mime = (file.content_type or "").lower()
|
|
if mime not in ALLOWED_IMAGE_MIME:
|
|
raise HTTPException(
|
|
status_code=415,
|
|
detail=f"지원하지 않는 이미지 형식: {mime} (PNG/JPEG/WEBP/GIF 만 허용)",
|
|
)
|
|
|
|
# 파일 읽기 + 크기 검증
|
|
raw = await file.read()
|
|
if len(raw) > MAX_IMAGE_BYTES:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"이미지 크기 초과 ({len(raw)} > {MAX_IMAGE_BYTES} 바이트, 10MB 제한)",
|
|
)
|
|
if len(raw) == 0:
|
|
raise HTTPException(status_code=400, detail="빈 파일")
|
|
|
|
# 확장자 결정
|
|
ext_map = {
|
|
"image/png": "png",
|
|
"image/jpeg": "jpg",
|
|
"image/webp": "webp",
|
|
"image/gif": "gif",
|
|
}
|
|
ext = ext_map[mime]
|
|
|
|
# DB 행 먼저 만들고 id 받음 (파일명에 사용)
|
|
img = StudyQuestionImage(
|
|
user_id=user.id,
|
|
study_question_id=q.id,
|
|
file_path="", # 일단 빈 값, id 받은 후 채움
|
|
file_size=len(raw),
|
|
mime_type=mime,
|
|
sort_order=0,
|
|
)
|
|
session.add(img)
|
|
await session.flush() # img.id 확보
|
|
|
|
# 파일 시스템 저장
|
|
target_dir = STUDY_IMG_ROOT / str(q.study_topic_id) / str(q.id)
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
target_path = target_dir / f"{img.id}.{ext}"
|
|
target_path.write_bytes(raw)
|
|
|
|
img.file_path = str(target_path)
|
|
# sort_order = 같은 question 의 max+1
|
|
max_sort = (
|
|
await session.execute(
|
|
select(func.coalesce(func.max(StudyQuestionImage.sort_order), -1))
|
|
.where(StudyQuestionImage.study_question_id == q.id)
|
|
.where(StudyQuestionImage.id != img.id)
|
|
)
|
|
).scalar() or -1
|
|
img.sort_order = int(max_sort) + 1
|
|
await session.commit()
|
|
|
|
return _image_to_item(img)
|
|
|
|
|
|
@router.get("/study-questions/{question_id}/images/{image_id}/raw")
|
|
async def get_question_image_raw(
|
|
question_id: int,
|
|
image_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""이미지 raw bytes. 같은 user 의 question 첨부만 응답."""
|
|
img = await session.get(StudyQuestionImage, image_id)
|
|
if img is None or img.user_id != user.id or img.study_question_id != question_id:
|
|
raise HTTPException(status_code=404, detail="이미지를 찾을 수 없습니다")
|
|
file_path = _Path(img.file_path)
|
|
if not file_path.is_file():
|
|
raise HTTPException(status_code=410, detail="파일이 사라졌습니다")
|
|
return FileResponse(str(file_path), media_type=img.mime_type)
|
|
|
|
|
|
@router.delete("/study-questions/{question_id}/images/{image_id}", status_code=204)
|
|
async def delete_question_image(
|
|
question_id: int,
|
|
image_id: int,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
img = await session.get(StudyQuestionImage, image_id)
|
|
if img is None or img.user_id != user.id or img.study_question_id != question_id:
|
|
raise HTTPException(status_code=404, detail="이미지를 찾을 수 없습니다")
|
|
# 파일 시스템 삭제 (실패해도 DB row 는 정리)
|
|
try:
|
|
file_path = _Path(img.file_path)
|
|
if file_path.is_file():
|
|
file_path.unlink()
|
|
except Exception as e:
|
|
logger.warning("study_q_image_unlink_failed id=%s: %s", image_id, e)
|
|
await session.delete(img)
|
|
await session.commit()
|
|
|
|
|
|
# ─── PR-3: AI 풀이 생성 엔드포인트 ───
|
|
|
|
# MLX 호출 timeout (초). MLX gate + 26B 추론 평균 ~10s, 안전 마진.
|
|
LLM_TIMEOUT_S = 30.0
|
|
# 프롬프트 템플릿 lazy load
|
|
_PROMPT_PATH = "study_question_explanation.txt"
|
|
_prompt_cache: str | None = None
|
|
|
|
|
|
def _load_explanation_prompt() -> str:
|
|
global _prompt_cache
|
|
if _prompt_cache is None:
|
|
from pathlib import Path
|
|
prompts_dir = Path(__file__).resolve().parent.parent / "prompts"
|
|
_prompt_cache = (prompts_dir / _PROMPT_PATH).read_text(encoding="utf-8")
|
|
return _prompt_cache
|
|
|
|
|
|
def _render_prompt(q: StudyQuestion, doc_block: str, question_block: str) -> str:
|
|
template = _load_explanation_prompt()
|
|
return (
|
|
template
|
|
.replace("{question_text}", q.question_text)
|
|
.replace("{choice_1}", q.choice_1)
|
|
.replace("{choice_2}", q.choice_2)
|
|
.replace("{choice_3}", q.choice_3)
|
|
.replace("{choice_4}", q.choice_4)
|
|
.replace("{correct_choice}", str(q.correct_choice))
|
|
.replace("{documents_evidence_block}", doc_block)
|
|
.replace("{questions_evidence_block}", question_block)
|
|
)
|
|
|
|
|
|
def _make_cache_response(q: StudyQuestion, is_stale: bool) -> AIExplanationResponse:
|
|
"""캐시 본문만 반환 (evidence 재계산 안 함). status/메타는 q row 그대로."""
|
|
return AIExplanationResponse(
|
|
ai_explanation=q.ai_explanation,
|
|
ai_explanation_status=q.ai_explanation_status,
|
|
ai_explanation_generated_at=q.ai_explanation_generated_at,
|
|
ai_explanation_model=q.ai_explanation_model,
|
|
evidence=[],
|
|
from_cache=True,
|
|
is_stale=is_stale,
|
|
can_regenerate=True,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/study-questions/{question_id}/ai-explanation",
|
|
response_model=AIExplanationResponse,
|
|
)
|
|
async def generate_ai_explanation(
|
|
question_id: int,
|
|
body: AIExplanationCreate,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
):
|
|
"""수동 트리거 AI 풀이 생성. RAG = 매핑 자료 + 같은 토픽 다른 문제 (자기 자신 제외).
|
|
primary 모델 단독 (Mac mini MLX gemma-4-26b). MLX gate Semaphore(1) 경유.
|
|
|
|
- regenerate=false 이고 status=ready → 캐시 즉시 반환 (MLX 호출 없음)
|
|
- regenerate=false 이고 status=stale → 캐시 본문 반환, is_stale=true 플래그
|
|
- status=pending → 409 (이미 다른 호출 진행 중, race-safe 조건부 UPDATE 로 보장)
|
|
- 그 외 (none/failed/regenerate=true) → 새로 생성
|
|
|
|
실패 시 status='failed', 직전 본문은 보존.
|
|
"""
|
|
q = await session.get(StudyQuestion, question_id)
|
|
q = _verify_question_ownership(q, user)
|
|
|
|
# 1) 캐시 단축 분기 (regenerate=false)
|
|
if not body.regenerate:
|
|
if q.ai_explanation_status == "ready":
|
|
return _make_cache_response(q, is_stale=False)
|
|
if q.ai_explanation_status == "stale":
|
|
return _make_cache_response(q, is_stale=True)
|
|
if q.ai_explanation_status == "pending":
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={"status": "pending", "detail": "이미 생성 중입니다"},
|
|
)
|
|
# status in {none, failed} → 신규 생성
|
|
|
|
else:
|
|
# regenerate=true 라도 pending 은 차단 (동시 호출 방어)
|
|
if q.ai_explanation_status == "pending":
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={"status": "pending", "detail": "이미 생성 중입니다"},
|
|
)
|
|
|
|
# 2) Race-safe pending 전이 (조건부 UPDATE — 동시 호출 차단)
|
|
lock_result = await session.execute(
|
|
update(StudyQuestion)
|
|
.where(
|
|
StudyQuestion.id == question_id,
|
|
StudyQuestion.user_id == user.id,
|
|
StudyQuestion.ai_explanation_status != "pending",
|
|
)
|
|
.values(ai_explanation_status="pending", updated_at=datetime.now(timezone.utc))
|
|
.returning(StudyQuestion.id)
|
|
)
|
|
if lock_result.scalar_one_or_none() is None:
|
|
# 다른 호출이 먼저 pending 박음
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={"status": "pending", "detail": "이미 생성 중입니다"},
|
|
)
|
|
await session.commit()
|
|
await session.refresh(q)
|
|
|
|
# 3) RAG 근거 수집
|
|
try:
|
|
ctx = await gather_explanation_context(session, user.id, q)
|
|
except Exception as e:
|
|
logger.warning("study_explanation_rag_failed: %s: %s", type(e).__name__, e)
|
|
# RAG 실패 시 빈 evidence 로 진행 (프롬프트가 "자료 근거 부족" 분기 처리)
|
|
from services.study.explanation_rag import ExplanationContext
|
|
ctx = ExplanationContext(documents=[], questions=[])
|
|
|
|
# 4) 프롬프트 조립 + MLX gate 안에서 primary 호출 + timeout
|
|
doc_block = render_evidence_block(ctx.documents)
|
|
q_block = render_evidence_block(ctx.questions)
|
|
prompt = _render_prompt(q, doc_block, q_block)
|
|
|
|
ai_client = AIClient()
|
|
raw_text: str | None = None
|
|
error_message: str | None = None
|
|
try:
|
|
async with get_mlx_gate():
|
|
async with asyncio.timeout(LLM_TIMEOUT_S):
|
|
raw_text = await ai_client.call_primary(prompt)
|
|
except asyncio.TimeoutError:
|
|
error_message = f"MLX timeout ({LLM_TIMEOUT_S}s)"
|
|
logger.warning("study_explanation_mlx_timeout qid=%s", question_id)
|
|
except Exception as e:
|
|
error_message = f"{type(e).__name__}: {e}"
|
|
logger.exception("study_explanation_mlx_failed qid=%s", question_id)
|
|
finally:
|
|
await ai_client.close()
|
|
|
|
# 5) 결과 저장 (성공/실패 분기) — q 객체 fresh refresh
|
|
q = await session.get(StudyQuestion, question_id)
|
|
if raw_text is None or not raw_text.strip():
|
|
# 실패 — 직전 본문 보존, status='failed'
|
|
q.ai_explanation_status = "failed"
|
|
q.updated_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
return AIExplanationResponse(
|
|
ai_explanation=q.ai_explanation,
|
|
ai_explanation_status="failed",
|
|
ai_explanation_generated_at=q.ai_explanation_generated_at,
|
|
ai_explanation_model=q.ai_explanation_model,
|
|
evidence=[e.to_dict() for e in ctx.all],
|
|
from_cache=False,
|
|
is_stale=False,
|
|
can_regenerate=True,
|
|
)
|
|
|
|
# 성공 — Qwen think 태그 등 제거
|
|
from ai.client import strip_thinking
|
|
cleaned = strip_thinking(raw_text).strip()
|
|
|
|
q.ai_explanation = cleaned
|
|
q.ai_explanation_status = "ready"
|
|
q.ai_explanation_generated_at = datetime.now(timezone.utc)
|
|
primary_name = ai_client.ai.primary.model if hasattr(ai_client.ai.primary, "model") else "primary"
|
|
q.ai_explanation_model = f"mlx:{primary_name}"
|
|
q.updated_at = q.ai_explanation_generated_at
|
|
await session.commit()
|
|
|
|
return AIExplanationResponse(
|
|
ai_explanation=q.ai_explanation,
|
|
ai_explanation_status="ready",
|
|
ai_explanation_generated_at=q.ai_explanation_generated_at,
|
|
ai_explanation_model=q.ai_explanation_model,
|
|
evidence=[e.to_dict() for e in ctx.all],
|
|
from_cache=False,
|
|
is_stale=False,
|
|
can_regenerate=True,
|
|
)
|