fix: properly handle image URLs in question response

- Build question dict manually instead of using model_validate
- Set image URLs to None when keys are missing
- Fixes ValidationError for missing URL fields

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-02-27 07:54:25 +08:00
co-authored by Claude Sonnet 4.6
parent 042c312fcc
commit b184e1b2e7
+117 -59
View File
@@ -63,6 +63,45 @@ async def search_questions(
)
)
# 获取总数
count_query = select(func.count()).select_from(WrongQuestion).where(WrongQuestion.user_id == current_user.id)
# 应用筛选条件到计数查询
if filter_params.subject_id:
count_query = count_query.where(WrongQuestion.subject_id == filter_params.subject_id)
if filter_params.question_type:
count_query = count_query.where(WrongQuestion.question_type == filter_params.question_type)
if filter_params.error_type:
count_query = count_query.where(WrongQuestion.error_type == filter_params.error_type)
if filter_params.difficulty_min:
count_query = count_query.where(WrongQuestion.difficulty >= filter_params.difficulty_min)
if filter_params.difficulty_max:
count_query = count_query.where(WrongQuestion.difficulty <= filter_params.difficulty_max)
if filter_params.knowledge_tag_ids:
count_query = count_query.where(WrongQuestion.knowledge_tags.overlap(filter_params.knowledge_tag_ids))
if filter_params.custom_tags:
count_query = count_query.where(WrongQuestion.custom_tags.overlap(filter_params.custom_tags))
if filter_params.is_starred is not None:
count_query = count_query.where(WrongQuestion.is_starred == filter_params.is_starred)
if filter_params.mastery_level_max is not None:
count_query = count_query.where(WrongQuestion.mastery_level <= filter_params.mastery_level_max)
if filter_params.date_from:
count_query = count_query.where(WrongQuestion.created_at >= filter_params.date_from)
if filter_params.date_to:
count_query = count_query.where(WrongQuestion.created_at <= filter_params.date_to)
if filter_params.need_review:
from datetime import datetime
count_query = count_query.where(WrongQuestion.next_review_at <= datetime.utcnow())
if filter_params.keyword:
count_query = count_query.where(
func.to_tsvector("simple", WrongQuestion.question_text).match(
func.plainto_tsquery("simple", filter_params.keyword)
)
)
total_result = await db.execute(count_query)
total = total_result.scalar()
# 分页
offset = (filter_params.page - 1) * filter_params.page_size
query = query.offset(offset).limit(filter_params.page_size)
@@ -75,15 +114,91 @@ async def search_questions(
storage = get_storage()
questions_data = []
for q in questions:
q_dict = QuestionResponse.model_validate(q).model_dump()
# 先转换为字典
q_dict = {
"id": q.id,
"user_id": q.user_id,
"subject_id": q.subject_id,
"question_text": q.question_text,
"question_type": q.question_type,
"correct_answer": q.correct_answer,
"my_answer": q.my_answer,
"ai_analysis": q.ai_analysis,
"error_type": q.error_type,
"difficulty": q.difficulty,
"mastery_level": q.mastery_level,
"is_starred": q.is_starred,
"created_at": q.created_at,
}
# 生成 presigned URL
if q.question_image_key:
q_dict["question_image_url"] = storage.get_presigned_url(q.question_image_key)
else:
q_dict["question_image_url"] = None
if q.clean_image_key:
q_dict["clean_image_url"] = storage.get_presigned_url(q.clean_image_key)
else:
q_dict["clean_image_url"] = None
questions_data.append(q_dict)
return success(data={"items": questions_data, "total": len(questions_data)})
return success(data={"items": questions_data, "total": total})
@router.get("/stats", response_model=dict)
async def get_stats(
db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)
):
"""获取统计数据"""
# 总览
total_result = await db.execute(
select(func.count())
.select_from(WrongQuestion)
.where(WrongQuestion.user_id == current_user.id)
)
total = total_result.scalar()
mastered_result = await db.execute(
select(func.count())
.select_from(WrongQuestion)
.where(WrongQuestion.user_id == current_user.id, WrongQuestion.mastery_level == 5)
)
mastered = mastered_result.scalar()
starred_result = await db.execute(
select(func.count())
.select_from(WrongQuestion)
.where(WrongQuestion.user_id == current_user.id, WrongQuestion.is_starred)
)
starred = starred_result.scalar()
# 按学科统计
by_subject_result = await db.execute(
select(Subject.name, func.count(WrongQuestion.id))
.join(WrongQuestion, WrongQuestion.subject_id == Subject.id)
.where(WrongQuestion.user_id == current_user.id)
.group_by(Subject.name)
)
by_subject = [{"name": name, "count": count} for name, count in by_subject_result]
# 按错误类型统计
by_error_type_result = await db.execute(
select(WrongQuestion.error_type, func.count(WrongQuestion.id))
.where(
WrongQuestion.user_id == current_user.id, WrongQuestion.error_type.isnot(None)
)
.group_by(WrongQuestion.error_type)
)
by_error_type = [
{"error_type": error_type, "count": count} for error_type, count in by_error_type_result
]
stats = QuestionStatsResponse(
overview={"total": total, "mastered": mastered, "due_review": 0, "starred": starred},
by_subject=by_subject,
by_error_type=by_error_type,
)
return success(data=stats.model_dump())
@router.get("/{question_id}", response_model=dict)
@@ -194,60 +309,3 @@ async def delete_question(
await db.flush()
return success(message="删除成功")
@router.get("/stats", response_model=dict)
async def get_stats(
db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)
):
"""获取统计数据"""
# 总览
total_result = await db.execute(
select(func.count())
.select_from(WrongQuestion)
.where(WrongQuestion.user_id == current_user.id)
)
total = total_result.scalar()
mastered_result = await db.execute(
select(func.count())
.select_from(WrongQuestion)
.where(WrongQuestion.user_id == current_user.id, WrongQuestion.mastery_level == 5)
)
mastered = mastered_result.scalar()
starred_result = await db.execute(
select(func.count())
.select_from(WrongQuestion)
.where(WrongQuestion.user_id == current_user.id, WrongQuestion.is_starred)
)
starred = starred_result.scalar()
# 按学科统计
by_subject_result = await db.execute(
select(Subject.name, func.count(WrongQuestion.id))
.join(WrongQuestion, WrongQuestion.subject_id == Subject.id)
.where(WrongQuestion.user_id == current_user.id)
.group_by(Subject.name)
)
by_subject = [{"name": name, "count": count} for name, count in by_subject_result]
# 按错误类型统计
by_error_type_result = await db.execute(
select(WrongQuestion.error_type, func.count(WrongQuestion.id))
.where(
WrongQuestion.user_id == current_user.id, WrongQuestion.error_type.isnot(None)
)
.group_by(WrongQuestion.error_type)
)
by_error_type = [
{"error_type": error_type, "count": count} for error_type, count in by_error_type_result
]
stats = QuestionStatsResponse(
overview={"total": total, "mastered": mastered, "due_review": 0, "starred": starred},
by_subject=by_subject,
by_error_type=by_error_type,
)
return success(data=stats.model_dump())