From 96a1a259f038df8c6fdef74b77e18e1957cb4edb Mon Sep 17 00:00:00 2001 From: Daniel Date: Wed, 1 Apr 2026 01:55:24 +0200 Subject: [PATCH] Async extraction with live progress + chunked large PDFs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extraction is now fully async via Celery — UI shows a live progress panel, job continues even if page is closed. Large documents are processed in 50-page chunks to extract all questions (not just first ~50 pages). Backend: - app/tasks/quiz_tasks.py: new Celery task 'extract_quiz' - Writes step-by-step progress to Redis (extraction:steps:{job_id}) - Splits large page ranges into 50-page chunks, processes each separately - Reports per-chunk results and running total - Falls back to synchronous if Celery/Redis unavailable - POST /quizzes/ now returns {job_id, status:"pending"} immediately - GET /quizzes/job/{job_id} polls progress: steps[], status, quiz_id on completion - Celery task list updated to include quiz_tasks Frontend (DocumentDetailPage): - ExtractionProgress modal component: monospace step log, auto-scrolls, spinner - Polls job status every 2 seconds via /quizzes/job/{job_id} - "Open Quiz →" button appears when done - "✕ closes — job continues in background" shown while running - beforeunload warning when job is active (preventing accidental close) Co-Authored-By: Claude Sonnet 4.6 (1M context) --- backend/app/routers/quizzes.py | 56 ++++-- backend/app/tasks/__init__.py | 2 +- backend/app/tasks/quiz_tasks.py | 209 ++++++++++++++++++++++ frontend/src/pages/DocumentDetailPage.jsx | 123 ++++++++++++- 4 files changed, 376 insertions(+), 14 deletions(-) create mode 100644 backend/app/tasks/quiz_tasks.py diff --git a/backend/app/routers/quizzes.py b/backend/app/routers/quizzes.py index 4787703..96fea01 100644 --- a/backend/app/routers/quizzes.py +++ b/backend/app/routers/quizzes.py @@ -17,25 +17,33 @@ from app.utils.quiz_questions import get_quiz_questions, question_in_quiz, remov router = APIRouter() -@router.post("/", response_model=QuizResponse) +@router.post("/") def create_quiz( quiz_data: QuizCreate, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): - """Create quiz by extracting questions from PDF section. Moderator/Admin only.""" + """Start async quiz extraction. Returns {job_id} immediately; poll /quizzes/job/{job_id} for progress.""" + import uuid section = db.query(Section).filter(Section.id == quiz_data.section_id).first() if not section: raise HTTPException(status_code=404, detail="Section not found") if not current_user.is_admin and section.document.user_id != current_user.id: raise HTTPException(status_code=403, detail="Not your document") - if quiz_data.mode not in ("timed", "learning"): raise HTTPException(status_code=400, detail="Mode must be 'timed' or 'learning'") + job_id = str(uuid.uuid4()) + try: - quiz = quiz_service.create_quiz_from_section( - db=db, + from app.tasks.quiz_tasks import extract_quiz + import redis as redis_lib + from app.config import settings + r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) + r.set(f"extraction:status:{job_id}", "pending", ex=3600) + + extract_quiz.delay( + job_id=job_id, user_id=current_user.id, section_id=quiz_data.section_id, title=quiz_data.title, @@ -44,11 +52,39 @@ def create_quiz( model_id=quiz_data.model_id, question_category_id=quiz_data.question_category_id, ) - return quiz - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - except RuntimeError as e: - raise HTTPException(status_code=500, detail=str(e)) + except Exception: + # Celery/Redis unavailable — fall back to synchronous extraction + try: + quiz = quiz_service.create_quiz_from_section( + db=db, user_id=current_user.id, + section_id=quiz_data.section_id, title=quiz_data.title, + mode=quiz_data.mode, time_limit_minutes=quiz_data.time_limit_minutes, + model_id=quiz_data.model_id, question_category_id=quiz_data.question_category_id, + ) + return {"job_id": job_id, "status": "completed", "quiz_id": quiz.id} + except (ValueError, RuntimeError) as e: + raise HTTPException(status_code=400, detail=str(e)) + + return {"job_id": job_id, "status": "pending"} + + +@router.get("/job/{job_id}") +def get_extraction_job(job_id: str, current_user: User = Depends(require_moderator)): + """Poll extraction job progress. Returns steps list, status, and quiz_id when done.""" + import json as _json + import redis as redis_lib + from app.config import settings + + r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) + status = r.get(f"extraction:status:{job_id}") or "unknown" + raw_steps = r.lrange(f"extraction:steps:{job_id}", 0, -1) + steps = [_json.loads(s) for s in raw_steps] + result = {"job_id": job_id, "status": status, "steps": steps} + if status == "completed": + result["quiz_id"] = int(r.get(f"extraction:quiz_id:{job_id}") or 0) + if status == "failed": + result["error"] = r.get(f"extraction:error:{job_id}") or "Unknown error" + return result @router.get("/search") diff --git a/backend/app/tasks/__init__.py b/backend/app/tasks/__init__.py index 433c39d..846a808 100644 --- a/backend/app/tasks/__init__.py +++ b/backend/app/tasks/__init__.py @@ -6,7 +6,7 @@ celery_app = Celery( "quiz_tasks", broker=settings.REDIS_URL, backend=settings.REDIS_URL, - include=["app.tasks.pdf_tasks"], + include=["app.tasks.pdf_tasks", "app.tasks.quiz_tasks"], ) celery_app.conf.task_serializer = "json" celery_app.conf.result_serializer = "json" diff --git a/backend/app/tasks/quiz_tasks.py b/backend/app/tasks/quiz_tasks.py new file mode 100644 index 0000000..e2e291b --- /dev/null +++ b/backend/app/tasks/quiz_tasks.py @@ -0,0 +1,209 @@ +"""Async quiz extraction task with step-by-step progress reporting via Redis.""" +import json +import logging +import time + +from app.tasks import celery_app +from app.database import SessionLocal + +logger = logging.getLogger(__name__) + +EXPIRE_SECONDS = 3600 # keep progress for 1 hour + + +def _redis(): + import redis + from app.config import settings + return redis.from_url(settings.REDIS_URL, decode_responses=True) + + +def _push_step(r, job_id: str, step: str, message: str): + key = f"extraction:steps:{job_id}" + entry = json.dumps({"step": step, "message": message, "ts": time.time()}) + r.rpush(key, entry) + r.expire(key, EXPIRE_SECONDS) + + +@celery_app.task(name="extract_quiz", bind=True) +def extract_quiz( + self, + job_id: str, + user_id: int, + section_id: int, + title: str, + mode: str, + time_limit_minutes: int | None, + model_id: str | None, + question_category_id: int | None, +): + """Background quiz extraction task with live progress reporting.""" + r = _redis() + r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS) + + db = SessionLocal() + try: + from app.models.section import Section + from app.models.pdf_document import PDFDocument + from app.services import ai_service, vector_service, pdf_service, embedding_service + from app.models.quiz import Quiz + from app.models.question import Question + from app.models.quiz_question_link import QuizQuestionLink + from app.config import settings + import os + + _push_step(r, job_id, "start", "Starting extraction…") + + section = db.query(Section).filter(Section.id == section_id).first() + if not section: + raise ValueError("Section not found") + + document = db.query(PDFDocument).filter(PDFDocument.id == section.document_id).first() + if not document: + raise ValueError("Document not found") + + total_pages = section.end_page - section.start_page + 1 + _push_step(r, job_id, "text", f"Loading text from pages {section.start_page}–{section.end_page} ({total_pages} pages)…") + + # Determine model first (used for chunk logging) + if model_id: + from app.models.ai_model_config import AIModelConfig + config = db.query(AIModelConfig).filter(AIModelConfig.model_id == model_id).first() + api_key = config.api_key if config and config.api_key else None + model_name = config.name if config else model_id + else: + model_id, api_key = ai_service.get_model_for_task(db, "extraction") + model_name = model_id + + # For large page ranges, process in chunks of 50 pages to avoid truncation + CHUNK_PAGES = 50 + import json as _json + + if total_pages <= CHUNK_PAGES: + chunks = [(section.start_page, section.end_page)] + else: + chunks = [] + p = section.start_page + while p <= section.end_page: + end = min(p + CHUNK_PAGES - 1, section.end_page) + chunks.append((p, end)) + p = end + 1 + + n_chunks = len(chunks) + if n_chunks > 1: + _push_step(r, job_id, "text", f"Large section: splitting into {n_chunks} chunks of up to {CHUNK_PAGES} pages each.") + + all_valid_questions = [] + all_skipped = [] + + for chunk_idx, (start_p, end_p) in enumerate(chunks, 1): + if n_chunks > 1: + _push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p} → {model_name}…") + else: + _push_step(r, job_id, "ai", f"Sending pages {start_p}–{end_p} to {model_name}…") + + chunk_content = vector_service.get_pages_text( + document_id=section.document_id, + start_page=start_p, + end_page=end_p, + ) + if not chunk_content: + _push_step(r, job_id, "ai", f" No text found for pages {start_p}–{end_p}, skipping.") + continue + + try: + chunk_data = ai_service.extract_questions( + chunk_content, + page_info=f"{start_p}-{end_p}", + page_ref=start_p, + model_id=model_id, + api_key=api_key, + ) + chunk_skipped = [] + if chunk_data and chunk_data[0].get("skipped"): + chunk_skipped = chunk_data[0].pop("skipped") + chunk_valid = [q for q in chunk_data if q.get("correct_answer")] + all_valid_questions.extend(chunk_valid) + all_skipped.extend(chunk_skipped) + _push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {len(chunk_valid)} questions extracted{f', {len(chunk_skipped)} skipped' if chunk_skipped else ''}. Total so far: {len(all_valid_questions)}.") + except Exception as e: + _push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}. Continuing…") + + valid_questions = all_valid_questions + skipped = all_skipped + + _push_step(r, job_id, "ai", f"Extraction complete: {len(valid_questions)} total valid questions{f', {len(skipped)} skipped' if skipped else ''}.") + + if not valid_questions: + raise ValueError("No valid questions extracted. The AI could not find questions with correct answers in this page range.") + + # Extract images + _push_step(r, job_id, "images", "Extracting question images…") + file_path = os.path.join(settings.UPLOAD_DIR, document.filename) + page_images = {} + if os.path.exists(file_path): + try: + page_images = pdf_service.extract_all_images( + file_path, document.id, section.start_page, section.end_page + ) + except Exception as e: + _push_step(r, job_id, "images", f"Image extraction skipped: {e}") + + # Create quiz + quiz = Quiz( + section_id=section_id, + user_id=user_id, + title=title, + questions_count=len(valid_questions), + mode=mode, + time_limit_minutes=time_limit_minutes, + skipped_questions=_json.dumps(skipped) if skipped else None, + ) + db.add(quiz) + db.flush() + + _push_step(r, job_id, "save", f"Saving {len(valid_questions)} questions and generating embeddings…") + + for pos, q in enumerate(valid_questions): + page_ref = q.get("page_reference") + image_path = None + if page_ref and page_ref in page_images and page_images[page_ref]: + image_path = page_images[page_ref].pop(0) + if not page_images[page_ref]: + del page_images[page_ref] + + question = Question( + source_quiz_id=quiz.id, + question_category_id=question_category_id, + question_text=q["question_text"], + question_type=q["question_type"], + options=q.get("options"), + correct_answer=q["correct_answer"], + explanation=q.get("explanation", ""), + page_reference=page_ref, + image_path=image_path, + ) + db.add(question) + db.flush() + db.add(QuizQuestionLink(quiz_id=quiz.id, question_id=question.id, position=pos)) + try: + embedding_service.embed_question(question) + except Exception: + pass + + db.commit() + db.refresh(quiz) + + _push_step(r, job_id, "done", f"Quiz ready! {len(valid_questions)} questions extracted and saved.") + + r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS) + r.set(f"extraction:quiz_id:{job_id}", str(quiz.id), ex=EXPIRE_SECONDS) + return quiz.id + + except Exception as e: + logger.exception(f"Quiz extraction failed for job {job_id}") + _push_step(r, job_id, "error", f"Extraction failed: {e}") + r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS) + r.set(f"extraction:error:{job_id}", str(e)[:500], ex=EXPIRE_SECONDS) + raise + finally: + db.close() diff --git a/frontend/src/pages/DocumentDetailPage.jsx b/frontend/src/pages/DocumentDetailPage.jsx index b916ef1..44acb78 100644 --- a/frontend/src/pages/DocumentDetailPage.jsx +++ b/frontend/src/pages/DocumentDetailPage.jsx @@ -1,8 +1,99 @@ -import { useState, useEffect } from 'react' +import { useState, useEffect, useRef, useCallback } from 'react' import { useParams, useNavigate, Link } from 'react-router-dom' import { useAuth } from '../context/AuthContext' import api from '../api/client' +function ExtractionProgress({ jobId, onDone, onClose }) { + const [steps, setSteps] = useState([]) + const [status, setStatus] = useState('pending') + const [quizId, setQuizId] = useState(null) + const [error, setError] = useState('') + const intervalRef = useRef(null) + const bottomRef = useRef(null) + + const poll = useCallback(async () => { + try { + const res = await api.get(`/quizzes/job/${jobId}`) + setSteps(res.data.steps || []) + setStatus(res.data.status) + if (res.data.status === 'completed') { + setQuizId(res.data.quiz_id) + clearInterval(intervalRef.current) + } + if (res.data.status === 'failed') { + setError(res.data.error || 'Extraction failed') + clearInterval(intervalRef.current) + } + } catch { } + }, [jobId]) + + useEffect(() => { + poll() + intervalRef.current = setInterval(poll, 2000) + return () => clearInterval(intervalRef.current) + }, [poll]) + + useEffect(() => { + bottomRef.current?.scrollIntoView({ behavior: 'smooth' }) + }, [steps.length]) + + const stepIcon = (step) => { + if (step === 'error') return '✗' + if (step === 'done') return '✓' + if (step === 'ai') return '🤖' + if (step === 'text') return '📄' + if (step === 'images') return '🖼' + if (step === 'save') return '💾' + return '→' + } + + return ( +
+
+
+

+ {status === 'completed' ? '✓ Extraction Complete' : status === 'failed' ? '✗ Extraction Failed' : '🤖 Extracting Questions…'} +

+ +
+ +
+ {steps.length === 0 &&
Starting…
} + {steps.map((s, i) => ( +
+ {stepIcon(s.step)} + {s.message} +
+ ))} + {(status === 'pending' || status === 'running') && ( +
+
+ Working… +
+ )} +
+
+ + {error &&
{error}
} + +
+ + {status === 'completed' || status === 'failed' ? '' : '✕ closes this panel — job continues in background'} + + {status === 'completed' && quizId && ( + + )} + {(status === 'failed' || status === 'completed') && ( + + )} +
+
+
+ ) +} + export default function DocumentDetailPage() { const { id } = useParams() const navigate = useNavigate() @@ -12,6 +103,7 @@ export default function DocumentDetailPage() { const [sectionForm, setSectionForm] = useState({ name: '', start_page: 1, end_page: 10 }) const [creating, setCreating] = useState(false) const [deletingSection, setDeletingSection] = useState(null) + const [activeJob, setActiveJob] = useState(null) // {jobId, sectionName} const [generating, setGenerating] = useState(null) const [quizTitle, setQuizTitle] = useState('') const [quizMode, setQuizMode] = useState('timed') @@ -85,9 +177,19 @@ export default function DocumentDetailPage() { model_id: selectedModelId || null, question_category_id: selectedQuestionCategoryId ? parseInt(selectedQuestionCategoryId) : null, }) - navigate(`/quizzes/${res.data.id}`) + // Async: show progress panel + if (res.data.job_id) { + setActiveJob({ jobId: res.data.job_id, sectionName }) + // If already completed (sync fallback), navigate directly + if (res.data.status === 'completed' && res.data.quiz_id) { + navigate(`/quizzes/${res.data.quiz_id}`) + } + } else if (res.data.id) { + navigate(`/quizzes/${res.data.id}`) + } } catch (err) { - setError(err.response?.data?.detail || 'Failed to generate quiz. Check AI model config.') + setError(err.response?.data?.detail || 'Failed to start extraction. Check AI model config.') + } finally { setGenerating(null) } } @@ -120,11 +222,26 @@ export default function DocumentDetailPage() { navigate('/') } + // Warn if user tries to close while a job is running + useEffect(() => { + if (!activeJob) return + const handler = (e) => { e.preventDefault(); e.returnValue = '' } + window.addEventListener('beforeunload', handler) + return () => window.removeEventListener('beforeunload', handler) + }, [activeJob]) + if (loading) return
Loading...
if (!doc) return null return (
+ {activeJob && ( + { setActiveJob(null); navigate(`/quizzes/${quizId}`) }} + onClose={() => setActiveJob(null)} + /> + )}