from fastapi import FastAPI, HTTPException, Depends, UploadFile, File, status, Request from fastapi.responses import FileResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.staticfiles import StaticFiles from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Boolean, Text, JSON, func from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session from pydantic import BaseModel from typing import List, Optional import pandas as pd import jwt import bcrypt from datetime import datetime, timedelta, date import json import io import csv import shutil import tempfile import uuid from pdf_splitter_hybrid import PDFSplitterHybrid as PDFSplitter from pdf_exam_parser_enhanced_v2 import PDFExamParserEnhancedV2 from pdf_image_extractor import PDFImageExtractor from utils.question_formatter import QuestionFormatter from claude_exam_parser import ClaudeExamParser import uvicorn import uuid import os import logging from dotenv import load_dotenv # 加载环境变量 load_dotenv() # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 从环境变量读取配置 DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/wordmaster") SECRET_KEY = os.getenv("SECRET_KEY", "your-super-secret-and-secure-key-change-me") ALGORITHM = os.getenv("ALGORITHM", "HS256") ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "30")) # 数据库配置 engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() # 数据库模型 class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String, unique=True, index=True) email = Column(String, unique=True, index=True) hashed_password = Column(String) created_at = Column(DateTime, default=datetime.utcnow) class Word(Base): __tablename__ = "words" id = Column(Integer, primary_key=True, index=True) word = Column(String, index=True) meaning = Column(String) example = Column(Text, nullable=True) phrases = Column(Text, nullable=True) # JSON string level = Column(String, default="middle_school") # elementary, middle_school, high_school created_at = Column(DateTime, default=datetime.utcnow) class UserWordProgress(Base): __tablename__ = "user_word_progress" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, index=True) word_id = Column(Integer, index=True) ef = Column(Float, default=2.5) # SM-2 ease factor interval = Column(Integer, default=0) # days reps = Column(Integer, default=0) # repetitions due = Column(DateTime, default=datetime.utcnow) correct_count = Column(Integer, default=0) wrong_count = Column(Integer, default=0) streak_wrong = Column(Integer, default=0) is_leech = Column(Boolean, default=False) last_reviewed = Column(DateTime, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) class Paper(Base): __tablename__ = "papers" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, index=True) title = Column(String, index=True) subject = Column(String, index=True) # math, chemistry, physics, english grade = Column(String) # grade7, grade8, grade9, etc. exam_type = Column(String) # midterm, final, monthly, practice exam_date = Column(DateTime) total_score = Column(Float) achieved_score = Column(Float, nullable=True) file_key = Column(String) # 文件存储路径 pdf_key = Column(String, nullable=True) # PDF文件路径 thumb_key = Column(String, nullable=True) # 缩略图路径 tags = Column(JSON, nullable=True) # 标签数组 ocr_text = Column(Text, nullable=True) # OCR提取的文本 created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class Mistake(Base): __tablename__ = "mistakes" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, index=True) paper_id = Column(Integer, index=True) page = Column(Integer) # 页码 bbox = Column(JSON) # 边界框 {x, y, width, height} image_key = Column(String, nullable=True) # 错题图片路径 subject = Column(String, index=True) knowledge_points = Column(JSON) # 知识点数组 reason_tag = Column(String) # concept, calc, careless, memory, steps, reading difficulty = Column(Integer, default=3) # 1-5 review_level = Column(Integer, default=1) # Leitner盒子级别 1-5 next_review_date = Column(DateTime, default=datetime.utcnow) last_score = Column(Integer, nullable=True) # 最后一次复习评分 1-5 answer = Column(Text, nullable=True) # 正确答案 analysis = Column(Text, nullable=True) # 解析 notes = Column(Text, nullable=True) # 家长/学生备注 created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class ReviewLog(Base): __tablename__ = "review_logs" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, index=True) mistake_id = Column(Integer, index=True) reviewed_at = Column(DateTime, default=datetime.utcnow) rating = Column(Integer) # 1-5评分 next_review_date = Column(DateTime) comment = Column(Text, nullable=True) time_spent = Column(Integer, nullable=True) # 秒 class Question(Base): __tablename__ = "questions" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, index=True) paper_id = Column(Integer, index=True, nullable=True) # 来源试卷 question_type = Column(String) # choice, fill_blank, essay, calculation subject = Column(String, index=True) # math, chemistry, physics, english grade = Column(String) # grade7, grade8, etc. chapter = Column(String, nullable=True) # 章节 knowledge_points = Column(JSON) # 知识点数组 difficulty = Column(Integer, default=3) # 1-5难度 question_content = Column(Text) # 题目内容 options = Column(JSON, nullable=True) # 选择题选项 {A, B, C, D} correct_answer = Column(Text) # 正确答案 explanation = Column(Text, nullable=True) # 解析 image_key = Column(String, nullable=True) # 题目图片 tags = Column(JSON, nullable=True) # 标签数组 # 拆题功能相关字段 page_spans = Column(JSON, nullable=True) # 页面位置信息 [{"page":1,"bbox":[x,y,w,h]}] source_text = Column(Text, nullable=True) # OCR原始文本 images = Column(JSON, nullable=True) # 题目中的图片信息 [{"page":1,"bbox":[...],"key":"..."}] formula = Column(JSON, nullable=True) # 公式信息 [{"latex":"...","bbox":[...]}] score = Column(Integer, nullable=True) # 分值 question_number = Column(String, nullable=True) # 题号 (1, 2, 3 等) auto_extracted = Column(Boolean, default=False) # 是否自动提取 created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class QuestionCollection(Base): __tablename__ = "question_collections" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, index=True) name = Column(String) # 题集名称 description = Column(Text, nullable=True) # 描述 subject = Column(String, index=True) question_ids = Column(JSON) # 题目ID数组 is_public = Column(Boolean, default=False) # 是否公开 created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # 创建表 Base.metadata.create_all(bind=engine) # Pydantic模型 class UserCreate(BaseModel): username: str email: str password: str class UserLogin(BaseModel): username: str password: str class Token(BaseModel): access_token: str token_type: str class WordCreate(BaseModel): word: str meaning: str example: Optional[str] = None phrases: Optional[List[str]] = None level: str = "middle_school" class WordResponse(BaseModel): id: int word: str meaning: str example: Optional[str] phrases: Optional[List[str]] level: str class Config: from_attributes = True class StudySession(BaseModel): word_id: int quality: int # 0-5 for SM-2 time_spent: int # seconds class StatsResponse(BaseModel): total_words: int due_today: int completed_today: int accuracy_rate: float streak_days: int class PaperCreate(BaseModel): title: str subject: str # math, chemistry, physics, english grade: str exam_type: str exam_date: datetime total_score: float achieved_score: Optional[float] = None file_key: str tags: Optional[List[str]] = None class PaperResponse(BaseModel): id: int title: str subject: str grade: str exam_type: str exam_date: datetime total_score: float achieved_score: Optional[float] file_key: str pdf_key: Optional[str] thumb_key: Optional[str] tags: Optional[List[str]] created_at: datetime class Config: from_attributes = True class MistakeCreate(BaseModel): paper_id: int page: int bbox: dict # {x, y, width, height} subject: str knowledge_points: List[str] reason_tag: str difficulty: int = 3 answer: Optional[str] = None analysis: Optional[str] = None notes: Optional[str] = None class MistakeResponse(BaseModel): id: int paper_id: int page: int bbox: dict image_key: Optional[str] subject: str knowledge_points: List[str] reason_tag: str difficulty: int review_level: int next_review_date: datetime last_score: Optional[int] answer: Optional[str] analysis: Optional[str] notes: Optional[str] created_at: datetime class Config: from_attributes = True class ReviewSubmit(BaseModel): mistake_id: int rating: int # 1-5 comment: Optional[str] = None time_spent: Optional[int] = None class FileUploadRequest(BaseModel): filename: str content_type: str class FileUploadResponse(BaseModel): file_key: str upload_url: str class DashboardStats(BaseModel): papers_count: int mistakes_count: int due_reviews: int weak_subjects: List[dict] recent_accuracy: float subject_stats: dict # 按学科分类的统计数据 class QuestionCreate(BaseModel): paper_id: Optional[int] = None question_type: str # choice, fill_blank, essay, calculation subject: str grade: str chapter: Optional[str] = None knowledge_points: List[str] difficulty: int = 3 question_content: str options: Optional[dict] = None # {A: "option1", B: "option2", ...} correct_answer: str explanation: Optional[str] = None tags: Optional[List[str]] = None class QuestionResponse(BaseModel): id: int paper_id: Optional[int] question_type: str subject: str grade: str chapter: Optional[str] knowledge_points: List[str] difficulty: int question_content: str options: Optional[dict] correct_answer: str explanation: Optional[str] image_key: Optional[str] images: Optional[List[dict]] # 题目关联的图片信息 question_number: Optional[str] = None # 添加题号字段，设为可选且有默认值 tags: Optional[List[str]] created_at: datetime class Config: from_attributes = True class QuestionCollectionCreate(BaseModel): name: str description: Optional[str] = None subject: str question_ids: List[int] is_public: bool = False class QuestionCollectionResponse(BaseModel): id: int name: str description: Optional[str] subject: str question_ids: List[int] is_public: bool created_at: datetime class Config: from_attributes = True # 拆题相关模型 class SplitRequest(BaseModel): paper_id: int class SplitQuestionResult(BaseModel): number: str type: str text: str options: Optional[dict] = None bbox: List[int] page: int confidence: float image_filename: Optional[str] = None class SplitResponse(BaseModel): success: bool paper_id: int total_questions: int questions: List[SplitQuestionResult] stats: dict error: Optional[str] = None # 导入图片API路由 from api_images import router as images_router # FastAPI应用 app = FastAPI(title="学习助手 Pro API", version="1.0.0", description="智能试卷管理与学习跟踪系统API") # 注册图片API路由 app.include_router(images_router) # CORS中间件 app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境中应该限制具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 静态文件服务 - 提供图片访问 app.mount("/images", StaticFiles(directory="uploads/images"), name="images") # 依赖 def get_db(): db = SessionLocal() try: yield db finally: db.close() security = HTTPBearer() def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db)): try: payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise HTTPException(status_code=401, detail="Invalid token") except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Invalid token") user = db.query(User).filter(User.username == username).first() if user is None: raise HTTPException(status_code=401, detail="User not found") return user # 工具函数 def verify_password(plain_password: str, hashed_password: str) -> bool: return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) def get_password_hash(password: str) -> str: return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def sm2_schedule(ef: float, interval: int, reps: int, quality: int) -> tuple: """SM-2算法实现""" ef = max(1.3, ef + (0.1 - (5 - quality) * (0.08 + (5 - quality) * 0.02))) if quality < 3: reps = 0 interval = 1 else: reps += 1 if reps == 1: interval = 1 elif reps == 2: interval = 3 else: interval = round(interval * ef) return ef, interval, reps # API路由 @app.post("/api/auth/register", response_model=Token) async def register(user: UserCreate, db: Session = Depends(get_db)): # 检查用户名是否存在 db_user = db.query(User).filter(User.username == user.username).first() if db_user: raise HTTPException(status_code=400, detail="Username already registered") # 检查邮箱是否存在 db_user = db.query(User).filter(User.email == user.email).first() if db_user: raise HTTPException(status_code=400, detail="Email already registered") # 创建用户 hashed_password = get_password_hash(user.password) db_user = User(username=user.username, email=user.email, hashed_password=hashed_password) db.add(db_user) db.commit() db.refresh(db_user) # 生成令牌 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.post("/api/auth/login", response_model=Token) async def login(user: UserLogin, db: Session = Depends(get_db)): db_user = db.query(User).filter(User.username == user.username).first() if not db_user or not verify_password(user.password, db_user.hashed_password): raise HTTPException(status_code=401, detail="Incorrect username or password") access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.get("/api/words", response_model=List[WordResponse]) async def get_words(skip: int = 0, limit: int = 100, level: Optional[str] = None, db: Session = Depends(get_db)): query = db.query(Word) if level: query = query.filter(Word.level == level) words = query.offset(skip).limit(limit).all() # 处理phrases字段 result = [] for word in words: word_dict = { "id": word.id, "word": word.word, "meaning": word.meaning, "example": word.example, "level": word.level, "phrases": json.loads(word.phrases) if word.phrases else [] } result.append(word_dict) return result @app.post("/api/words", response_model=WordResponse) async def create_word(word: WordCreate, db: Session = Depends(get_db)): phrases_json = json.dumps(word.phrases) if word.phrases else None db_word = Word( word=word.word, meaning=word.meaning, example=word.example, phrases=phrases_json, level=word.level ) db.add(db_word) db.commit() db.refresh(db_word) return { "id": db_word.id, "word": db_word.word, "meaning": db_word.meaning, "example": db_word.example, "phrases": json.loads(db_word.phrases) if db_word.phrases else [], "level": db_word.level } @app.post("/api/words/import") async def import_words(file: UploadFile = File(...), db: Session = Depends(get_db)): if not file.filename.endswith(('.csv', '.json')): raise HTTPException(status_code=400, detail="Only CSV and JSON files are supported") content = await file.read() imported_count = 0 try: if file.filename.endswith('.csv'): # 处理CSV文件 csv_content = content.decode('utf-8') csv_reader = csv.DictReader(io.StringIO(csv_content)) for row in csv_reader: if not row.get('word') or not row.get('meaning'): continue phrases = [] if row.get('phrases'): phrases = [p.strip() for p in row['phrases'].split(';') if p.strip()] phrases_json = json.dumps(phrases) if phrases else None db_word = Word( word=row['word'].strip(), meaning=row['meaning'].strip(), example=row.get('example', '').strip() or None, phrases=phrases_json, level=row.get('level', 'middle_school').strip() ) db.add(db_word) imported_count += 1 elif file.filename.endswith('.json'): # 处理JSON文件 json_data = json.loads(content.decode('utf-8')) for item in json_data: if not item.get('word') or not item.get('meaning'): continue phrases_json = json.dumps(item.get('phrases', [])) if item.get('phrases') else None db_word = Word( word=item['word'].strip(), meaning=item['meaning'].strip(), example=item.get('example', '').strip() or None, phrases=phrases_json, level=item.get('level', 'middle_school').strip() ) db.add(db_word) imported_count += 1 db.commit() return {"message": f"Successfully imported {imported_count} words"} except Exception as e: db.rollback() raise HTTPException(status_code=400, detail=f"Import failed: {str(e)}") @app.get("/api/study/due") async def get_due_words(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): """获取今日待学习的单词""" now = datetime.utcnow() # 获取用户的学习进度 progress_query = db.query(UserWordProgress).filter( UserWordProgress.user_id == current_user.id, UserWordProgress.due <= now ) due_progress = progress_query.all() # 如果没有待学习的单词，添加新单词 if len(due_progress) < 10: # 保持至少10个待学习单词 # 获取用户尚未学习的单词 learned_word_ids = db.query(UserWordProgress.word_id).filter( UserWordProgress.user_id == current_user.id ).subquery() new_words = db.query(Word).filter( ~Word.id.in_(learned_word_ids) ).limit(10 - len(due_progress)).all() # 为新单词创建学习进度 for word in new_words: progress = UserWordProgress( user_id=current_user.id, word_id=word.id, due=now ) db.add(progress) due_progress.append(progress) db.commit() # 获取完整的单词信息 word_ids = [p.word_id for p in due_progress] words = db.query(Word).filter(Word.id.in_(word_ids)).all() result = [] for word in words: progress = next(p for p in due_progress if p.word_id == word.id) result.append({ "id": word.id, "word": word.word, "meaning": word.meaning, "example": word.example, "phrases": json.loads(word.phrases) if word.phrases else [], "level": word.level, "progress": { "ef": progress.ef, "interval": progress.interval, "reps": progress.reps, "correct_count": progress.correct_count, "wrong_count": progress.wrong_count, "is_leech": progress.is_leech } }) return result @app.post("/api/study/submit") async def submit_study_session( session: StudySession, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """提交学习结果""" progress = db.query(UserWordProgress).filter( UserWordProgress.user_id == current_user.id, UserWordProgress.word_id == session.word_id ).first() if not progress: raise HTTPException(status_code=404, detail="Word progress not found") # 更新统计信息 is_correct = session.quality >= 3 if is_correct: progress.correct_count += 1 progress.streak_wrong = 0 if progress.is_leech and progress.correct_count % 3 == 0: progress.is_leech = False progress.ef = min(2.5, progress.ef + 0.1) else: progress.wrong_count += 1 progress.streak_wrong += 1 if progress.streak_wrong >= 3: progress.is_leech = True progress.ef = max(1.3, progress.ef - 0.2) # 使用SM-2算法计算下次复习时间 ef, interval, reps = sm2_schedule(progress.ef, progress.interval, progress.reps, session.quality) progress.ef = ef progress.interval = interval progress.reps = reps progress.due = datetime.utcnow() + timedelta(days=interval) progress.last_reviewed = datetime.utcnow() db.commit() return {"message": "Study session submitted successfully", "next_due": progress.due} @app.get("/api/stats", response_model=StatsResponse) async def get_stats(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): """获取学习统计""" now = datetime.utcnow() today_start = datetime(now.year, now.month, now.day) # 总词汇量 total_words = db.query(UserWordProgress).filter( UserWordProgress.user_id == current_user.id ).count() # 今日待复习 due_today = db.query(UserWordProgress).filter( UserWordProgress.user_id == current_user.id, UserWordProgress.due <= now ).count() # 今日已完成 completed_today = db.query(UserWordProgress).filter( UserWordProgress.user_id == current_user.id, UserWordProgress.last_reviewed >= today_start ).count() # 总体正确率 all_progress = db.query(UserWordProgress).filter( UserWordProgress.user_id == current_user.id ).all() total_correct = sum(p.correct_count for p in all_progress) total_attempts = sum(p.correct_count + p.wrong_count for p in all_progress) accuracy_rate = (total_correct / total_attempts * 100) if total_attempts > 0 else 0 # 连续学习天数（简化计算） streak_days = 1 # 这里可以实现更复杂的连续天数计算 return StatsResponse( total_words=total_words, due_today=due_today, completed_today=completed_today, accuracy_rate=accuracy_rate, streak_days=streak_days ) @app.get("/api/words/leech") async def get_leech_words(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): """获取顽固词汇""" leech_progress = db.query(UserWordProgress).filter( UserWordProgress.user_id == current_user.id, UserWordProgress.is_leech == True ).all() word_ids = [p.word_id for p in leech_progress] words = db.query(Word).filter(Word.id.in_(word_ids)).all() result = [] for word in words: progress = next(p for p in leech_progress if p.word_id == word.id) result.append({ "id": word.id, "word": word.word, "meaning": word.meaning, "wrong_count": progress.wrong_count, "streak_wrong": progress.streak_wrong }) return result # 文件上传相关 @app.post("/api/file/presign", response_model=FileUploadResponse) async def create_presigned_url( request: FileUploadRequest, current_user: User = Depends(get_current_user) ): """生成预签名上传URL""" # 简化版本：直接使用本地存储 file_key = f"uploads/{current_user.id}/{uuid.uuid4()}_{request.filename}" upload_dir = os.path.dirname(file_key) os.makedirs(upload_dir, exist_ok=True) return FileUploadResponse( file_key=file_key, upload_url=f"http://localhost:8888/api/file/upload/{file_key}" ) @app.post("/api/file/upload/") async def upload_file_direct( file: UploadFile = File(...), current_user: User = Depends(get_current_user) ): """直接文件上传""" try: # 检查文件名 if not file.filename or file.filename == "": return {"error": "No filename provided", "filename": repr(file.filename)} # 安全的文件名处理 safe_filename = file.filename.replace(" ", "_").replace("/", "_") file_key = f"uploads/{current_user.id}/{uuid.uuid4()}_{safe_filename}" # 确保目录存在 upload_dir = os.path.dirname(file_key) if upload_dir: os.makedirs(upload_dir, exist_ok=True) # 读取并保存文件 content = await file.read() with open(file_key, "wb") as buffer: buffer.write(content) return { "message": "File uploaded successfully", "file_key": file_key, "original_filename": file.filename, "file_size": len(content) } except Exception as e: return {"error": str(e), "filename": getattr(file, 'filename', 'unknown')} @app.post("/api/file/upload/{file_path:path}") async def upload_file(file_path: str, file: UploadFile = File(...)): """文件上传""" try: with open(file_path, "wb") as buffer: content = await file.read() buffer.write(content) return {"message": "File uploaded successfully", "file_key": file_path} except Exception as e: raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}") # 试卷管理 @app.post("/api/papers", response_model=PaperResponse) async def create_paper( paper: PaperCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """创建试卷记录""" db_paper = Paper( user_id=current_user.id, title=paper.title, subject=paper.subject, grade=paper.grade, exam_type=paper.exam_type, exam_date=paper.exam_date, total_score=paper.total_score, achieved_score=paper.achieved_score, file_key=paper.file_key, tags=paper.tags ) db.add(db_paper) db.commit() db.refresh(db_paper) return PaperResponse( id=db_paper.id, title=db_paper.title, subject=db_paper.subject, grade=db_paper.grade, exam_type=db_paper.exam_type, exam_date=db_paper.exam_date, total_score=db_paper.total_score, achieved_score=db_paper.achieved_score, file_key=db_paper.file_key, pdf_key=db_paper.pdf_key, thumb_key=db_paper.thumb_key, tags=db_paper.tags, created_at=db_paper.created_at ) @app.get("/api/papers", response_model=List[PaperResponse]) async def get_papers( subject: Optional[str] = None, grade: Optional[str] = None, exam_type: Optional[str] = None, skip: int = 0, limit: int = 50, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取试卷列表""" query = db.query(Paper).filter(Paper.user_id == current_user.id) if subject: query = query.filter(Paper.subject == subject) if grade: query = query.filter(Paper.grade == grade) if exam_type: query = query.filter(Paper.exam_type == exam_type) papers = query.offset(skip).limit(limit).all() return [PaperResponse( id=paper.id, title=paper.title, subject=paper.subject, grade=paper.grade, exam_type=paper.exam_type, exam_date=paper.exam_date, total_score=paper.total_score, achieved_score=paper.achieved_score, file_key=paper.file_key, pdf_key=paper.pdf_key, thumb_key=paper.thumb_key, tags=paper.tags, created_at=paper.created_at ) for paper in papers] @app.get("/api/papers/{paper_id}", response_model=PaperResponse) async def get_paper( paper_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取单个试卷详情""" paper = db.query(Paper).filter( Paper.id == paper_id, Paper.user_id == current_user.id ).first() if not paper: raise HTTPException(status_code=404, detail="Paper not found") return PaperResponse( id=paper.id, title=paper.title, subject=paper.subject, grade=paper.grade, exam_type=paper.exam_type, exam_date=paper.exam_date, total_score=paper.total_score, achieved_score=paper.achieved_score, file_key=paper.file_key, pdf_key=paper.pdf_key, thumb_key=paper.thumb_key, tags=paper.tags, created_at=paper.created_at ) # 错题管理 @app.post("/api/mistakes", response_model=MistakeResponse) async def create_mistake( mistake: MistakeCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """创建错题记录""" # 验证试卷是否属于当前用户 paper = db.query(Paper).filter( Paper.id == mistake.paper_id, Paper.user_id == current_user.id ).first() if not paper: raise HTTPException(status_code=404, detail="Paper not found") db_mistake = Mistake( user_id=current_user.id, paper_id=mistake.paper_id, page=mistake.page, bbox=mistake.bbox, subject=mistake.subject, knowledge_points=mistake.knowledge_points, reason_tag=mistake.reason_tag, difficulty=mistake.difficulty, answer=mistake.answer, analysis=mistake.analysis, notes=mistake.notes ) db.add(db_mistake) db.commit() db.refresh(db_mistake) return MistakeResponse( id=db_mistake.id, paper_id=db_mistake.paper_id, page=db_mistake.page, bbox=db_mistake.bbox, image_key=db_mistake.image_key, subject=db_mistake.subject, knowledge_points=db_mistake.knowledge_points, reason_tag=db_mistake.reason_tag, difficulty=db_mistake.difficulty, review_level=db_mistake.review_level, next_review_date=db_mistake.next_review_date, last_score=db_mistake.last_score, answer=db_mistake.answer, analysis=db_mistake.analysis, notes=db_mistake.notes, created_at=db_mistake.created_at ) @app.get("/api/mistakes", response_model=List[MistakeResponse]) async def get_mistakes( subject: Optional[str] = None, due: Optional[str] = None, # today, overdue, upcoming skip: int = 0, limit: int = 50, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取错题列表""" query = db.query(Mistake).filter(Mistake.user_id == current_user.id) if subject: query = query.filter(Mistake.subject == subject) if due == "today": today = datetime.utcnow().date() query = query.filter(Mistake.next_review_date <= datetime.combine(today, datetime.min.time()) + timedelta(days=1)) elif due == "overdue": query = query.filter(Mistake.next_review_date < datetime.utcnow()) mistakes = query.offset(skip).limit(limit).all() return [MistakeResponse( id=mistake.id, paper_id=mistake.paper_id, page=mistake.page, bbox=mistake.bbox, image_key=mistake.image_key, subject=mistake.subject, knowledge_points=mistake.knowledge_points, reason_tag=mistake.reason_tag, difficulty=mistake.difficulty, review_level=mistake.review_level, next_review_date=mistake.next_review_date, last_score=mistake.last_score, answer=mistake.answer, analysis=mistake.analysis, notes=mistake.notes, created_at=mistake.created_at ) for mistake in mistakes] @app.post("/api/mistakes/{mistake_id}/review") async def submit_review( mistake_id: int, review: ReviewSubmit, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """提交错题复习结果""" mistake = db.query(Mistake).filter( Mistake.id == mistake_id, Mistake.user_id == current_user.id ).first() if not mistake: raise HTTPException(status_code=404, detail="Mistake not found") # 计算下次复习时间 (Leitner算法) def calculate_next_review(level: int, rating: int, today: date) -> tuple: buckets = {1: 1, 2: 2, 3: 4, 4: 7, 5: 14} if rating <= 2: return today + timedelta(days=1), 1 new_level = min(5, level + (1 if rating >= 4 else 0)) return today + timedelta(days=buckets[new_level]), new_level today = datetime.utcnow().date() next_date, new_level = calculate_next_review(mistake.review_level, review.rating, today) # 更新错题记录 mistake.last_score = review.rating mistake.review_level = new_level mistake.next_review_date = datetime.combine(next_date, datetime.min.time()) # 创建复习记录 review_log = ReviewLog( user_id=current_user.id, mistake_id=mistake_id, rating=review.rating, next_review_date=mistake.next_review_date, comment=review.comment, time_spent=review.time_spent ) db.add(review_log) db.commit() return {"message": "Review submitted successfully", "next_review_date": mistake.next_review_date} @app.get("/api/dashboard", response_model=DashboardStats) async def get_dashboard_stats( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取仪表盘统计数据""" # 试卷总数 papers_count = db.query(Paper).filter(Paper.user_id == current_user.id).count() # 错题总数 mistakes_count = db.query(Mistake).filter(Mistake.user_id == current_user.id).count() # 今日待复习错题数 today = datetime.utcnow().date() due_reviews = db.query(Mistake).filter( Mistake.user_id == current_user.id, Mistake.next_review_date <= datetime.combine(today, datetime.min.time()) + timedelta(days=1) ).count() # 薄弱学科分析 from sqlalchemy import func weak_subjects = db.query( Mistake.subject, func.count(Mistake.id).label('count'), func.avg(Mistake.last_score).label('avg_score') ).filter( Mistake.user_id == current_user.id ).group_by(Mistake.subject).all() weak_subjects_data = [ {"subject": ws.subject, "mistake_count": ws.count, "avg_score": ws.avg_score or 0} for ws in weak_subjects ] # 最近准确率 recent_reviews = db.query(ReviewLog).filter( ReviewLog.user_id == current_user.id, ReviewLog.reviewed_at >= datetime.utcnow() - timedelta(days=7) ).all() recent_accuracy = 0.0 if recent_reviews: good_reviews = sum(1 for r in recent_reviews if r.rating >= 3) recent_accuracy = (good_reviews / len(recent_reviews)) * 100 # 按学科统计试卷、错题和待复习数量 subject_stats = {} # 统计各学科试卷数量 papers_by_subject = db.query( Paper.subject, func.count(Paper.id).label('count') ).filter( Paper.user_id == current_user.id ).group_by(Paper.subject).all() # 统计各学科错题数量 mistakes_by_subject = db.query( Mistake.subject, func.count(Mistake.id).label('count') ).filter( Mistake.user_id == current_user.id ).group_by(Mistake.subject).all() # 统计各学科待复习数量 due_by_subject = db.query( Mistake.subject, func.count(Mistake.id).label('count') ).filter( Mistake.user_id == current_user.id, Mistake.next_review_date <= datetime.combine(today, datetime.min.time()) + timedelta(days=1) ).group_by(Mistake.subject).all() # 初始化所有学科数据 all_subjects = set() for item in papers_by_subject: all_subjects.add(item.subject) for item in mistakes_by_subject: all_subjects.add(item.subject) for item in due_by_subject: all_subjects.add(item.subject) # 构建学科统计字典 for subject in all_subjects: subject_stats[subject] = { 'papers': 0, 'mistakes': 0, 'due': 0 } # 填充试卷数据 for item in papers_by_subject: subject_stats[item.subject]['papers'] = item.count # 填充错题数据 for item in mistakes_by_subject: subject_stats[item.subject]['mistakes'] = item.count # 填充待复习数据 for item in due_by_subject: subject_stats[item.subject]['due'] = item.count return DashboardStats( papers_count=papers_count, mistakes_count=mistakes_count, due_reviews=due_reviews, weak_subjects=weak_subjects_data, recent_accuracy=recent_accuracy, subject_stats=subject_stats ) # 题目管理 @app.post("/api/questions", response_model=QuestionResponse) async def create_question( question: QuestionCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """创建题目""" db_question = Question( user_id=current_user.id, paper_id=question.paper_id, question_type=question.question_type, subject=question.subject, grade=question.grade, chapter=question.chapter, knowledge_points=question.knowledge_points, difficulty=question.difficulty, question_content=question.question_content, options=question.options, correct_answer=question.correct_answer, explanation=question.explanation, tags=question.tags ) db.add(db_question) db.commit() db.refresh(db_question) return QuestionResponse( id=db_question.id, paper_id=db_question.paper_id, question_type=db_question.question_type, subject=db_question.subject, grade=db_question.grade, chapter=db_question.chapter, knowledge_points=db_question.knowledge_points, difficulty=db_question.difficulty, question_content=db_question.question_content, options=db_question.options, correct_answer=db_question.correct_answer, explanation=db_question.explanation, image_key=db_question.image_key, tags=db_question.tags, created_at=db_question.created_at ) @app.get("/api/questions", response_model=List[QuestionResponse]) async def get_questions( paper_id: Optional[int] = None, subject: Optional[str] = None, grade: Optional[str] = None, chapter: Optional[str] = None, question_type: Optional[str] = None, difficulty: Optional[int] = None, search: Optional[str] = None, skip: int = 0, limit: int = 50, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取题目列表""" query = db.query(Question).filter(Question.user_id == current_user.id) if paper_id: query = query.filter(Question.paper_id == paper_id) if subject: query = query.filter(Question.subject == subject) if grade: query = query.filter(Question.grade == grade) if chapter: query = query.filter(Question.chapter == chapter) if question_type: query = query.filter(Question.question_type == question_type) if difficulty: query = query.filter(Question.difficulty == difficulty) if search: query = query.filter(Question.question_content.contains(search)) questions = query.offset(skip).limit(limit).all() return [QuestionResponse( id=q.id, paper_id=q.paper_id, question_type=q.question_type, subject=q.subject, grade=q.grade, chapter=q.chapter, knowledge_points=q.knowledge_points, difficulty=q.difficulty, question_content=q.question_content, options=q.options, correct_answer=q.correct_answer, explanation=q.explanation, image_key=q.image_key, images=q.images, tags=q.tags, created_at=q.created_at ) for q in questions] @app.get("/api/questions/{question_id}", response_model=QuestionResponse) async def get_question( question_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取单个题目详情""" question = db.query(Question).filter( Question.id == question_id, Question.user_id == current_user.id ).first() if not question: raise HTTPException(status_code=404, detail="Question not found") return QuestionResponse( id=question.id, paper_id=question.paper_id, question_type=question.question_type, subject=question.subject, grade=question.grade, chapter=question.chapter, knowledge_points=question.knowledge_points, difficulty=question.difficulty, question_content=question.question_content, options=question.options, correct_answer=question.correct_answer, explanation=question.explanation, image_key=question.image_key, images=question.images, tags=question.tags, created_at=question.created_at ) @app.delete("/api/questions/{question_id}") async def delete_question( question_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """删除题目""" question = db.query(Question).filter( Question.id == question_id, Question.user_id == current_user.id ).first() if not question: raise HTTPException(status_code=404, detail="Question not found") db.delete(question) db.commit() return {"message": "Question deleted successfully"} @app.get("/api/papers/{paper_id}/file") @app.options("/api/papers/{paper_id}/file") async def get_paper_file( request: Request, paper_id: int, access_token: Optional[str] = None, credentials: Optional[HTTPAuthorizationCredentials] = Depends(HTTPBearer(auto_error=False)), db: Session = Depends(get_db) ): """获取试卷文件 - 支持token查询参数或Authorization头""" # 处理OPTIONS预检请求 if request.method == "OPTIONS": from fastapi.responses import Response return Response( headers={ "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "GET, OPTIONS", "Access-Control-Allow-Headers": "Authorization" } ) # 尝试从查询参数获取token token = access_token # 如果没有查询参数token，尝试从Authorization头获取 if not token and credentials: token = credentials.credentials if not token: raise HTTPException(status_code=401, detail="No authentication token provided") # 验证token并获取用户 try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise HTTPException(status_code=401, detail="Invalid token") except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Invalid token") user = db.query(User).filter(User.username == username).first() if user is None: raise HTTPException(status_code=401, detail="User not found") # 获取试卷 paper = db.query(Paper).filter( Paper.id == paper_id, Paper.user_id == user.id ).first() if not paper: raise HTTPException(status_code=404, detail="Paper not found") from fastapi.responses import FileResponse import os if os.path.exists(paper.file_key): # 根据文件扩展名设置content type content_type = "application/octet-stream" filename = os.path.basename(paper.file_key) if filename.lower().endswith('.pdf'): content_type = "application/pdf" elif filename.lower().endswith(('.jpg', '.jpeg')): content_type = "image/jpeg" elif filename.lower().endswith('.png'): content_type = "image/png" # 处理中文文件名编码 import urllib.parse encoded_filename = urllib.parse.quote(filename.encode('utf-8')) return FileResponse( paper.file_key, media_type=content_type, headers={ "Content-Disposition": f"inline; filename*=UTF-8''{encoded_filename}", "Access-Control-Allow-Origin": "*" } ) else: raise HTTPException(status_code=404, detail="File not found") @app.get("/api/files/paper/{paper_id}") async def get_paper_file_public( paper_id: int, access_token: str, db: Session = Depends(get_db) ): """公开访问试卷文件（通过token验证）""" # 验证token try: payload = jwt.decode(access_token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise HTTPException(status_code=401, detail="Invalid token") except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Invalid token") # 获取用户 user = db.query(User).filter(User.username == username).first() if user is None: raise HTTPException(status_code=401, detail="User not found") # 获取试卷 paper = db.query(Paper).filter( Paper.id == paper_id, Paper.user_id == user.id ).first() if not paper: raise HTTPException(status_code=404, detail="Paper not found") from fastapi.responses import FileResponse import os if os.path.exists(paper.file_key): # 根据文件扩展名设置content type content_type = "application/octet-stream" filename = os.path.basename(paper.file_key) if filename.lower().endswith('.pdf'): content_type = "application/pdf" elif filename.lower().endswith(('.jpg', '.jpeg')): content_type = "image/jpeg" elif filename.lower().endswith('.png'): content_type = "image/png" return FileResponse( paper.file_key, media_type=content_type, headers={ "Content-Disposition": f"inline; filename={filename}", "Cache-Control": "private, max-age=3600" } ) else: raise HTTPException(status_code=404, detail="File not found") # 静态文件服务 - 用于直接访问上传的文件 @app.get("/uploads/{user_id}/{filename}") async def serve_uploaded_file( user_id: int, filename: str, current_user: User = Depends(get_current_user) ): """直接访问上传的文件""" # 只允许用户访问自己的文件 if current_user.id != user_id: raise HTTPException(status_code=403, detail="Access denied") file_path = f"uploads/{user_id}/{filename}" from fastapi.responses import FileResponse import os if os.path.exists(file_path): # 根据文件扩展名设置content type content_type = "application/octet-stream" if filename.lower().endswith('.pdf'): content_type = "application/pdf" elif filename.lower().endswith(('.jpg', '.jpeg')): content_type = "image/jpeg" elif filename.lower().endswith('.png'): content_type = "image/png" return FileResponse( file_path, media_type=content_type, headers={"Content-Disposition": f"inline; filename={filename}"} ) else: raise HTTPException(status_code=404, detail="File not found") # PDF拆题API - 使用 Claude Sonnet 4.5 智能解析 @app.post("/api/papers/{paper_id}/split", response_model=SplitResponse) async def split_paper( paper_id: int, use_claude: bool = True, # 默认使用 Claude 解析 current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """ 拆解试卷PDF，自动提取题目 Args: paper_id: 试卷ID use_claude: 是否使用 Claude Sonnet 4.5 智能解析（默认 True） Returns: 解析结果，包含题目列表和统计信息 """ # 检查试卷是否存在 paper = db.query(Paper).filter( Paper.id == paper_id, Paper.user_id == current_user.id ).first() if not paper: raise HTTPException(status_code=404, detail="试卷不存在") # 检查文件是否存在 file_path = paper.file_key if not file_path or not os.path.exists(file_path): raise HTTPException(status_code=404, detail="试卷文件不存在") try: if use_claude: # 使用 Claude Sonnet 4.5 智能解析器 logger.info(f"使用 Claude Sonnet 4.5 解析试卷: {file_path}") parser = ClaudeExamParser() result = parser.parse_exam_file( file_path, subject=paper.subject, grade=paper.grade ) if not result['success']: logger.error(f"Claude 解析失败: {result.get('error')}") return SplitResponse( success=False, paper_id=paper_id, error=result.get('error', 'Claude 解析失败') ) # 转换 Claude 的输出格式为系统格式 claude_result = result['result'] questions_data = [] for q in claude_result.get('questions', []): # 映射题型 type_mapping = { 'choice': 'choice', 'fill_blank': 'fill_blank', 'calculation': 'calculation', 'essay': 'essay' } question_data = { 'number': q.get('number'), 'type': type_mapping.get(q.get('type', 'choice'), 'choice'), 'question': q.get('question', ''), 'options': q.get('options', []), 'answer': q.get('answer', ''), 'analysis': q.get('analysis', ''), 'difficulty': q.get('difficulty', 3), 'knowledge_points': q.get('knowledge_points', []), 'page': q.get('page', 1), 'has_image': q.get('has_image', False), 'image_description': q.get('image_description', ''), 'formulas': q.get('formulas', []), 'confidence': 0.95 # Claude 高置信度 } questions_data.append(question_data) # 更新试卷学科（如果 AI 识别出来了） if claude_result.get('subject') and not paper.subject: paper.subject = claude_result['subject'] db.commit() stats = claude_result.get('stats', {}) logger.info(f"Claude 解析成功，共 {len(questions_data)} 道题目") return SplitResponse( success=True, paper_id=paper_id, questions=questions_data, stats={ 'total_questions': claude_result.get('total_questions', len(questions_data)), 'choice_count': stats.get('choice_count', 0), 'fill_blank_count': stats.get('fill_blank_count', 0), 'calculation_count': stats.get('calculation_count', 0), 'essay_count': stats.get('essay_count', 0), 'parser_version': 'claude-sonnet-4.5' } ) else: # 使用传统解析器 logger.info(f"使用传统解析器: {file_path}") parser = PDFExamParserEnhancedV2() # 创建临时目录（如果需要的话） temp_dir = tempfile.mkdtemp(prefix="pdf_split_") try: # 处理PDF logger.info(f"开始处理PDF: {file_path}") result = parser.process_pdf(file_path) logger.info(f"PDF处理结果: {result}") if not result['success']: logger.error(f"PDF解析失败: {result.get('error')}") return SplitResponse( success=False, paper_id=paper_id, total_questions=0, questions=[], stats={}, error=result.get('error', '未知错误') ) questions = result['questions'] stats = result['stats'] logger.info(f"解析到题目数量: {len(questions)}") # 提取图片并与题目关联 image_extractor = PDFImageExtractor() extracted_images = image_extractor.extract_images_from_pdf(file_path) logger.info(f"提取到图片数量: {len(extracted_images)}") # 将图片关联到题目（基于页面位置） for img_info in extracted_images: img_page = img_info.get('page', 1) logger.info(f"处理图片: 页面{img_page}, 路径{img_info.get('path', 'unknown')}") # 找到同一页面的题目，将图片关联到最近的题目 for i, question in enumerate(questions): q_page = question.get('page', 1) logger.info(f"检查题目{question.get('number')}: 页面{q_page}") if q_page == img_page: if 'images' not in question: question['images'] = [] question['images'].append(img_info) logger.info(f"✅ 图片关联到题目{question.get('number')} (页面{img_page})") break # 每张图片只关联一个题目 else: logger.warning(f"⚠️ 无法关联图片到页面{img_page}的题目") # 清理该试卷的现有题目，避免重复累加 existing_questions = db.query(Question).filter(Question.paper_id == paper_id).all() if existing_questions: logger.info(f"清理现有题目：删除 {len(existing_questions)} 道旧题目") for q in existing_questions: db.delete(q) db.commit() # 保存提取的题目到数据库 saved_questions = [] for i, question_data in enumerate(questions): # 从question_data中获取已关联的图片信息 question_images = question_data.get('images', []) # 如果没有图片但有image_path，移动图片到uploads目录 image_key = None if not question_images and question_data.get('image_path') and os.path.exists(question_data['image_path']): # 生成新的文件名 file_extension = os.path.splitext(question_data['image_filename'])[1] new_filename = f"{uuid.uuid4()}{file_extension}" dest_dir = os.path.join("uploads", str(current_user.id)) os.makedirs(dest_dir, exist_ok=True) dest_path = os.path.join(dest_dir, new_filename) try: shutil.copy2(question_data['image_path'], dest_path) image_key = dest_path # 创建图片对象 question_images = [{ 'page': question_data.get('page', 1), 'bbox': question_data.get('bbox', []), 'key': os.path.basename(dest_path) # 只保存文件名 }] except Exception as e: print(f"移动题目图片失败: {e}") # 创建题目记录 db_question = Question( user_id=current_user.id, paper_id=paper_id, question_type=question_data.get('type', 'unknown'), subject=paper.subject, grade=paper.grade, knowledge_points=[], # 后续可以通过AI分析填充 difficulty=3, # 默认难度 question_content=question_data.get('text', ''), options=question_data.get('options'), correct_answer='', # 需要手动填写或AI识别 explanation=None, image_key=image_key, # 拆题相关字段 page_spans=question_data.get('page_spans', [{'page': question_data.get('page', 1), 'bbox': question_data.get('bbox', [])}]), source_text=question_data.get('text', ''), images=question_images, # 使用从PDFSplitterHybrid获取的图片信息 formula=[], # 后续可以添加公式识别 score=None, # 需要手动填写 question_number=str(question_data.get('number', i+1)), auto_extracted=True ) db.add(db_question) # 调试信息 q_number = str(question_data.get('number', i+1)) logger.info(f"保存题目: number='{q_number}', type='{question_data.get('type')}'") # 构建返回数据 saved_questions.append(SplitQuestionResult( number=q_number, type=question_data.get('type', 'unknown'), text=question_data.get('text', ''), options=question_data.get('options'), bbox=question_data.get('bbox', []), page=question_data.get('page', 1), confidence=question_data.get('confidence', 0.0), image_filename=new_filename if image_key else None )) # 提交数据库更改 db.commit() return SplitResponse( success=True, paper_id=paper_id, total_questions=len(saved_questions), questions=saved_questions, stats=stats ) finally: # 清理临时目录 if os.path.exists(temp_dir): shutil.rmtree(temp_dir) except Exception as e: db.rollback() return SplitResponse( success=False, paper_id=paper_id, total_questions=0, questions=[], stats={}, error=f"处理失败: {str(e)}" ) # 新增API端点：导出格式化的题目数据 @app.get("/api/papers/{paper_id}/export/json") async def export_paper_questions_json( paper_id: int, current_user: User = Depends(get_current_user), include_validation: bool = True, include_statistics: bool = True ): """ 导出试卷的格式化JSON数据 Args: paper_id: 试卷ID include_validation: 是否包含验证信息 include_statistics: 是否包含统计信息 Returns: 格式化的JSON数据 """ try: # 获取试卷信息 paper = db.query(Paper).filter( Paper.id == paper_id, Paper.user_id == current_user.id ).first() if not paper: raise HTTPException(status_code=404, detail="试卷不存在") # 获取题目列表 questions = db.query(Question).filter( Question.paper_id == paper_id, Question.user_id == current_user.id ).all() if not questions: return { "success": True, "data": { "paper_info": { "id": paper.id, "title": paper.title, "subject": paper.subject, "grade": paper.grade, "exam_type": paper.exam_type, "exam_date": paper.exam_date.isoformat() if paper.exam_date else None, "total_score": paper.total_score, "created_at": paper.created_at.isoformat() }, "questions": [], "statistics": { "total_questions": 0, "message": "该试卷暂无题目数据" } } } # 转换为字典格式 questions_data = [] for q in questions: question_dict = { "id": q.id, "question_type": q.question_type, "subject": q.subject, "grade": q.grade, "chapter": q.chapter, "difficulty": q.difficulty, "question_content": q.question_content, "options": q.options, "correct_answer": q.correct_answer, "explanation": q.explanation, "image_key": q.image_key, "knowledge_points": q.knowledge_points, "score": q.score, "question_number": q.question_number, "auto_extracted": q.auto_extracted, "images": q.images, "page_spans": q.page_spans, "source_text": q.source_text, "created_at": q.created_at, "updated_at": q.updated_at } questions_data.append(question_dict) # 使用格式化器 formatter = QuestionFormatter() # 构建完整数据结构 raw_data = { "paper_id": paper.id, "title": paper.title, "subject": paper.subject, "grade": paper.grade, "questions": questions_data } # 格式化数据 formatted_data = formatter.format_paper_questions(raw_data) # 移除不需要的验证和统计信息 if not include_validation: for question in formatted_data["questions"]: question.pop("validation", None) if not include_statistics: formatted_data.pop("statistics", None) return { "success": True, "data": formatted_data } except HTTPException: raise except Exception as e: logger.error(f"导出JSON失败: {e}") raise HTTPException(status_code=500, detail=f"导出失败: {str(e)}") # 新增API端点：获取格式化的单个题目 @app.get("/api/questions/{question_id}/format") async def get_formatted_question( question_id: int, current_user: User = Depends(get_current_user) ): """获取格式化的单个题目数据""" try: # 获取题目 question = db.query(Question).filter( Question.id == question_id, Question.user_id == current_user.id ).first() if not question: raise HTTPException(status_code=404, detail="题目不存在") # 获取关联的试卷信息 paper = None if question.paper_id: paper = db.query(Paper).filter( Paper.id == question.paper_id ).first() paper_info = None if paper: paper_info = { "id": paper.id, "title": paper.title, "subject": paper.subject, "grade": paper.grade } # 转换为字典格式 question_dict = { "id": question.id, "question_type": question.question_type, "subject": question.subject, "grade": question.grade, "chapter": question.chapter, "difficulty": question.difficulty, "question_content": question.question_content, "options": question.options, "correct_answer": question.correct_answer, "explanation": question.explanation, "image_key": question.image_key, "knowledge_points": question.knowledge_points, "score": question.score, "question_number": question.question_number, "auto_extracted": question.auto_extracted, "images": question.images, "page_spans": question.page_spans, "source_text": question.source_text, "created_at": question.created_at, "updated_at": question.updated_at, "view_count": 0, # 可以添加访问统计 "attempt_count": 0, "correct_count": 0 } # 格式化 formatter = QuestionFormatter() formatted_question = formatter.format_question(question_dict, paper_info) return { "success": True, "data": formatted_question } except HTTPException: raise except Exception as e: logger.error(f"获取格式化题目失败: {e}") raise HTTPException(status_code=500, detail=f"获取失败: {str(e)}") # 新增API端点：批量导出题目数据 @app.post("/api/questions/batch/export") async def batch_export_questions( request: dict, current_user: User = Depends(get_current_user) ): """ 批量导出题目数据 Request body: { "question_ids": [1, 2, 3], "format": "json", "include_validation": true, "include_statistics": true } """ try: question_ids = request.get("question_ids", []) include_validation = request.get("include_validation", True) include_statistics = request.get("include_statistics", True) if not question_ids: raise HTTPException(status_code=400, detail="请提供题目ID列表") # 获取题目列表 questions = db.query(Question).filter( Question.id.in_(question_ids), Question.user_id == current_user.id ).all() if not questions: return { "success": True, "data": { "questions": [], "statistics": {"total_questions": 0} } } # 转换为字典格式 questions_data = [] for q in questions: question_dict = { "id": q.id, "question_type": q.question_type, "subject": q.subject, "grade": q.grade, "chapter": q.chapter, "difficulty": q.difficulty, "question_content": q.question_content, "options": q.options, "correct_answer": q.correct_answer, "explanation": q.explanation, "image_key": q.image_key, "knowledge_points": q.knowledge_points, "score": q.score, "question_number": q.question_number, "auto_extracted": q.auto_extracted, "images": q.images, "page_spans": q.page_spans, "source_text": q.source_text, "created_at": q.created_at, "updated_at": q.updated_at } questions_data.append(question_dict) # 使用格式化器 formatter = QuestionFormatter() formatted_questions = [] for question_dict in questions_data: formatted = formatter.format_question(question_dict) if not include_validation: formatted.pop("validation", None) formatted_questions.append(formatted) # 计算统计信息 stats = {} if include_statistics: raw_data = { "paper_id": None, "title": "批量导出", "subject": "混合", "grade": "", "questions": questions_data } stats = formatter.format_paper_questions(raw_data)["statistics"] return { "success": True, "data": { "questions": formatted_questions, "statistics": stats, "export_info": { "exported_at": datetime.now().isoformat(), "total_questions": len(formatted_questions), "requested_ids": question_ids, "found_ids": [q.id for q in questions] } } } except HTTPException: raise except Exception as e: logger.error(f"批量导出失败: {e}") raise HTTPException(status_code=500, detail=f"批量导出失败: {str(e)}") # 挂载静态文件服务 app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads") app.mount("/api/uploads", StaticFiles(directory="uploads"), name="api-uploads") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8888)