# parent/utils/ai.py import time import json # Gemini key now comes from the dynamic credential store (master admin # panel → Credentials), resolved lazily on each call. from core.utils.gemini import model def answer_parent_question(question: str, student_data: dict, retries: int = 3) -> dict: """ Answer a parent's natural language question using aggregated student context. Attendance data covers multiple periods (today, this week, this month, last month, academic year) so the AI picks the right one automatically. Returns dict with: answer (str), intent (str), data_found (bool) """ context = _build_context(student_data) prompt = f""" You are a friendly school assistant helping a parent about their child's school information. Student Information: {context} Parent's Question: "{question}" Instructions: - Answer ONLY based on the provided data above. Do not make up information. - Be warm, concise, and clear (2-4 sentences max). - Attendance: multiple time periods are provided — pick the one matching the question. - Homework: PENDING = not yet submitted. COMPLETED = already submitted (paginated). If more pages exist, tell the parent to ask for "more" homework. - Fee: use component breakdown. If all paid say so clearly. Mention next due date if pending. Summarize payment history if asked. - Exams: mention upcoming, ongoing, or published results as relevant. For results include subject-wise marks, grade, rank, and pass/fail per subject. - Timetable: answer day-wise schedule questions. If asked "what class does he have on Monday" list the periods for that day. - Teachers: provide name, role, subject, email, and phone when asked. - Transport: share bus number, driver name/phone, stop name, and arrival time when asked. - Announcements: summarize school-wide and class-specific notices if asked. - If the question is unrelated to any of the above data, politely say what you can help with. - Format dates as human-readable (e.g., "May 15, 2026"). - Use the student's first name to personalize the response. Respond ONLY in valid JSON (no extra text outside): {{ "answer": "Your friendly natural language answer here", "intent": "attendance" or "homework" or "fee" or "general" or "unknown", "data_found": true or false }} """ for attempt in range(retries): try: response = model.generate_content(prompt) response_text = response.text.strip() if "```json" in response_text: response_text = response_text.split("```json")[1].split("```")[0] elif "```" in response_text: response_text = response_text.split("```")[1].split("```")[0] result = json.loads(response_text) return result except json.JSONDecodeError: return { "answer": _fallback_answer(question, student_data), "intent": "unknown", "data_found": False, } except Exception as e: if "429" in str(e) and attempt < retries - 1: time.sleep(60 * (attempt + 1)) elif attempt < retries - 1: time.sleep(2) else: return { "answer": "I'm sorry, I couldn't process your question right now. Please try again.", "intent": "error", "data_found": False, } return { "answer": "I'm sorry, I couldn't process your question right now.", "intent": "error", "data_found": False, } # --------------------------------------------------------------------------- # Context builder # --------------------------------------------------------------------------- def _build_context(student_data: dict) -> str: student_name = student_data.get("student_name", "the student") lines = [ f"Student Name: {student_name}", f"Class: {student_data.get('class_name', 'N/A')}", f"Academic Year: {student_data.get('academic_year', 'N/A')}", "", ] # Attendance — multi-period attendance = student_data.get("attendance", {}) if attendance: lines.append( "ATTENDANCE DATA (multiple periods — pick the one matching the question):" ) lines.append("") period_labels = { "today": "Today", "this_week": "This Week", "this_month": "This Month", "last_month": "Last Month", "academic_year": "Academic Year (up to today)", } for key, display in period_labels.items(): period = attendance.get(key, {}) if not period: continue lines.append(f" [{display}] — {period.get('label', '')}:") lines.append( f" Total Working Days : {period.get('total_working_days', 0)}" ) lines.append(f" Present Days : {period.get('present_days', 0)}") lines.append(f" Absent Days : {period.get('absent_days', 0)}") lines.append(f" Late Days : {period.get('late_days', 0)}") lines.append(f" Leave Days : {period.get('leave_days', 0)}") lines.append( f" Attendance % : {period.get('attendance_percentage', 0)}%" ) lines.append("") # Pending Homework pending_hw = student_data.get("pending_homework", []) if pending_hw: lines.append(f"PENDING HOMEWORK ({len(pending_hw)} item(s) not yet submitted):") for hw in pending_hw[:10]: lines.append( f" - [{hw.get('subject', 'N/A')}] {hw.get('title', 'N/A')} | Due: {hw.get('due_date', 'N/A')}" ) lines.append("") else: lines += ["PENDING HOMEWORK: None — all caught up!", ""] # Completed Homework (paginated) completed_hw = student_data.get("completed_homework", []) pagination = student_data.get("completed_homework_pagination", {}) page = pagination.get("page", 1) total_count = pagination.get("total_count", 0) has_more = pagination.get("has_more", False) if completed_hw: lines.append( f"RECENTLY COMPLETED HOMEWORK (page {page}, showing {len(completed_hw)} of {total_count} total submissions):" ) for hw in completed_hw: marks_str = "" if ( hw.get("marks_obtained") is not None and hw.get("total_marks") is not None ): marks_str = f" | Marks: {hw['marks_obtained']}/{hw['total_marks']}" late_str = " [LATE]" if hw.get("is_late") else "" remarks_str = f" | Remarks: {hw['remarks']}" if hw.get("remarks") else "" lines.append( f" - [{hw.get('subject', 'N/A')}] {hw.get('title', 'N/A')}" f" | Submitted: {hw.get('submitted_on', 'N/A')}" f" | Due: {hw.get('due_date', 'N/A')}" f" | Status: {hw.get('status', 'N/A')}" f"{late_str}{marks_str}{remarks_str}" ) if has_more: lines.append( f" (More homeworks available — send homework_page={page + 1} to see next {pagination.get('page_size', 5)})" ) lines.append("") else: lines += ["COMPLETED HOMEWORK: No submissions found yet.", ""] # Fees fee = student_data.get("fee", {}) if fee: lines += [ "FEE INFORMATION (Current Academic Year):", f" Overall Status : {fee.get('overall_status', 'N/A')}", f" Total Fees : ₹{fee.get('total_fees', 0)}", f" Paid Amount : ₹{fee.get('paid_amount', 0)}", f" Pending Amount : ₹{fee.get('pending_amount', 0)}", ] if fee.get("next_due_date"): lines.append(f" Next Due Date : {fee['next_due_date']}") lines.append("") # Per-component breakdown components = fee.get("components", []) if components: lines.append(" Fee Breakdown (per component / term):") for comp in components: overdue_tag = " [OVERDUE]" if comp.get("is_overdue") else "" lines.append( f" • {comp.get('component', 'N/A')} [{comp.get('term', 'N/A')}]{overdue_tag}" ) lines.append( f" Total: ₹{comp.get('total_amount', 0)} | Paid: ₹{comp.get('paid_amount', 0)} | Pending: ₹{comp.get('pending_amount', 0)} | Status: {comp.get('status', 'N/A')} | Due: {comp.get('due_date', 'N/A')}" ) lines.append("") # Payment history history = fee.get("payment_history", []) if history: lines.append(" Payment History (recent completed payments):") for p in history: comps_str = ", ".join(p.get("components_paid", [])) or "N/A" lines.append( f" • {p.get('payment_date', 'N/A')} — ₹{p.get('amount_paid', 0)} via {p.get('payment_method', 'N/A')} | Receipt: {p.get('receipt_number', 'N/A')} | Components: {comps_str}" ) lines.append("") else: lines += [" Payment History: No payments recorded yet.", ""] # Exams exams = student_data.get("exams", {}) if exams: lines.append("EXAM INFORMATION:") upcoming = exams.get("upcoming", []) ongoing = exams.get("ongoing", []) results = exams.get("results", []) if upcoming: lines.append(f" Upcoming Exams ({len(upcoming)}):") for e in upcoming: lines.append(f" • {e['name']} [{e['type']}] | Start: {e['start_date']} | End: {e['end_date']}") if ongoing: lines.append(f" Ongoing Exams ({len(ongoing)}):") for e in ongoing: lines.append(f" • {e['name']} [{e['type']}] | {e['start_date']} – {e['end_date']}") if results: lines.append(f" Results Published ({len(results)}):") for e in results: r = e.get("result") or {} lines.append( f" • {e['name']} [{e['type']}] | " f"Marks: {r.get('total_marks', 'N/A')}/{r.get('max_marks', 'N/A')} | " f"Percentage: {r.get('percentage', 'N/A')}% | " f"Grade: {r.get('grade', 'N/A')} | Rank: {r.get('rank', 'N/A')} | " f"Status: {r.get('result_status', 'N/A')}" ) for sm in e.get("subject_marks", []): pass_str = "Pass" if sm.get("is_passed") else "Fail" lines.append( f" - {sm['subject']}: {sm['obtained']}/{sm['max']} | " f"Grade: {sm.get('grade', 'N/A')} | {pass_str}" ) if not upcoming and not ongoing and not results: lines.append(" No exams scheduled.") lines.append("") # Timetable timetable = student_data.get("timetable", {}) if timetable: lines.append("WEEKLY TIMETABLE:") for day, periods in timetable.items(): lines.append(f" {day}:") for p in periods: lines.append( f" P{p.get('period', '?')} | {p.get('start_time', '')}–{p.get('end_time', '')} | " f"{p.get('subject', 'N/A')} | Teacher: {p.get('teacher', 'N/A')}" ) lines.append("") # Teachers teachers = student_data.get("teachers", []) if teachers: lines.append("TEACHERS:") for t in teachers: subj = f" [{t['subject']}]" if t.get("subject") else "" email = f" | Email: {t['email']}" if t.get("email") else "" phone = f" | Phone: {t['phone']}" if t.get("phone") else "" lines.append(f" • {t['name']} — {t['role']}{subj}{email}{phone}") lines.append("") # Transport transport = student_data.get("transport") if transport: lines += [ "TRANSPORT DETAILS:", f" Route : {transport.get('route', 'N/A')}", f" Bus Number : {transport.get('bus_number', 'N/A')}", f" Vehicle No : {transport.get('vehicle_number', 'N/A')}", f" Driver : {transport.get('driver_name', 'N/A')}", f" Driver Phone: {transport.get('driver_phone', 'N/A')}", f" Stop : {transport.get('stop_name', 'N/A')}", f" Arrival Time: {transport.get('stop_arrival_time', 'N/A')}", "", ] else: lines += ["TRANSPORT: No transport assigned.", ""] # Announcements announcements = student_data.get("announcements", {}) school_wide = announcements.get("school_wide", []) class_specific = announcements.get("class_specific", []) if school_wide or class_specific: lines.append("RECENT ANNOUNCEMENTS:") if school_wide: lines.append(" School-Wide:") for a in school_wide: lines.append( f" • [{a.get('priority', '').upper()}] {a['title']} ({a['date']}): {a['message']}" ) if class_specific: lines.append(" Class-Specific:") for a in class_specific: lines.append( f" • [{a.get('priority', '').upper()}] {a['title']} ({a['date']}): {a['message']}" ) lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # Fallback (no AI) — keyword-based, time-range aware # --------------------------------------------------------------------------- def _fallback_answer(question: str, student_data: dict) -> str: q = question.lower() name = student_data.get("student_name", "Your child") att = student_data.get("attendance", {}) if any(w in q for w in ["absent", "attendance", "present", "miss", "late"]): # Detect time period from question if any( w in q for w in [ "academic year", "whole year", "full year", "entire year", "this year", ] ): period = att.get("academic_year", {}) label = period.get("label", "this academic year") elif any(w in q for w in ["last week", "previous week"]): period = att.get("this_week", {}) # best available fallback label = "last week" elif any(w in q for w in ["this week", "week"]): period = att.get("this_week", {}) label = period.get("label", "this week") elif any(w in q for w in ["last month", "previous month"]): period = att.get("last_month", {}) label = period.get("label", "last month") elif any(w in q for w in ["today"]): period = att.get("today", {}) label = period.get("label", "today") else: # Default to this month period = att.get("this_month", {}) label = period.get("label", "this month") absent = period.get("absent_days", 0) return f"{name} was absent for {absent} day(s) in {label}." if any(w in q for w in ["homework", "assignment", "task"]): if any( w in q for w in ["done", "completed", "submitted", "finished", "recent"] ): completed = student_data.get("completed_homework", []) pagination = student_data.get("completed_homework_pagination", {}) total = pagination.get("total_count", 0) if completed: titles = ", ".join(hw["title"] for hw in completed[:3]) more = f" and {total - 3} more" if total > 3 else "" return f"{name} has completed {total} homework(s). Recent ones: {titles}{more}." return f"{name} has no completed homework submissions yet." pending = student_data.get("pending_homework", []) count = len(pending) if count: return f"{name} has {count} pending homework assignment(s)." return f"{name} has no pending homework — all caught up!" if any(w in q for w in ["fee", "due", "payment", "paid", "pending"]): fee = student_data.get("fee", {}) if not fee: return f"No fee information is available for {name} at this time." status = fee.get("overall_status", "") pending = fee.get("pending_amount", 0) paid = fee.get("paid_amount", 0) total = fee.get("total_fees", 0) next_due = fee.get("next_due_date") if status == "All Paid": return ( f"Great news! {name} has paid all fees (₹{total}). No pending balance." ) due_str = f" Next due date is {next_due}." if next_due else "" return f"{name}'s total fees: ₹{total}. Paid: ₹{paid}. Pending: ₹{pending}.{due_str}" return "I can help you with attendance, homework, and fee information. Please try rephrasing your question."