193 lines
6.5 KiB
Python
193 lines
6.5 KiB
Python
"""笔记 CRUD 接口，全部按用户隔离。"""
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.db import get_db
|
|
from app.deps import get_current_user
|
|
from app.models.question import Question, QuestionType, Tag
|
|
from app.models.user import User
|
|
from app.schemas.question import (
|
|
QuestionCreate,
|
|
QuestionList,
|
|
QuestionOut,
|
|
QuestionUpdate,
|
|
)
|
|
from app.services.tags import get_or_create_tags
|
|
|
|
router = APIRouter(prefix="/api/questions", tags=["questions"])
|
|
|
|
|
|
async def _get_owned_question(db: AsyncSession, user_id: str, question_id: str) -> Question:
|
|
q = await db.scalar(
|
|
select(Question)
|
|
.where(Question.id == question_id, Question.user_id == user_id)
|
|
.options(selectinload(Question.tags))
|
|
)
|
|
if q is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="题目不存在")
|
|
return q
|
|
|
|
|
|
def _options_to_json(options) -> list | None:
|
|
if options is None:
|
|
return None
|
|
return [o.model_dump() for o in options]
|
|
|
|
|
|
@router.get("", response_model=QuestionList)
|
|
async def list_questions(
|
|
db: AsyncSession = Depends(get_db),
|
|
current: User = Depends(get_current_user),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
type: QuestionType | None = None,
|
|
tag: list[str] | None = Query(None, description="知识点，可多个（全部满足）"),
|
|
q: str | None = Query(None, description="关键词，搜题干与我的笔记"),
|
|
difficulty: int | None = Query(None, ge=1, le=5),
|
|
difficulty_min: int | None = Query(None, ge=1, le=5),
|
|
difficulty_max: int | None = Query(None, ge=1, le=5),
|
|
created_after: datetime | None = Query(None, description="记录时间下限（含）"),
|
|
created_before: datetime | None = Query(None, description="记录时间上限（含）"),
|
|
needs_review: bool | None = Query(None, description="只看待复习的"),
|
|
untagged: bool = Query(False, description="只看还没归类知识点的"),
|
|
sort: Literal["created_desc", "created_asc", "updated_desc"] = "created_desc",
|
|
):
|
|
"""多维检索：知识点 + 时间 + 难度 + 待复习状态，可自由组合。"""
|
|
stmt = select(Question).where(Question.user_id == current.id)
|
|
|
|
if type is not None:
|
|
stmt = stmt.where(Question.type == type)
|
|
if q:
|
|
# 题干和我自己的笔记都要能搜到
|
|
stmt = stmt.where(
|
|
or_(
|
|
Question.stem_markdown.contains(q),
|
|
Question.my_note_markdown.contains(q),
|
|
)
|
|
)
|
|
if difficulty is not None:
|
|
stmt = stmt.where(Question.difficulty == difficulty)
|
|
if difficulty_min is not None:
|
|
stmt = stmt.where(Question.difficulty >= difficulty_min)
|
|
if difficulty_max is not None:
|
|
stmt = stmt.where(Question.difficulty <= difficulty_max)
|
|
if created_after is not None:
|
|
stmt = stmt.where(Question.created_at >= created_after)
|
|
if created_before is not None:
|
|
stmt = stmt.where(Question.created_at <= created_before)
|
|
if needs_review is not None:
|
|
stmt = stmt.where(Question.needs_review.is_(needs_review))
|
|
if untagged:
|
|
stmt = stmt.where(~Question.tags.any())
|
|
if tag:
|
|
# 多个知识点取交集：每个都要命中
|
|
for name in tag:
|
|
stmt = stmt.where(
|
|
Question.tags.any(and_(Tag.user_id == current.id, Tag.name == name))
|
|
)
|
|
|
|
total = await db.scalar(
|
|
select(func.count()).select_from(stmt.order_by(None).subquery())
|
|
)
|
|
|
|
order = {
|
|
"created_desc": Question.created_at.desc(),
|
|
"created_asc": Question.created_at.asc(),
|
|
"updated_desc": Question.updated_at.desc(),
|
|
}[sort]
|
|
|
|
stmt = (
|
|
stmt.options(selectinload(Question.tags))
|
|
.order_by(order)
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
)
|
|
rows = (await db.scalars(stmt)).unique().all()
|
|
return QuestionList(
|
|
items=[QuestionOut.from_model(x) for x in rows],
|
|
total=total or 0,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.post("", response_model=QuestionOut, status_code=status.HTTP_201_CREATED)
|
|
async def create_question(
|
|
payload: QuestionCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current: User = Depends(get_current_user),
|
|
):
|
|
question = Question(
|
|
user_id=current.id,
|
|
type=payload.type,
|
|
stem_markdown=payload.stem_markdown,
|
|
options=_options_to_json(payload.options),
|
|
correct_answer=payload.correct_answer,
|
|
explanation_markdown=payload.explanation_markdown,
|
|
my_note_markdown=payload.my_note_markdown,
|
|
needs_review=payload.needs_review,
|
|
difficulty=payload.difficulty,
|
|
)
|
|
question.tags = await get_or_create_tags(db, current.id, payload.tags)
|
|
db.add(question)
|
|
await db.commit()
|
|
await db.refresh(question, ["tags"])
|
|
return QuestionOut.from_model(question)
|
|
|
|
|
|
@router.get("/{question_id}", response_model=QuestionOut)
|
|
async def get_question(
|
|
question_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current: User = Depends(get_current_user),
|
|
):
|
|
q = await _get_owned_question(db, current.id, question_id)
|
|
return QuestionOut.from_model(q)
|
|
|
|
|
|
@router.put("/{question_id}", response_model=QuestionOut)
|
|
async def update_question(
|
|
question_id: str,
|
|
payload: QuestionUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current: User = Depends(get_current_user),
|
|
):
|
|
q = await _get_owned_question(db, current.id, question_id)
|
|
data = payload.model_dump(exclude_unset=True)
|
|
|
|
if "options" in data:
|
|
q.options = _options_to_json(payload.options)
|
|
if "tags" in data and payload.tags is not None:
|
|
q.tags = await get_or_create_tags(db, current.id, payload.tags)
|
|
for field in (
|
|
"type",
|
|
"stem_markdown",
|
|
"correct_answer",
|
|
"explanation_markdown",
|
|
"my_note_markdown",
|
|
"needs_review",
|
|
"difficulty",
|
|
):
|
|
if field in data:
|
|
setattr(q, field, data[field])
|
|
|
|
await db.commit()
|
|
await db.refresh(q, ["tags"])
|
|
return QuestionOut.from_model(q)
|
|
|
|
|
|
@router.delete("/{question_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_question(
|
|
question_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current: User = Depends(get_current_user),
|
|
):
|
|
q = await _get_owned_question(db, current.id, question_id)
|
|
await db.delete(q)
|
|
await db.commit()
|