# ----------------------- # -------------- IMPORTS # ----------------------- from flask import * import os import dotenv from dotenv import load_dotenv import pymysql from pymysql.cursors import DictCursor from werkzeug.security import generate_password_hash, check_password_hash from datetime import timedelta import json from datetime import date, time import markdown from flask_sse import sse import time import calendar as pycalendar import logging import os import requests # ----------------------- # --------------- CONFIG # ----------------------- load_dotenv() UPLOAD_FOLDER=os.path.join('static','uploads','homework') DOWNLOADS_FOLDER = os.path.join('static', 'uploads', 'downloads') os.makedirs(DOWNLOADS_FOLDER, exist_ok=True) os.makedirs(UPLOAD_FOLDER,exist_ok=True) logger = logging.getLogger(__name__) DOWNLOAD_CATEGORIES = ['SYLLABUS', 'DATESHEET', 'HOLIDAY_HOMEWORK', 'CIRCULAR', 'PLANNER', 'MISC'] app = Flask(__name__) def get_db(): if 'db' not in g: g.db = pymysql.connect( host=os.getenv('DB_HOST'), user=os.getenv('DB_USER'), password=os.getenv('DB_PASSWORD'), database=os.getenv('DB_NAME'), port=3306, cursorclass=DictCursor, autocommit=True ) return g.db @app.teardown_appcontext def close_db(error=None): db = g.pop('db', None) if db: db.close() def select_query(query, fetch='fetchall'): db = get_db() with db.cursor() as cursor: sql, params = query if isinstance(query, tuple) else (query, ()) cursor.execute(sql, params) if fetch == 'fetchone': return cursor.fetchone() elif fetch == 'fetchall': return cursor.fetchall() else: return None def execute_query(query): db = get_db() with db.cursor() as cursor: sql, params = query if isinstance(query, tuple) else (query, ()) cursor.execute(sql, params) def authenticate_user(role, username, password): if role not in ["ADMIN", "STUDENT", "TEACHER"]: return "ERROR" if role == "ADMIN": auth_query = "SELECT password FROM admin_users WHERE username = %s" elif role == "STUDENT": auth_query = "SELECT password FROM students WHERE addm_no = %s" elif role == "TEACHER": auth_query = "SELECT password FROM teachers WHERE username = %s" result = select_query( (auth_query, (username,)) , 'fetchone') if result: if check_password_hash(result['password'], password): return 'VALID' else: return 'INVALID' else: return '404' def authorized(conversation_id, role, username): """Check if user can access this conversation""" conv = select_query(("SELECT * FROM conversations WHERE id=%s", (conversation_id,)), 'fetchone') if not conv: return False if role == 'TEACHER' and conv['teacher_username'] == username: return True if role == 'STUDENT' and conv['student_addm_no'] == username: return True return False app.secret_key = os.getenv('FLASK_SECRET_KEY', os.getenv('SECRET_KEY')) app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=24) app.register_blueprint(sse, url_prefix='/stream') def get_or_create_conversation(teacher_username,student_addm_no): row=select_query(("SELECT id FROM conversations WHERE teacher_username=%s AND student_addm_no=%s",(teacher_username,student_addm_no)),'fetchone') if row: return row['id'] execute_query(("INSERT INTO conversations (teacher_username,student_addm_no) VALUES (%s,%s)",(teacher_username,student_addm_no))) return select_query(("SELECT id FROM conversations WHERE teacher_username=%s AND student_addm_no=%s",(teacher_username,student_addm_no)),'fetchone')['id'] def call_groq_for_doubt(student_class, student_section, doubt_text): """ Call Groq Chat Completions API and return explanation text. Raises on error; caller handles exceptions and HTTP codes. """ prompt = f"""You are an expert NCERT tutor for Class {student_class} Section {student_section}. Explain the student's doubt in clear step-by-step Hinglish (Hindi+English mix), suitable for an Indian school student. Keep it under 400–500 words and focus on concept clarity. Doubt: {doubt_text} """ api_key = 'GROQ_API_KEY=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' #os.getenv('GROQ_API_KEY') if not api_key: raise RuntimeError("GROQ_API_KEY not configured") headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": "llama3-8b-8192", "messages": [ {"role": "user", "content": prompt} ], "max_tokens": 900, "temperature": 0.6 } logger.info("Calling Groq API for doubt explanation") resp = requests.post( "https://api.groq.com/openai/v1/chat/completions", headers=headers, json=payload, timeout=30 ) # Log non‑2xx for debugging, like you do in /analyze if not resp.ok: logger.error( "Groq API error: status=%s, body=%s", resp.status_code, resp.text[:500] ) raise RuntimeError(f"Groq API error {resp.status_code}") data = resp.json() try: return data["choices"][0]["message"]["content"] except Exception as e: logger.error("Unexpected Groq response format: %s", data, exc_info=True) raise RuntimeError("Bad Groq response format") from e def get_doubt_history_from_session(limit=6): raw = session.get('DOUBTS_HISTORY') or [] history = [] for h in raw: # accept only dicts with expected keys if isinstance(h, dict) and 'role' in h and 'content' in h: history.append({'role': h['role'], 'content': h['content']}) return history[-limit:] def append_doubt_turn(role, content): raw = session.get('DOUBTS_HISTORY') or [] if not isinstance(raw, list): raw = [] raw.append({'role': role, 'content': content}) session['DOUBTS_HISTORY'] = raw[-10:] # ------------------------ # --------------- ROUTES # ------------------------ ### HOMEPAGE (Redirect to login) @app.route('/') def homepage(): return redirect(url_for('loginpage')) ### LOGIN PAGE @app.route('/login', methods=['GET', 'POST']) def loginpage(): if session.get('LOGGED_IN'): return redirect(url_for('dashboard')) if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') role = request.form.get('role') result = authenticate_user(role, username, password) if result == 'VALID': session.clear() session.permanent = True session['LOGGED_IN'] = True session['USERNAME'] = username session['ROLE'] = role return redirect(url_for('dashboard')) if result == 'INVALID': return render_template('login.html', error="Invalid credentials") if result == '404': return render_template('login.html', error="User not found") if result == 'ERROR': return render_template('login.html', error="System Error - contact mail@thepolar.in or +91 9999791415") return render_template('login.html') ### DASHBOARD @app.route('/dashboard') def dashboard(): if not session.get('LOGGED_IN'): return redirect(url_for('loginpage')) role = session.get('ROLE') if role == 'STUDENT': details = select_query(("SELECT * FROM students WHERE addm_no = %s", (session["USERNAME"],)), 'fetchone') return render_template('students/dashboard.html', details=details) elif role == 'TEACHER': details = select_query(("SELECT * FROM teachers WHERE username = %s", (session["USERNAME"],)), 'fetchone') return render_template('teachers/dashboard.html', details=details) else: return redirect(url_for('loginpage')) ### PROFILE @app.route('/profile') def profile(): if not session.get('LOGGED_IN'): return redirect(url_for('loginpage')) role = session.get('ROLE') username = session.get('USERNAME') if role == 'TEACHER': details = select_query(("SELECT * FROM teachers WHERE username = %s", (username,)), 'fetchone') raw_classes = details.get('classes_json') or '[]' raw_subjects = details.get('subjects_json') or '[]' classes = json.loads(raw_classes) subjects = json.loads(raw_subjects) classes_list = [(c[0], c[1]) for c in classes] subjects_list = subjects return render_template('teachers/profile.html', details=details, classes_list=classes_list, subjects_list=subjects_list) elif role == 'STUDENT': details = select_query(("SELECT * FROM students WHERE addm_no = %s", (username,)), 'fetchone') return render_template('students/profile.html', details=details) else: return redirect(url_for('loginpage')) ### HOMEWORK @app.route('/homework', methods=['GET','POST']) def homework(): if not session.get('LOGGED_IN'): return redirect(url_for('loginpage')) role=session.get('ROLE') username=session.get('USERNAME') if role=='TEACHER': details=select_query(("SELECT * FROM teachers WHERE username = %s",(username,)),'fetchone') raw_classes=details.get('classes_json') or '[]' raw_subjects=details.get('subjects_json') or '[]' classes=json.loads(raw_classes) subjects=json.loads(raw_subjects) classes_list=[(c[0],c[1]) for c in classes] subjects_list=subjects if request.method=='POST': subject=request.form.get('subject') title=request.form.get('title') class_=request.form.get('class') section=request.form.get('section') file=request.files.get('file') if not file or file.filename=='': return render_template('teachers/homework.html',details=details,classes_list=classes_list,subjects_list=subjects_list,homework_list=select_query(("SELECT class, section, subject, title, filepath, created_at FROM homework WHERE teacher_username = %s ORDER BY created_at DESC",(username,)),'fetchall'),error="NO FILE SELECTED") counter_row=select_query(("SELECT current_number FROM homework_counter WHERE class = %s AND section = %s",(class_,section)),'fetchone') if counter_row: current_number=counter_row['current_number']+1 execute_query(("UPDATE homework_counter SET current_number = %s WHERE class = %s AND section = %s",(current_number,class_,section))) else: current_number=0 execute_query(("INSERT INTO homework_counter (class,section,current_number) VALUES (%s,%s,%s)",(class_,section,current_number))) number_str=str(current_number).zfill(8) base_name=f"{class_}{section}_HOMEWORK-{number_str}" _,ext=os.path.splitext(file.filename) filename=base_name+ext.lower() save_path=os.path.join(UPLOAD_FOLDER,filename) file.save(save_path) db_path=f"static/uploads/homework/{filename}" execute_query(("INSERT INTO homework (class,section,subject,title,filepath,teacher_username) VALUES (%s,%s,%s,%s,%s,%s)",(class_,section,subject,title,db_path,username))) return redirect(url_for('homework')) homework_list=select_query(("SELECT class, section, subject, title, filepath, created_at FROM homework WHERE teacher_username = %s ORDER BY created_at DESC",(username,)),'fetchall') return render_template('teachers/homework.html',details=details,classes_list=classes_list,subjects_list=subjects_list,homework_list=homework_list) elif role=='STUDENT': details=select_query(("SELECT * FROM students WHERE addm_no = %s",(username,)),'fetchone') student_class=details['class'] student_section=details['section'] all_hw=select_query(("SELECT class, section, subject, title, filepath, created_at, teacher_username FROM homework WHERE class = %s AND section = %s ORDER BY created_at DESC",(student_class,student_section)),'fetchall') today_str=date.today().strftime("%Y-%m-%d") homework_today=[hw for hw in all_hw if str(hw['created_at']).startswith(today_str)] homework_previous=[hw for hw in all_hw if not str(hw['created_at']).startswith(today_str)] return render_template('students/homework.html',details=details,homework_today=homework_today,homework_previous=homework_previous) else: return redirect(url_for('loginpage')) @app.route('/homework/download/') def download_homework(filename): if not session.get('LOGGED_IN'): return redirect(url_for('loginpage')) return send_from_directory(UPLOAD_FOLDER, filename, as_attachment=True) ## DOWNLOADS @app.route('/downloads',methods=['GET','POST']) def downloads(): if not session.get('LOGGED_IN'): return redirect(url_for('loginpage')) role=session.get('ROLE') username=session.get('USERNAME') if role=='TEACHER': details=select_query(("SELECT * FROM teachers WHERE username = %s",(username,)),'fetchone') classes=json.loads(details.get('classes_json') or '[]') classes_list=[(c[0],c[1]) for c in classes] error=None if request.method=='POST': title=request.form.get('title') category=request.form.get('category') class_=request.form.get('class') section=request.form.get('section') file=request.files.get('file') if not title or not category or category not in DOWNLOAD_CATEGORIES: error="INVALID TITLE OR CATEGORY" elif not file or file.filename=='': error="NO FILE SELECTED" else: import time _,ext=os.path.splitext(file.filename) filename=f"{category}_{int(time.time())}{ext.lower()}" file.save(os.path.join(DOWNLOADS_FOLDER,filename)) db_path=f"static/uploads/downloads/{filename}" execute_query(("INSERT INTO downloads (title,file_path,category,class,section,teacher_username) VALUES (%s,%s,%s,%s,%s,%s)",(title,db_path,category,class_,section,username))) return redirect(url_for('downloads')) downloads_list=select_query(("SELECT id,title,file_path,category,class,section,created_at FROM downloads WHERE teacher_username = %s ORDER BY created_at DESC",(username,)),'fetchall') return render_template('teachers/downloads.html',details=details,classes_list=classes_list,categories=DOWNLOAD_CATEGORIES,downloads_list=downloads_list,error=error) elif role=='STUDENT': details=select_query(("SELECT * FROM students WHERE addm_no = %s",(username,)),'fetchone') student_class=details['class'] student_section=details['section'] downloads_list=select_query(("SELECT id,title,file_path,category,class,section,created_at,teacher_username FROM downloads WHERE class = %s AND section = %s ORDER BY created_at DESC",(student_class,student_section)),'fetchall') return render_template('students/downloads.html',details=details,downloads_list=downloads_list) else: return redirect(url_for('loginpage')) @app.route('/downloads/file/') def download_file(filename): if not session.get('LOGGED_IN'): return redirect(url_for('loginpage')) return send_from_directory(DOWNLOADS_FOLDER,filename,as_attachment=True) ### NOTICE BOARD @app.route('/noticeboard', methods=['GET', 'POST']) def noticeboard(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') username = session.get('USERNAME') if role == 'STUDENT': details = select_query(("SELECT * FROM students WHERE addm_no = %s",(username,)), 'fetchone') rows = select_query('SELECT * FROM notice_board ORDER BY is_important DESC, date_for DESC, created_at DESC','fetchall') notices = [] for r in rows: content_html = markdown.markdown(r['content_md'], extensions=['extra']) r['content_html'] = content_html notices.append(r) return render_template('students/notices.html', details=details, notices=notices) elif role=='TEACHER': details=select_query(("SELECT * FROM teachers WHERE username = %s",(username,)),'fetchone') error=None success=None if request.method=='POST': title=request.form.get('title','').strip() date_for=request.form.get('date_for','').strip() content_md=request.form.get('content_md','').strip() is_important=1 if request.form.get('is_important')=='on' else 0 if not title or not content_md or not date_for: error="Please fill in title, date and content." else: try: execute_query(("INSERT INTO notice_board (title,content_md,date_for,is_important,teacher_username) VALUES (%s,%s,%s,%s,%s)",(title,content_md,date_for,is_important,username))) success="Notice added to the board." except: error="Could not save notice. Please try again." notices=select_query(("SELECT * FROM notice_board ORDER BY is_important DESC,date_for DESC,created_at DESC"),"fetchall") for n in notices: n['content_html']=markdown.markdown(n['content_md'],extensions=['extra']) return render_template('teachers/notices.html',details=details,error=error,success=success,notices=notices) import calendar as pycalendar from datetime import date, datetime from flask import render_template, redirect, url_for, session # day_type values: # 'SCHOOL','SUNDAY','EXTRA_HOLIDAY','EXAM','EVENT' @app.route('/calendar') def calendar_root(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) today = date.today() return redirect(url_for('calendar_view', year=today.year, month=today.month)) @app.route('/calendar//') def calendar_view(year, month): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') username = session.get('USERNAME') # basic guard if month < 1 or month > 12: today = date.today() return redirect(url_for('calendar_view', year=today.year, month=today.month)) # user details (student example; adapt for teacher if needed) if role == 'STUDENT': details = select_query( ("SELECT * FROM students WHERE addm_no = %s", (username,)), 'fetchone' ) elif role == 'TEACHER': details = select_query( ("SELECT * FROM teachers WHERE username = %s", (username,)), 'fetchone' ) else: details = None # --- build month days --- first_weekday, num_days = pycalendar.monthrange(year, month) # monthrange: Monday=0..Sunday=6, num_days=28..31 today = date.today() days_in_month = [] for day in range(1, num_days + 1): d = date(year, month, day) days_in_month.append({ 'date': d, 'day': day, 'is_today': (d == today), }) # --- fetch calendar info from DB for this month --- start_date = date(year, month, 1) end_date = date(year, month, num_days) rows = select_query( ("SELECT date, day_type, title, description " "FROM academic_calendar " "WHERE date BETWEEN %s AND %s"), 'fetchall' ) if False else select_query( # adapt to your helper signature ("SELECT date, day_type, title, description " "FROM academic_calendar " "WHERE date BETWEEN %s AND %s", (start_date, end_date)), 'fetchall' ) calendar_map = {} for r in rows: # r['date'] is a date object calendar_map[r['date']] = r # --- previous/next month for navigation --- if month == 1: prev_month = 12 prev_year = year - 1 else: prev_month = month - 1 prev_year = year if month == 12: next_month = 1 next_year = year + 1 else: next_month = month + 1 next_year = year month_name = pycalendar.month_name[month] # first_weekday_offset: convert Monday(0)..Sunday(6) to our grid starting Monday first_weekday_offset = first_weekday # 0..6, already Monday-based return render_template( 'students/calendar.html', details=details, role=role, year=year, month=month, month_name=month_name, prev_year=prev_year, prev_month=prev_month, next_year=next_year, next_month=next_month, first_weekday_offset=first_weekday_offset, days_in_month=days_in_month, calendar_map=calendar_map ) ### COMMUNICATE @app.route('/messages') def messages_home(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role,username=session.get('ROLE'),session.get('USERNAME') if role=='TEACHER': conversations=select_query(( "SELECT c.id,c.student_addm_no,s.fullname student_name,c.created_at," "(SELECT body FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_message," "(SELECT created_at FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_time " "FROM conversations c LEFT JOIN students s ON s.addm_no=c.student_addm_no " "WHERE c.teacher_username=%s " "ORDER BY (CASE WHEN last_time IS NULL THEN 1 ELSE 0 END), last_time DESC, c.created_at DESC", (username,)),'fetchall') elif role=='STUDENT': conversations=select_query(( "SELECT c.id,c.teacher_username,t.fullname teacher_name,c.created_at," "(SELECT body FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_message," "(SELECT created_at FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_time " "FROM conversations c LEFT JOIN teachers t ON t.username=c.teacher_username " "WHERE c.student_addm_no=%s " "ORDER BY (CASE WHEN last_time IS NULL THEN 1 ELSE 0 END), last_time DESC, c.created_at DESC", (username,)),'fetchall') else: conversations=[] if role == 'TEACHER': details = select_query( ("SELECT * FROM teachers WHERE username=%s", (username,)), 'fetchone' ) else: details = select_query( ("SELECT * FROM students WHERE addm_no=%s", (username,)), 'fetchone' ) return render_template('messages/list.html',role=role,conversations=conversations, details=details) @app.route('/messages/with/') def messages_with(partner_id): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role,username=session.get('ROLE'),session.get('USERNAME') if role=='TEACHER': conv_id=get_or_create_conversation(username,partner_id) elif role=='STUDENT': conv_id=get_or_create_conversation(partner_id,username) else: return redirect(url_for('homepage')) return redirect(url_for('messages_conversation',conversation_id=conv_id)) @app.route('/messages/conversation/', methods=['GET', 'POST']) def messages_conversation(conversation_id): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role, username = session.get('ROLE'), session.get('USERNAME') conv = select_query(("SELECT * FROM conversations WHERE id=%s",(conversation_id,)),'fetchone') if not conv or (role=='TEACHER' and conv['teacher_username']!=username) or (role=='STUDENT' and conv['student_addm_no']!=username): return redirect(url_for('messages_home')) if request.method == 'POST': body = (request.form.get('body') or '').strip() if not body: return {"ok": False, "error": "empty"}, 400 # 🔒 derive sender from conversation ownership if role == 'STUDENT' and conv['student_addm_no'] == username: sender_role = 'STUDENT' sender_username = username elif role == 'TEACHER' and conv['teacher_username'] == username: sender_role = 'TEACHER' sender_username = username else: return {"ok": False, "error": "unauthorized"}, 403 execute_query(( "INSERT INTO messages (conversation_id,sender_role,sender_username,body) " "VALUES (%s,%s,%s,%s)", (conversation_id, sender_role, sender_username, body) )) return {"ok": True} # GET → normal render msgs = select_query(("SELECT * FROM messages WHERE conversation_id=%s ORDER BY created_at ASC,id ASC",(conversation_id,)),'fetchall') # fetch ALL conversations for sidebar (includes the active one) if role == 'TEACHER': conversations = select_query(( "SELECT c.id,c.student_addm_no,s.fullname student_name,c.created_at," "(SELECT body FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_message," "(SELECT created_at FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_time " "FROM conversations c LEFT JOIN students s ON s.addm_no=c.student_addm_no " "WHERE c.teacher_username=%s " "ORDER BY (CASE WHEN last_time IS NULL THEN 1 ELSE 0 END), last_time DESC, c.created_at DESC", (username,)),'fetchall') else: # STUDENT conversations = select_query(( "SELECT c.id,c.teacher_username,t.fullname teacher_name,c.created_at," "(SELECT body FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_message," "(SELECT created_at FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) last_time " "FROM conversations c LEFT JOIN teachers t ON t.username=c.teacher_username " "WHERE c.student_addm_no=%s " "ORDER BY (CASE WHEN last_time IS NULL THEN 1 ELSE 0 END), last_time DESC, c.created_at DESC", (username,)),'fetchall') # details for topbar if role == 'TEACHER': details = select_query(("SELECT * FROM teachers WHERE username=%s",(username,)),'fetchone') else: details = select_query(("SELECT * FROM students WHERE addm_no=%s",(username,)),'fetchone') partner = select_query(( "SELECT fullname,addm_no,class,section FROM students WHERE addm_no=%s" if role=='TEACHER' else "SELECT fullname,username FROM teachers WHERE username=%s", (conv['student_addm_no'] if role=='TEACHER' else conv['teacher_username'],) ),'fetchone') return render_template( 'messages/conversation.html', role=role, details=details, conversation=conv, conversations=conversations, messages=msgs, partner=partner, me_username=username, ) @app.route('/messages/stream/') def messages_stream(conversation_id): if not session.get('LOGGED_IN'): return "UNAUTH", 401 # ✅ snapshot session ONCE role = session.get('ROLE') username = session.get('USERNAME') conv = select_query( ("SELECT * FROM conversations WHERE id=%s", (conversation_id,)), 'fetchone' ) if not conv or ( role == 'TEACHER' and conv['teacher_username'] != username ) or ( role == 'STUDENT' and conv['student_addm_no'] != username ): return "UNAUTH", 403 def event_gen(): last_id = 0 while True: rows = select_query(( "SELECT * FROM messages " "WHERE conversation_id=%s AND id>%s " "ORDER BY id ASC", (conversation_id, last_id) ), 'fetchall') for r in rows: last_id = r['id'] data = { 'id': r['id'], 'body': r['body'], 'sender_role': r['sender_role'], 'sender_username': r['sender_username'], 'created_at': r['created_at'].isoformat() } yield f"data: {json.dumps(data)}\n\n" time.sleep(1) return Response( stream_with_context(event_gen()), mimetype="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no" # nginx fix } ) @app.route('/api/search-people') def api_search_people(): if not session.get('LOGGED_IN'): return jsonify({"results": []}) role = request.args.get('role', '').upper() adm = (request.args.get('adm') or '').strip() name = (request.args.get('name') or '').strip() rows = [] if role == 'STUDENT': # simple example, you can refine rows = select_query(( "SELECT addm_no, fullname, class, section " "FROM students " "WHERE (%s = '' OR addm_no LIKE %s) " "AND (%s = '' OR fullname LIKE %s) " "ORDER BY fullname LIMIT 20", (adm, f"%{adm}%", name, f"%{name}%") ), 'fetchall') results = [{ "initial": (r['fullname'] or r['addm_no'])[0].upper(), "display_name": r['fullname'], "meta": f"Class {r['class']} {r['section']} • {r['addm_no']}", "addm_no": r['addm_no'] } for r in rows] elif role == 'TEACHER': rows = select_query(( "SELECT username, fullname, subjects_json, class, section " "FROM teachers " "WHERE (%s = '' OR fullname LIKE %s) " "ORDER BY fullname LIMIT 20", (name, f"%{name}%") ), 'fetchall') results = [] for r in rows: # subjects_json → list of subjects subjects = [] if r.get('subjects_json'): try: subjects = json.loads(r['subjects_json']) except Exception: subjects = [] subj_str = ', '.join(subjects[:3]) if isinstance(subjects, list) else '' meta_parts = [] if subj_str: meta_parts.append(f"Subjects: {subj_str}") # use class + section to show class teacher info if r.get('class'): meta_parts.append(f"Class teacher: {r['class']} {r.get('section','')}") results.append({ "initial": (r['fullname'] or r['username'])[0].upper(), "display_name": r['fullname'], "meta": ' • '.join(meta_parts) or r['username'], "username": r['username'] }) else: results = [] return jsonify({"results": results}) @app.route('/library') def library(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') username = session.get('USERNAME') # for student, this is addm_no if role != 'STUDENT': return redirect(url_for('dashboard')) # student details (for topbar) details = select_query( ("SELECT * FROM students WHERE addm_no = %s", (username,)), 'fetchone' ) # Fetch issued books for this student, latest first issued = select_query( ("SELECT * FROM issued_books " "WHERE addm_no = %s " "ORDER BY issue_date DESC, id DESC", (username,)), 'fetchall' ) return render_template( 'students/library.html', details=details, issued_books=issued ) @app.route('/attendance') def attendance_root(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') if role != 'STUDENT': return redirect(url_for('dashboard')) today = date.today() return redirect(url_for('attendance_view', year=today.year, month=today.month)) @app.route('/attendance//') def attendance_view(year, month): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') username = session.get('USERNAME') if role != 'STUDENT': # later you can add teacher view for any student abort(403) # guard month if month < 1 or month > 12: today = date.today() return redirect(url_for('attendance_view', year=today.year, month=today.month)) # student details details = select_query( ("SELECT * FROM students WHERE addm_no=%s", (username,)), 'fetchone' ) # month info first_weekday, num_days = pycalendar.monthrange(year, month) today = date.today() days_in_month = [] for dday in range(1, num_days + 1): d = date(year, month, dday) days_in_month.append({ 'date': d, 'day': dday, 'is_today': (d == today), }) start_date = date(year, month, 1) end_date = date(year, month, num_days) # attendance rows ONLY for this student and month att_rows = select_query( ("SELECT att_date, status FROM attendance " "WHERE student_id=%s AND att_date BETWEEN %s AND %s", (username, start_date, end_date)), 'fetchall' ) att_map = {r['att_date']: r['status'] for r in att_rows} # academic_calendar for holidays (optional) cal_rows = select_query( ("SELECT date, day_type FROM academic_calendar " "WHERE date BETWEEN %s AND %s", (start_date, end_date)), 'fetchall' ) cal_map = {r['date']: r['day_type'] for r in cal_rows} # prev/next month nav if month == 1: prev_month, prev_year = 12, year - 1 else: prev_month, prev_year = month - 1, year if month == 12: next_month, next_year = 1, year + 1 else: next_month, next_year = month + 1, year month_name = pycalendar.month_name[month] first_weekday_offset = first_weekday # Monday=0..Sunday=6 # stats total_days = 0 present_count = 0 absent_count = 0 leave_count = 0 for d in days_in_month: dt = d['date'] # skip future dates if dt > today: continue # treat Sundays & extra holidays as non-working is_sunday = (dt.weekday() == 6) day_type = cal_map.get(dt) # e.g. 'HOLIDAY' if is_sunday or day_type == 'HOLIDAY': continue total_days += 1 st = att_map.get(dt) # None, 'ABSENT', 'LEAVE' if st == 'ABSENT': absent_count += 1 elif st == 'LEAVE': leave_count += 1 else: present_count += 1 overall_pct = round((present_count / total_days) * 100, 1) if total_days else 0.0 # monthly stats dict (one month for now; easy to extend later) monthly_stats = [{ 'name': month_name, 'present': present_count, 'total': total_days, 'percent': overall_pct }] return render_template( 'students/attendance_calendar.html', details=details, role=role, year=year, month=month, month_name=month_name, prev_year=prev_year, prev_month=prev_month, next_year=next_year, next_month=next_month, first_weekday_offset=first_weekday_offset, days_in_month=days_in_month, att_map=att_map, cal_map=cal_map, present_count=present_count, absent_count=absent_count, leave_count=leave_count, total_days=total_days, overall_pct=overall_pct, monthly_stats=monthly_stats ) @app.route('/timetable') def timetable_view(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') username = session.get('USERNAME') if role == 'STUDENT': details = select_query( ("SELECT fullname, class, section, addm_no FROM students WHERE addm_no=%s", (username,)), 'fetchone' ) target_class = details['class'] target_section = details['section'] elif role == 'TEACHER': # later you can choose class/section via query params return "Teacher timetable view not implemented yet", 501 else: abort(403) days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat'] rows = select_query(( "SELECT day, period1,period2,period3,period4,period5,period6,period7,period8 " "FROM timetable WHERE class=%s AND section=%s", (target_class, target_section) ), 'fetchall') data_by_day = {r['day']: r for r in rows} periods = [ {'no': 1, 'label': 'Period 1'}, {'no': 2, 'label': 'Period 2'}, {'no': 3, 'label': 'Period 3'}, {'no': 4, 'label': 'Period 4'}, {'no': 5, 'label': 'Period 5'}, {'no': 6, 'label': 'Period 6'}, {'no': 7, 'label': 'Period 7'}, {'no': 8, 'label': 'Period 8'}, ] return render_template( 'students/timetable.html', details=details, role=role, days=days, periods=periods, data_by_day=data_by_day ) # ------------------------ # -------- RESULTS ------- # ------------------------ @app.route('/results') def results_view(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') username = session.get('USERNAME') if role != 'STUDENT': return redirect(url_for('dashboard')) # student details details = select_query( ("SELECT * FROM students WHERE addm_no=%s", (username,)), 'fetchone' ) student_class = int(details['class']) student_section = details['section'] # fetch exam config for this class (1–10) cfg_rows = select_query(( "SELECT subject_code, exam_code, max_marks " "FROM exam_config " "WHERE %s BETWEEN class_from AND class_to " "ORDER BY subject_code, FIELD(exam_code,'AC1','AC2','AC3','AC4','UT1','HY','UT2','ANNUAL')", (student_class,) ), 'fetchall') # group config by subject subjects = {} for r in cfg_rows: subj = r['subject_code'] if subj not in subjects: subjects[subj] = {} subjects[subj][r['exam_code']] = r['max_marks'] # fetch obtained marks for this student mark_rows = select_query(( "SELECT subject_code, exam_code, obtained_marks " "FROM marks " "WHERE addm_no=%s", (username,) ), 'fetchall') obtained = {} for r in mark_rows: obtained.setdefault(r['subject_code'], {})[r['exam_code']] = r['obtained_marks'] # derive exam list in order from config exam_order = [] for subj, exams in subjects.items(): for ex in exams.keys(): if ex not in exam_order: exam_order.append(ex) # build per-subject rows subject_rows = [] total_obtained = 0 total_max = 0 for subj, exams in subjects.items(): row = {'subject': subj, 'exams': [], 'total_obt': 0, 'total_max': 0} for ex in exam_order: if ex in exams: max_m = exams[ex] obt_m = obtained.get(subj, {}).get(ex) row['exams'].append({ 'exam': ex, 'obtained': obt_m, 'max': max_m }) if obt_m is not None: row['total_obt'] += obt_m row['total_max'] += max_m total_obtained += row['total_obt'] total_max += row['total_max'] subject_rows.append(row) overall_percent = round((total_obtained / total_max) * 100, 2) if total_max else 0.0 # term-wise: Term 1 = UT1+HY, Term 2 = UT2+ANNUAL (or AC1–AC4 for junior) # keep simple summary: term1_obt = term1_max = term2_obt = term2_max = 0 for subj, exams in subjects.items(): for ex, max_m in exams.items(): obt_m = obtained.get(subj, {}).get(ex) or 0 if ex in ('AC1', 'AC2', 'UT1', 'HY'): term1_obt += obt_m term1_max += max_m if ex in ('AC3', 'AC4', 'UT2', 'ANNUAL'): term2_obt += obt_m term2_max += max_m term1_percent = round((term1_obt / term1_max) * 100, 2) if term1_max else 0.0 term2_percent = round((term2_obt / term2_max) * 100, 2) if term2_max else 0.0 return render_template( 'students/results.html', details=details, subject_rows=subject_rows, exam_order=exam_order, overall_percent=overall_percent, term1_percent=term1_percent, term2_percent=term2_percent, total_obtained=total_obtained, total_max=total_max ) @app.route('/results/manage', methods=['GET', 'POST']) def results_manage(): if not session.get('LOGGED_IN'): return redirect(url_for('homepage')) role = session.get('ROLE') username = session.get('USERNAME') if role != 'TEACHER': return redirect(url_for('dashboard')) details = select_query( ("SELECT * FROM teachers WHERE username=%s", (username,)), 'fetchone' ) # classes_json examples: # ["9A","10B"] OR [["9","A"],["10","B"]] raw_classes = json.loads(details.get('classes_json') or '[]') class_set = set() section_set = set() for c in raw_classes: if isinstance(c, (list, tuple)) and len(c) == 2: cls, sec = str(c[0]), str(c[1]) else: s = str(c) cls, sec = s[:-1], s[-1] # "9A" -> "9","A" class_set.add(cls) section_set.add(sec) classes_only = sorted(class_set, key=lambda x: int(x)) sections_only = sorted(section_set) subjects = json.loads(details.get('subjects_json') or '[]') # from query string selected_class = request.args.get('cls') or '' selected_section = request.args.get('section') or '' selected_exam = request.args.get('exam') or '' # FIX: fixed exam list (for now; you can branch by class if needed) exams_available = ['UT1', 'HY', 'UT2', 'ANNUAL'] students = [] marks_map = {} if selected_class and selected_section: if not selected_exam: selected_exam = exams_available[0] students = select_query(( "SELECT addm_no, fullname " "FROM students " "WHERE class=%s AND section=%s " "ORDER BY fullname", (selected_class, selected_section) ), 'fetchall') if subjects and selected_exam: mark_rows = select_query(( "SELECT addm_no, subject_code, obtained_marks " "FROM marks " "WHERE class=%s AND section=%s AND exam_code=%s " "AND subject_code IN (" + ",".join(["%s"] * len(subjects)) + ")", tuple([selected_class, selected_section, selected_exam] + subjects) ), 'fetchall') for r in mark_rows: marks_map.setdefault(r['addm_no'], {})[r['subject_code']] = r['obtained_marks'] if request.method == 'POST': selected_class = request.form.get('class') selected_section = request.form.get('section') selected_exam = request.form.get('exam') for adm in request.form: if not adm.startswith('m_'): continue _, adm_no, subj = adm.split('_', 2) val = (request.form.get(adm) or '').strip() if val == '': continue try: obt = float(val) except ValueError: continue cfg = select_query(( "SELECT max_marks FROM exam_config " "WHERE %s BETWEEN class_from AND class_to " "AND subject_code=%s AND exam_code=%s", (int(selected_class), subj, selected_exam) ), 'fetchone') if not cfg: continue existing = select_query(( "SELECT id FROM marks " "WHERE addm_no=%s AND subject_code=%s AND exam_code=%s", (adm_no, subj, selected_exam) ), 'fetchone') if existing: execute_query(( "UPDATE marks SET obtained_marks=%s WHERE id=%s", (obt, existing['id']) )) else: execute_query(( "INSERT INTO marks (addm_no, class, section, subject_code, exam_code, obtained_marks) " "VALUES (%s,%s,%s,%s,%s,%s)", (adm_no, selected_class, selected_section, subj, selected_exam, obt) )) return redirect(url_for( 'results_manage', cls=selected_class, section=selected_section, exam=selected_exam )) return render_template( 'teachers/results_manage.html', details=details, classes_only=classes_only, sections_only=sections_only, subjects=subjects, selected_class=selected_class, selected_section=selected_section, selected_exam=selected_exam, exams_available=exams_available, students=students, marks_map=marks_map ) @app.route('/doubts') def doubts_page(): if not session.get('LOGGED_IN') or session.get('ROLE') != 'STUDENT': return redirect(url_for('loginpage')) student = select_query( ("SELECT fullname, class, section, addm_no FROM students WHERE addm_no = %s", session['USERNAME']), fetch='fetchone' ) if not student: return redirect(url_for('dashboard')) return render_template('students/doubts.html', details=student) @app.route('/doubts/api', methods=['POST']) def doubts_api(): from flask import session # ensure imported at top of file import os, requests if not session.get('LOGGED_IN') or session.get('ROLE') != 'STUDENT': return jsonify({'error': 'Student only'}), 403 data = request.get_json(silent=True) or {} doubt_text = (data.get('doubt') or '').strip() if not doubt_text: return jsonify({'error': 'Doubt text required'}), 400 student = select_query( ("SELECT class, section, fullname FROM students WHERE addm_no = %s", session['USERNAME']), fetch='fetchone' ) if not student: return jsonify({'error': 'Student not found'}), 404 student_class = student['class'] student_section = student['section'] student_name = student['fullname'] # ---------- SAFE HISTORY READ ---------- raw_history = session.get('DOUBTS_HISTORY') if not isinstance(raw_history, list): raw_history = [] history = [] for h in raw_history: # accept only dicts with expected keys if isinstance(h, dict) and 'role' in h and 'content' in h: history.append({'role': h['role'], 'content': h['content']}) # last 6 turns history = history[-6:] # ---------- BUILD MESSAGES ---------- system_msg = ( "You are an NCERT science/maths tutor for Indian school students.\n" f"- The student is in Class {student_class} Section {student_section}.\n" f"- The student's name is {student_name}. Use their name sometimes in the explanation.\n" "- LANGUAGE STYLE:\n" " - Your answer must be mostly in simple, clear ENGLISH sentences.\n" " - You may add a few very simple Hindi words in Roman script (Hinglish) to make it friendly.\n" " - Hindi should be a light mix only (like mirch-masala), not the main language.\n" " - Example Hindi words: asaan, mushkil, baccha, padhai, samajh, dhyaan.\n" " - Do NOT use Devanagari script, only Roman letters for Hindi.\n" " - Do NOT write the full answer in Hindi. Always keep English as the main language.\n" "- MARKDOWN FORMAT:\n" " - Start with a main heading: `# Topic`.\n" " - Use `##` subheadings like `## Concept`, `## Formula`, `## Example`, `## Key Points`.\n" " - Use bullet lists `-` and numbered steps `1.` where helpful.\n" " - Use fenced code blocks ``` for formulas, derivations, or step-by-step working when useful.\n" " - Include at least one simple visual description starting with `Figure:` to help the student imagine the situation.\n" "- TONE & LENGTH:\n" " - Talk to the student directly, for example: \"dekho {student_name}, ...\" or \"so {student_name}, ...\".\n" " - Keep the answer clear, friendly, and roughly 400–500 words.\n" "- CONTEXT:\n" " - Use the previous conversation as context if it helps answer the current doubt.\n" "- EXAMPLE ANSWER STYLE (for reference only, do not repeat this text exactly):\n" " ```markdown\n" " # Newton’s Third Law of Motion\n" "\n" " ## Concept\n" "\n" " Dekho {student_name}, Newton’s third law simply says:\n" "\n" " > **For every action, there is an equal and opposite reaction.**\n" "\n" " That means jab bhi ek object kisi doosre object par **force** lagata hai,\n" " toh second object bhi pehle object par **same magnitude** ka force lagata hai, but **opposite direction** me.\n" " ```\n" ) messages = [{"role": "system", "content": system_msg}] # previous turns for h in history: r = h.get('role') c = h.get('content') if not c: continue if r == 'student': messages.append({"role": "user", "content": c}) elif r == 'ai': messages.append({"role": "assistant", "content": c}) # current doubt messages.append({ "role": "user", "content": ( f"Student name: {student_name}\n" f"Class: {student_class} Section: {student_section}\n\n" f"Doubt: {doubt_text}" ) }) # ---------- CALL GROQ ---------- api_key = os.getenv('GROQ_API_KEY') if not api_key: return jsonify({ "error": "AI explanation failed", "detail": "GROQ_API_KEY is not set" }), 500 headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": "llama-3.1-8b-instant", "messages": messages, "max_tokens": 900, "temperature": 0.6 } try: r = requests.post( "https://api.groq.com/openai/v1/chat/completions", headers=headers, json=payload, timeout=30 ) if not r.ok: return jsonify({ "error": "AI explanation failed", "detail": f"Groq status {r.status_code}: {r.text[:300]}" }), 502 resp = r.json() choices = resp.get("choices") if not isinstance(choices, list) or not choices: return jsonify({ "error": "AI explanation failed", "detail": f"Unexpected Groq response format: {repr(resp)[:300]}" }), 502 first_choice = choices[0] message_obj = first_choice.get("message") or {} ai_text = message_obj.get("content") if not ai_text: return jsonify({ "error": "AI explanation failed", "detail": f"No content in Groq message: {repr(first_choice)[:200]}" }), 502 # ---------- SAFE HISTORY WRITE ---------- new_hist = history[:] # already cleaned list of dicts new_hist.append({'role': 'student', 'content': doubt_text}) new_hist.append({'role': 'ai', 'content': ai_text}) # keep last 10 turns session['DOUBTS_HISTORY'] = new_hist[-10:] return jsonify({ "success": True, "class": student_class, "section": student_section, "doubt": doubt_text, "explanation": ai_text }) except Exception as e: return jsonify({ "error": "AI explanation failed", "detail": str(e) }), 500 @app.route('/doubts/clear-history') def clear_doubts_history(): session.pop('DOUBTS_HISTORY', None) return "cleared" ### LOGOUT @app.route('/logout', methods=['GET', 'POST']) def logout(): session.clear() return redirect(url_for('dashboard')) # ------+------+------+------+------+------+------+------+------ # ### RUN THE APP if __name__ == "__main__": app.run(debug=True, host='0.0.0.0', port=7860)