"""Background tasks: document parsing and AI quiz generation.""" import asyncio from uuid import UUID from app.core.config import settings from app.core.logging import get_logger from app.workers.celery_app import celery_app logger = get_logger("tasks") # ── Sync wrapper helpers ──────────────────────────────── def _run_async(coro): """Run an async coroutine from within a sync Celery task.""" loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro) finally: loop.close() # ── Document parsing ──────────────────────────────────── def parse_document_sync(file_path: str, file_type: str) -> str: """Extract text from a document. Returns plain text.""" if file_type == "pdf": try: from pypdf import PdfReader reader = PdfReader(file_path) text = "\n".join(page.extract_text() for page in reader.pages if page.extract_text()) if text.strip(): return text except Exception as e: logger.warning("pypdf_parsing_fallback", error=str(e)) try: import fitz doc = fitz.open(file_path) text = "\n".join(page.get_text() for page in doc) doc.close() return text except Exception as e: logger.warning("fitz_parsing_failed", error=str(e)) with open(file_path, "r", errors="ignore") as f: return f.read() elif file_type in ("docx", "doc"): try: from docx import Document doc = Document(file_path) return "\n".join(para.text for para in doc.paragraphs if para.text.strip()) except Exception: with open(file_path, "r", errors="ignore") as f: return f.read() elif file_type in ("pptx", "ppt"): try: from pptx import Presentation prs = Presentation(file_path) texts = [] for slide in prs.slides: for shape in slide.shapes: if hasattr(shape, "text") and shape.text.strip(): texts.append(shape.text) return "\n".join(texts) except Exception: with open(file_path, "r", errors="ignore") as f: return f.read() else: with open(file_path, "r", errors="ignore") as f: return f.read() # ── AI quiz generation ────────────────────────────────── def generate_mcqs_from_text(text: str, num_questions: int = 10) -> list: """ Generate MCQs from text using an OpenAI-compatible API, with automatic NLP fallback. Each question: text, 4 options, 1 correct, explanation, topic_tag, confidence. """ try: from openai import OpenAI client = OpenAI( api_key=settings.AI_API_KEY, base_url=settings.AI_API_BASE, ) prompt = f"""You are a quiz generator for sales training at a financial services company. Given the following training material, generate {num_questions} multiple-choice questions. Rules: - Each question must have exactly 4 options with exactly 1 correct answer. - Include a clear explanation for why the correct answer is right. - Tag each question with a topic (1-3 words). - Assign a confidence score (0.0-1.0) for how well the question tests the material. Return JSON with this structure: {{ "questions": [ {{ "question_text": "...", "options": ["A", "B", "C", "D"], "correct_index": 0, "explanation": "...", "topic_tag": "...", "confidence_score": 0.9 }} ] }} Training material: {text[:8000]} """ response = client.chat.completions.create( model=settings.AI_MODEL, messages=[ {"role": "system", "content": "You are a quiz generation assistant."}, {"role": "user", "content": prompt}, ], response_format={"type": "json_object"}, temperature=0.7, ) import json content = response.choices[0].message.content result = json.loads(content) questions = result.get("questions", []) if questions: return questions except Exception as e: logger.warning("ai_generation_fallback_triggered", error=str(e)) # Rule-based NLP fallback from extracted document sentences paragraphs = [p.strip() for p in text.split("\n") if len(p.strip()) > 20] questions = [] for i, p in enumerate(paragraphs[:num_questions]): words = p.split() if len(words) < 5: continue q_text = f"According to the policy: '{' '.join(words[:12])}...', which statement is correct?" options = [ f"{' '.join(words[:8])}", f"Not applicable under Credit Fair guidelines", f"Requires secondary branch sign-off", f"Applies only to premium loans", ] questions.append({ "question_text": q_text, "options": options, "correct_index": 0, "explanation": f"Based on document section: '{p[:100]}...'", "topic_tag": "Policy Rules", "confidence_score": 0.85, }) if not questions: questions.append({ "question_text": "What is the primary objective of Credit Fair loan policies?", "options": [ "Streamlined customer onboarding and compliance", "Manual paper verification only", "High interest rates without audit", "Third-party processing", ], "correct_index": 0, "explanation": "Standard Credit Fair operational guideline.", "topic_tag": "Compliance", "confidence_score": 0.95, }) return questions # ── Celery tasks & Native Async Functions ──────────────── async def process_document_job_async(job_id: str): """Parse a document and extract text asynchronously.""" from app.db.base import SessionLocal from app.models import DocJobStatus, DocumentJob, SourceDocument async with SessionLocal() as db: job = await db.get(DocumentJob, UUID(job_id)) if not job: logger.error("job_not_found", job_id=job_id) return try: job.status = DocJobStatus.PARSING await db.commit() doc = await db.get(SourceDocument, job.document_id) if not doc: raise ValueError("Source document not found") # Parse document text = parse_document_sync(doc.file_path, doc.file_type) doc.extracted_text = text job.status = DocJobStatus.REVIEW_READY await db.commit() logger.info("document_parsed", job_id=job_id, text_len=len(text)) except Exception as e: logger.error("parse_failed", job_id=job_id, error=str(e)) job.status = DocJobStatus.FAILED job.error_message = str(e) await db.commit() @celery_app.task(name="process_document_job", bind=True, max_retries=3) def process_document_job(self, job_id: str): """Parse a document and extract text. Called as a background task by Celery.""" logger.info("processing_document", job_id=job_id) _run_async(process_document_job_async(job_id)) async def generate_quiz_from_document_async(job_id: str, user_id: str): """Generate a quiz from a parsed document using AI asynchronously.""" logger.info("generating_quiz", job_id=job_id) from sqlalchemy import select from app.db.base import SessionLocal from app.models import ( DocJobStatus, DocumentJob, ModuleModel, Question, QuestionOption, QuizModel, QuizStatus, SourceDocument, ) async with SessionLocal() as db: job = await db.get(DocumentJob, UUID(job_id)) if not job: return try: job.status = DocJobStatus.GENERATING await db.commit() doc = await db.get(SourceDocument, job.document_id) if not doc or not doc.extracted_text: raise ValueError("Document not parsed yet") # Find a module to attach the quiz to, or auto-provision a default one module_result = await db.execute(select(ModuleModel).limit(1)) module = module_result.scalar_one_or_none() if not module: from app.models import ProgramModel prog_res = await db.execute(select(ProgramModel).limit(1)) prog = prog_res.scalar_one_or_none() if not prog: prog = ProgramModel(name="General Onboarding", description="Default training program") db.add(prog) await db.flush() module = ModuleModel(program_id=prog.id, name="General Compliance", description="Default module") db.add(module) await db.flush() # Generate MCQs via AI questions_data = generate_mcqs_from_text(doc.extracted_text, num_questions=10) # Create quiz quiz = QuizModel( module_id=module.id, title=f"AI Quiz: {doc.filename}", description=f"Auto-generated from {doc.filename}", created_by=UUID(user_id), status=QuizStatus.DRAFT, ) db.add(quiz) await db.flush() # Create questions and options for q_data in questions_data: question = Question( quiz_id=quiz.id, question_text=q_data["question_text"], topic_tag=q_data.get("topic_tag", "general"), explanation=q_data.get("explanation", ""), confidence_score=q_data.get("confidence_score", 0.0), is_ai_generated=True, source_document_id=doc.id, ) db.add(question) await db.flush() correct_idx = q_data.get("correct_index", 0) for i, opt_text in enumerate(q_data["options"]): option = QuestionOption( question_id=question.id, option_text=opt_text, is_correct=(i == correct_idx), order_index=i, ) db.add(option) job.status = DocJobStatus.REVIEW_READY job.generated_quiz_id = quiz.id await db.commit() logger.info("quiz_generated", job_id=job_id, quiz_id=str(quiz.id), question_count=len(questions_data)) except Exception as e: logger.error("generation_failed", job_id=job_id, error=str(e)) job.status = DocJobStatus.FAILED job.error_message = str(e) await db.commit() @celery_app.task(name="generate_quiz_from_document", bind=True, max_retries=3) def generate_quiz_from_document(self, job_id: str, user_id: str): """Generate a quiz from a parsed document using AI. Called by Celery.""" _run_async(generate_quiz_from_document_async(job_id, user_id)) # ── Leaderboard snapshot job ─────────────────────────── @celery_app.task(name="compute_weekly_leaderboard") def compute_weekly_leaderboard(): """Compute the weekly leaderboard snapshot. Run via celery beat.""" logger.info("computing_leaderboard") async def _compute(): from datetime import datetime, timezone from sqlalchemy import and_, func, select from app.db.base import SessionLocal from app.models import ( Attempt, AttemptStatus, LeaderboardSnapshot, LearnerProgress, UserModel, ) async with SessionLocal() as db: # Get current ISO week now = datetime.now(timezone.utc) week_key = f"{now.year}-W{now.isocalendar()[1]:02d}" # Get all active users users_result = await db.execute( select(UserModel).where(UserModel.is_active == True) ) users = users_result.scalars().all() # Calculate points for each user this week entries = [] for user in users: # Mastery: average best score across modules progress_result = await db.execute( select(func.avg(LearnerProgress.best_score)) .where(LearnerProgress.learner_id == user.id) ) mastery = progress_result.scalar() or 0.0 # Improvement: score delta (latest - first) — simplified for MVP improvement = 0.0 # Would need to track first vs latest attempt # Consistency: number of attempts this week attempts_result = await db.execute( select(func.count(Attempt.id)) .where( and_( Attempt.learner_id == user.id, Attempt.status == AttemptStatus.COMPLETED, ) ) ) consistency = float(attempts_result.scalar() or 0) # Streak: consecutive weeks with at least 1 attempt — simplified streak = 1 # Would need to check previous weeks # Total score: mastery * 0.4 + improvement * 0.3 + consistency * 0.3 total = mastery * 0.4 + improvement * 0.3 + consistency * 0.3 entries.append({ "user_id": user.id, "user_name": user.full_name, "score": total, "streak": streak, "mastery": mastery, "improvement": improvement, "consistency": consistency, }) # Sort by score descending and assign ranks entries.sort(key=lambda x: x["score"], reverse=True) for rank, entry in enumerate(entries, 1): # Upsert leaderboard snapshot existing = await db.execute( select(LeaderboardSnapshot).where( and_( LeaderboardSnapshot.user_id == entry["user_id"], LeaderboardSnapshot.week_key == week_key, ) ) ) snap = existing.scalar_one_or_none() if not snap: snap = LeaderboardSnapshot( user_id=entry["user_id"], week_key=week_key, rank=rank, score=entry["score"], streak_count=entry["streak"], mastery_points=entry["mastery"], improvement_points=entry["improvement"], consistency_points=entry["consistency"], ) db.add(snap) else: snap.rank = rank snap.score = entry["score"] snap.streak_count = entry["streak"] snap.mastery_points = entry["mastery"] snap.improvement_points = entry["improvement"] snap.consistency_points = entry["consistency"] await db.commit() logger.info("leaderboard_computed", week=week_key, entries=len(entries)) _run_async(_compute())