import base64 import json import logging import mimetypes import re from datetime import datetime, date # pyrefly: ignore [missing-import] from odoo import http # pyrefly: ignore [missing-import] from odoo.http import request _logger = logging.getLogger(__name__) def json_response(success, data=None, error_code=None, error_message=None, status=200): if success: body = { "success": True, "data": data or {} } else: body = { "success": False, "error": { "code": error_code or "ERROR", "message": error_message or "An error occurred" } } return request.make_response( json.dumps(body, default=str), headers=[('Content-Type', 'application/json')], status=status ) def require_student(func): def wrapper(self, *args, **kwargs): if request.env.user._is_public(): return json_response(False, error_code="AUTH_REQUIRED", error_message="Authentication required", status=401) student = request.env.user.sudo().student_id if not student: return json_response(False, error_code="STUDENT_NOT_LINKED", error_message="No student profile is linked to this account", status=403) return func(self, student, *args, **kwargs) return wrapper def format_float_time(float_time): if float_time is None or float_time is False: return "" hours = int(float_time) minutes = int(round((float_time - hours) * 60)) if minutes == 60: minutes = 0 hours += 1 hours = hours % 24 am_pm = 'AM' if hours < 12 else 'PM' display_hours = hours % 12 if display_hours == 0: display_hours = 12 return f"{display_hours:02d}:{minutes:02d} {am_pm}" class MobileStudent(http.Controller): def _strip_html(self, text): if not text: return "" clean = re.sub(r'<[^>]+>', ' ', str(text)) return " ".join(clean.split()).strip() def _format_announcement(self, a): title = a.title or getattr(a, 'name', '') or 'Announcement' raw_content = a.content or '' summary = self._strip_html(raw_content)[:140] return { "id": a.id, "title": title, "name": title, "date": str(a.date_published or ''), "date_published": str(a.date_published or ''), "expiry_date": str(a.expiry_date) if a.expiry_date else None, "summary": summary, "content": raw_content, "type": a.announcement_type or 'notice', "priority": a.priority or '0', "author": a.author_id.name if a.author_id else None, "has_attachment": bool(a.circular_file), "circular_filename": a.circular_filename or None, } def _get_rel(self, record): if record: return {"id": record.id, "name": record.name} return None def _format_profile(self, student): return { "id": student.id, "student_id": student.student_id, "admission_number": student.admission_number, "registration_number": student.registration_number or None, "roll_number": student.roll_number or None, "full_name": student.full_name, "first_name": student.first_name, "middle_name": student.middle_name or None, "last_name": student.last_name, "email": student.email, "mobile": student.mobile, "phone": student.phone or None, "gender": student.gender, "date_of_birth": str(student.date_of_birth) if student.date_of_birth else None, "blood_group": student.blood_group or None, "emergency_contact": student.emergency_contact or None, "emergency_phone": student.emergency_phone or None, "street": student.street or None, "street2": student.street2 or None, "city": student.city or None, "zip": student.zip or None, "state": self._get_rel(student.state_id), "country": self._get_rel(student.country_id), "guardian": { "name": student.guardian_name or None, "relation": student.guardian_relation or None, "phone": student.guardian_phone or None, "mobile": student.guardian_mobile or None, "email": student.guardian_email or None, "address": student.guardian_address or None, }, "status": student.status, "admission_date": str(student.admission_date) if student.admission_date else None, "enrollment_date": str(student.enrollment_date) if student.enrollment_date else None, "profile_completion": getattr(student, 'profile_completion', 0), "has_photo": bool(student.image), "campus": self._get_rel(student.campus_id), "department": self._get_rel(student.department_id), "program": self._get_rel(student.program_id), "level": self._get_rel(student.level_id), "academic_year": self._get_rel(student.academic_year_id), "academic_term": self._get_rel(student.academic_term_id), "batch": self._get_rel(student.batch_id), "semester": self._get_rel(student.semester_id), "section": self._get_rel(student.section_id), "editable_fields": [ "email", "mobile", "phone", "blood_group", "emergency_contact", "emergency_phone", "street", "street2", "city", "zip", "image" ], "protected_fields": [ "student_id", "admission_number", "registration_number", "roll_number", "campus", "department", "program", "level", "academic_year", "academic_term", "batch", "semester", "section", "status" ] } @http.route('/api/mobile/v1/student/profile', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_profile(self, student, **kwargs): try: return json_response(True, data={"profile": self._format_profile(student), **self._format_profile(student)}) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/profile', type='http', auth='public', methods=['PATCH', 'POST'], cors='*', csrf=False) @require_student def update_profile(self, student, **kwargs): try: raw_body = request.httprequest.data vals = {} if raw_body: try: vals = json.loads(raw_body.decode('utf-8')) except Exception: vals = {} if not vals: vals = dict(kwargs) # Check protected fields protected = [ 'student_id', 'admission_number', 'registration_number', 'roll_number', 'program', 'program_id', 'department', 'department_id', 'campus', 'campus_id', 'level', 'level_id', 'academic_year', 'academic_year_id', 'academic_term', 'academic_term_id', 'batch', 'batch_id', 'semester', 'semester_id', 'section', 'section_id', 'status', 'company_id', 'admission_date', 'enrollment_date', 'graduation_date', 'leave_date' ] for p in protected: if p in vals: return json_response( False, error_code="PROTECTED_FIELD_MODIFICATION", error_message=f"Modifying protected academic or administrative field '{p}' is strictly prohibited.", status=400 ) try: student.sudo().update_student_profile(vals) except Exception as e: return json_response(False, error_code="VALIDATION_ERROR", error_message=str(e), status=400) return json_response(True, data={"profile": self._format_profile(student), **self._format_profile(student)}) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) def _get_attendance_summary(self, student, from_date=None, to_date=None): domain = [('student_id', '=', student.id)] if from_date: domain.append(('date', '>=', from_date)) if to_date: domain.append(('date', '<=', to_date)) records = request.env['education.attendance'].search(domain) total_sessions = len(records) present = len([a for a in records if a.status == 'present']) absent = len([a for a in records if a.status == 'absent']) late = len([a for a in records if a.status == 'late']) medical_leave = len([a for a in records if a.status == 'medical_leave']) on_duty = len([a for a in records if a.status == 'on_duty']) attended = present + on_duty + late + medical_leave percentage = round((attended / total_sessions * 100.0), 1) if total_sessions > 0 else 100.0 return { "percentage": percentage, "present": present, "absent": absent, "late": late, "medical_leave": medical_leave, "on_duty": on_duty } def _get_timetable_slots(self, student, date_filter=None, from_date=None, to_date=None): if not student.program_id or not student.academic_year_id: return [] domain = [ ('program_id', '=', student.program_id.id), ('academic_year_id', '=', student.academic_year_id.id), ('active', '=', True), ('state', '!=', 'cancelled'), ] if student.company_id: domain.append(('company_id', '=', student.company_id.id)) if student.academic_term_id: domain += ['|', ('term_id', '=', False), ('term_id', '=', student.academic_term_id.id)] else: domain += [('term_id', '=', False)] if student.batch_id: domain += ['|', ('batch_id', '=', False), ('batch_id', '=', student.batch_id.id)] else: domain += [('batch_id', '=', False)] if hasattr(student, 'semester_id') and student.semester_id: domain += ['|', ('semester_id', '=', False), ('semester_id', '=', student.semester_id.id)] else: domain += [('semester_id', '=', False)] if hasattr(student, 'section_id') and student.section_id: domain += ['|', ('section_id', '=', False), ('section_id', '=', student.section_id.id)] else: domain += [('section_id', '=', False)] if date_filter: domain += [('date', '=', date_filter)] if from_date: domain += [('date', '>=', from_date)] if to_date: domain += [('date', '<=', to_date)] env = student.env if student else request.env slots = env['education.timetable.slot'].sudo().search(domain, order="date asc, start_time asc") return [{ "id": s.id, "date": s.date, "day": s.date.strftime('%A') if s.date else '', "start_time": format_float_time(s.start_time), "end_time": format_float_time(s.end_time), "subject": self._get_rel(s.subject_id), "faculty": self._get_rel(s.faculty_id), "room": self._get_rel(s.room_id) } for s in slots] @http.route('/api/mobile/v1/student/dashboard', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_dashboard(self, student, **kwargs): try: today = date.today() # Upcoming Exams exams = request.env['education.exam'].sudo().search_read([ ('state', 'in', ['scheduled', 'ongoing']), ('program_id', '=', student.program_id.id), ('academic_year_id', '=', student.academic_year_id.id), ], fields=['id', 'name', 'state', 'start_date'], order='start_date asc', limit=5) # Announcements (filtered by audience, state, and publication window) ann_records, _ = request.env['education.announcement'].get_announcements_for_student(student, limit=5) announcements = [self._format_announcement(a) for a in ann_records] # Recent Results results = request.env['education.exam.result'].sudo().search([ ('student_id', '=', student.id), ('published', '=', True) ], order='id desc', limit=5) results_data = [] for r in results: results_data.append({ "id": r.id, "exam": self._get_rel(r.exam_id), "percentage": r.percentage, "grade": r.grade_id.name if r.grade_id else None, "status": r.result_status }) data = { "student": { "id": student.id, "full_name": student.full_name, "program": self._get_rel(student.program_id) }, "attendance": self._get_attendance_summary(student), "today_timetable": self._get_timetable_slots(student, date_filter=today), "upcoming_exams": exams, "recent_results": results_data, "announcements": announcements, "fees": { "total_fees": float(student.total_fees), "total_paid": float(student.total_paid), "outstanding_balance": float(student.outstanding_amount), "overdue_amount": float(student.overdue_amount), "payment_status": student.payment_status or "no_dues", "next_due_date": str(student.next_due_date) if student.next_due_date else None, "currency_symbol": student.currency_id.symbol or student.currency_id.name or '₹' } } return json_response(True, data=data) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/attendance', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_attendance(self, student, **kwargs): try: from_date_str = kwargs.get('from_date') to_date_str = kwargs.get('to_date') from_date = None to_date = None if from_date_str: from_date = datetime.strptime(from_date_str, '%Y-%m-%d').date() if to_date_str: to_date = datetime.strptime(to_date_str, '%Y-%m-%d').date() if from_date and to_date and from_date > to_date: return json_response(False, error_code="INVALID_DATE_RANGE", error_message="'from_date' cannot be later than 'to_date'", status=400) summary = self._get_attendance_summary(student, from_date, to_date) domain = [('student_id', '=', student.id)] if from_date: domain.append(('date', '>=', from_date)) if to_date: domain.append(('date', '<=', to_date)) records = request.env['education.attendance'].search_read( domain, fields=['id', 'date', 'status', 'remarks'], order='date desc, id desc', limit=100 ) formatted_records = [] for r in records: formatted_records.append({ "id": r["id"], "date": str(r["date"]), "status": r["status"], "remarks": r.get("remarks") or "" }) return json_response(True, data={ "summary": summary, "records": formatted_records }) except ValueError: return json_response(False, error_code="INVALID_DATE", error_message="Dates must be in YYYY-MM-DD format", status=400) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/timetable', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_timetable(self, student, **kwargs): try: date_filter = kwargs.get('date') from_date = kwargs.get('from_date') to_date = kwargs.get('to_date') if date_filter: try: datetime.strptime(date_filter, '%Y-%m-%d') except ValueError: return json_response(False, error_code="INVALID_DATE", error_message="Date must be in YYYY-MM-DD format", status=400) if from_date: try: datetime.strptime(from_date, '%Y-%m-%d') except ValueError: return json_response(False, error_code="INVALID_DATE", error_message="from_date must be in YYYY-MM-DD format", status=400) if to_date: try: datetime.strptime(to_date, '%Y-%m-%d') except ValueError: return json_response(False, error_code="INVALID_DATE", error_message="to_date must be in YYYY-MM-DD format", status=400) if from_date and to_date and from_date > to_date: return json_response(False, error_code="INVALID_DATE_RANGE", error_message="from_date must be before or equal to to_date", status=400) slots = self._get_timetable_slots(student, date_filter=date_filter, from_date=from_date, to_date=to_date) return json_response(True, data={"slots": slots}) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/exams', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_exams(self, student, **kwargs): try: exams = request.env['education.exam'].sudo().search([ ('program_id', '=', student.program_id.id), ('academic_year_id', '=', student.academic_year_id.id), ('state', 'in', ['scheduled', 'ongoing', 'completed', 'results_published']), ], order='start_date desc') exams_data = [{ "id": exam.id, "name": exam.name, "state": exam.state, "start_date": exam.start_date, "end_date": exam.end_date, "program": self._get_rel(exam.program_id), "academic_year": self._get_rel(exam.academic_year_id) } for exam in exams] return json_response(True, data={"exams": exams_data}) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/results', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_results(self, student, **kwargs): try: results = request.env['education.exam.result'].sudo().search([ ('student_id', '=', student.id), ('published', '=', True) ], order='id desc') results_data = [] for r in results: lines = [] for line in r.line_ids: lines.append({ "subject": self._get_rel(line.subject_id), "max_marks": line.max_marks, "pass_marks": line.pass_marks, "obtained_marks": line.obtained_marks, "percentage": line.percentage, "grade": line.grade_id.name if line.grade_id else None, "status": line.status }) results_data.append({ "id": r.id, "exam": self._get_rel(r.exam_id), "total_max_marks": r.total_max_marks, "total_obtained_marks": r.total_obtained_marks, "percentage": r.percentage, "grade": r.grade_id.name if r.grade_id else None, "status": r.result_status, "published_date": r.published_date, "lines": lines }) return json_response(True, data={"results": results_data}) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/announcements', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_announcements(self, student, **kwargs): try: page = int(kwargs.get('page', 1) or 1) limit = int(kwargs.get('limit', 20) or 20) if page < 1: page = 1 if limit < 1 or limit > 100: limit = 20 offset = (page - 1) * limit ann_type = kwargs.get('type') if ann_type and ann_type not in ['notice', 'circular', 'event']: ann_type = None records, total_count = request.env['education.announcement'].get_announcements_for_student( student, limit=limit, offset=offset, announcement_type=ann_type ) announcements_data = [self._format_announcement(a) for a in records] return json_response(True, data={ "announcements": announcements_data, "total_count": total_count, "page": page, "limit": limit }) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/announcements/', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_announcement_detail(self, student, announcement_id, **kwargs): try: ann = request.env['education.announcement'].sudo().browse(announcement_id) if not ann.exists() or not ann.active or ann.state != 'published': return json_response(False, error_code="NOT_FOUND", error_message="Announcement not found or unavailable", status=404) # Date window validation today = date.today() if ann.date_published > today: return json_response(False, error_code="NOT_FOUND", error_message="Announcement not found or unavailable", status=404) if ann.expiry_date and ann.expiry_date < today: return json_response(False, error_code="NOT_FOUND", error_message="Announcement has expired", status=404) # Multi-company isolation if student.company_id and ann.company_id and ann.company_id.id != student.company_id.id: return json_response(False, error_code="ACCESS_DENIED", error_message="You do not have access to this announcement", status=403) # Audience and cohort targeting if not ann.check_student_audience(student): return json_response(False, error_code="ACCESS_DENIED", error_message="You do not have access to this announcement", status=403) return json_response(True, data={ "announcement": self._format_announcement(ann) }) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/fees/summary', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_fee_summary(self, student, **kwargs): try: curr = student.currency_id or student.company_id.currency_id data = { "total_fees": float(student.total_fees), "total_paid": float(student.total_paid), "outstanding_balance": float(student.outstanding_amount), "overdue_amount": float(student.overdue_amount), "payment_status": student.payment_status or "no_dues", "next_due_date": str(student.next_due_date) if student.next_due_date else None, "last_payment_date": str(student.last_payment_date) if student.last_payment_date else None, "last_receipt_number": student.last_receipt_number or None, "currency": { "id": curr.id if curr else 1, "name": curr.name if curr else "INR", "symbol": curr.symbol if curr else "₹" } } return json_response(True, data=data) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/fees/invoices', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_fee_invoices(self, student, **kwargs): try: status_filter = kwargs.get('status') page = int(kwargs.get('page', 1) or 1) limit = int(kwargs.get('limit', 20) or 20) if page < 1: page = 1 if limit < 1 or limit > 100: limit = 20 offset = (page - 1) * limit domain = [ ('student_id', '=', student.id), ('active', '=', True), ] if student.company_id: domain.append(('company_id', '=', student.company_id.id)) if status_filter in ['unpaid', 'partial', 'paid', 'overdue']: domain.append(('payment_status', '=', status_filter)) total_count = request.env['education.fee.assignment'].sudo().search_count(domain) assignments = request.env['education.fee.assignment'].sudo().search( domain, order='due_date asc, id desc', limit=limit, offset=offset ) curr = student.currency_id or student.company_id.currency_id curr_dict = { "id": curr.id if curr else 1, "name": curr.name if curr else "INR", "symbol": curr.symbol if curr else "₹" } invoices = [] for a in assignments: invoices.append({ "id": a.id, "invoice_number": a.name or f"INV/{a.id:05d}", "title": a.fee_structure_id.name or "Semester Fees", "academic_year": self._get_rel(a.academic_year_id), "academic_term": self._get_rel(a.academic_term_id), "semester": self._get_rel(a.semester_id), "assignment_date": str(a.assignment_date or ''), "due_date": str(a.due_date or ''), "amount": float(a.amount), "amount_paid": float(a.amount_paid), "amount_due": float(a.amount_due), "payment_status": a.payment_status, "state": a.state, "currency": curr_dict }) return json_response(True, data={ "invoices": invoices, "total_count": total_count, "page": page, "limit": limit }) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/fees/invoices/', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_fee_invoice_detail(self, student, invoice_id, **kwargs): try: assignment = request.env['education.fee.assignment'].sudo().browse(invoice_id) if not assignment.exists() or not assignment.active: return json_response(False, error_code="NOT_FOUND", error_message="Invoice not found", status=404) # Security isolation: must match authenticated student if assignment.student_id.id != student.id: return json_response(False, error_code="ACCESS_DENIED", error_message="You do not have access to this invoice", status=403) # Company isolation if student.company_id and assignment.company_id and assignment.company_id.id != student.company_id.id: return json_response(False, error_code="ACCESS_DENIED", error_message="You do not have access to this invoice", status=403) curr = assignment.currency_id or student.currency_id or student.company_id.currency_id curr_dict = { "id": curr.id if curr else 1, "name": curr.name if curr else "INR", "symbol": curr.symbol if curr else "₹" } lines = [] if assignment.fee_structure_id and assignment.fee_structure_id.line_ids: for line in assignment.fee_structure_id.line_ids: lines.append({ "id": line.id, "name": line.fee_type_id.name, "category": line.fee_type_id.category_id.name if line.fee_type_id.category_id else None, "amount": float(line.amount), "mandatory": line.mandatory }) payments = [] for p in assignment.payment_ids.filtered(lambda x: x.status == 'confirmed'): payments.append({ "id": p.id, "receipt_number": p.receipt_number, "payment_date": str(p.payment_date or ''), "payment_method": p.payment_method, "payment_method_label": dict(p._fields['payment_method'].selection).get(p.payment_method, p.payment_method), "amount": float(p.amount), "reference_number": p.reference_number }) data = { "invoice": { "id": assignment.id, "invoice_number": assignment.name or f"INV/{assignment.id:05d}", "title": assignment.fee_structure_id.name or "Semester Fees", "academic_year": self._get_rel(assignment.academic_year_id), "academic_term": self._get_rel(assignment.academic_term_id), "semester": self._get_rel(assignment.semester_id), "assignment_date": str(assignment.assignment_date or ''), "due_date": str(assignment.due_date or ''), "amount": float(assignment.amount), "amount_paid": float(assignment.amount_paid), "amount_due": float(assignment.amount_due), "payment_status": assignment.payment_status, "state": assignment.state, "currency": curr_dict, "lines": lines, "payments": payments } } return json_response(True, data=data) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/fees/payments', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_fee_payments(self, student, **kwargs): try: page = int(kwargs.get('page', 1) or 1) limit = int(kwargs.get('limit', 20) or 20) if page < 1: page = 1 if limit < 1 or limit > 100: limit = 20 offset = (page - 1) * limit domain = [ ('student_id', '=', student.id), ('status', '=', 'confirmed'), ] if student.company_id: domain.append(('company_id', '=', student.company_id.id)) total_count = request.env['education.student.payment'].sudo().search_count(domain) payments = request.env['education.student.payment'].sudo().search( domain, order='payment_date desc, id desc', limit=limit, offset=offset ) curr = student.currency_id or student.company_id.currency_id curr_dict = { "id": curr.id if curr else 1, "name": curr.name if curr else "INR", "symbol": curr.symbol if curr else "₹" } payments_data = [] for p in payments: payments_data.append({ "id": p.id, "receipt_number": p.receipt_number, "payment_date": str(p.payment_date or ''), "payment_method": p.payment_method, "payment_method_label": dict(p._fields['payment_method'].selection).get(p.payment_method, p.payment_method), "amount": float(p.amount), "reference_number": p.reference_number, "remarks": p.remarks, "status": p.status, "invoice": { "id": p.fee_assignment_id.id, "invoice_number": p.fee_assignment_id.name } if p.fee_assignment_id else None, "currency": curr_dict }) return json_response(True, data={ "payments": payments_data, "total_count": total_count, "page": page, "limit": limit }) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) # ------------------------------------------------------------------------- # Student Documents API # ------------------------------------------------------------------------- @http.route('/api/mobile/v1/student/documents', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_documents(self, student, **kwargs): try: docs = request.env['education.student.document'].sudo().search([ ('student_id', '=', student.id) ], order='upload_date desc, id desc') data = [{ "id": doc.id, "name": doc.name, "category": doc.category, "category_label": dict(doc._fields['category'].selection).get(doc.category, doc.category), "file_name": doc.file_name, "file_size": doc.file_size, "mime_type": doc.mime_type, "verification_status": doc.verification_status, "upload_date": str(doc.upload_date or ''), "verification_date": str(doc.verification_date or '') if doc.verification_date else None, "rejection_reason": doc.rejection_reason or None, } for doc in docs] return json_response(True, data={"documents": data}) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/documents', type='http', auth='public', methods=['POST'], cors='*', csrf=False) @require_student def upload_document(self, student, **kwargs): try: raw_body = request.httprequest.data payload = {} if raw_body: try: payload = json.loads(raw_body.decode('utf-8')) except Exception: payload = {} if not payload: payload = dict(kwargs) file_content = None filename = payload.get('file_name') if 'file' in request.httprequest.files: file_obj = request.httprequest.files['file'] filename = filename or file_obj.filename file_content = base64.b64encode(file_obj.read()).decode('utf-8') elif payload.get('file'): file_content = payload.get('file') if not file_content: return json_response(False, error_code="MISSING_FILE", error_message="Document file is required.", status=400) title = payload.get('name') or payload.get('title') if not title: title = filename or 'Student Document' category = payload.get('category') or 'academic' valid_categories = ['identity', 'academic', 'transfer', 'address', 'other'] if category not in valid_categories: category = 'other' mime_type = mimetypes.guess_type(filename or '')[0] or 'application/octet-stream' doc = request.env['education.student.document'].sudo().create({ 'name': title, 'student_id': student.id, 'category': category, 'document_file': file_content, 'file_name': filename or f"doc_{student.id}.pdf", 'mime_type': mime_type, 'verification_status': 'submitted', }) return json_response(True, data={ "document": { "id": doc.id, "name": doc.name, "category": doc.category, "category_label": dict(doc._fields['category'].selection).get(doc.category, doc.category), "file_name": doc.file_name, "file_size": doc.file_size, "mime_type": doc.mime_type, "verification_status": doc.verification_status, "upload_date": str(doc.upload_date or '') } }, status=201) except Exception as e: return json_response(False, error_code="UPLOAD_ERROR", error_message=str(e), status=400) @http.route('/api/mobile/v1/student/documents/', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_document_detail(self, student, document_id, **kwargs): try: doc = request.env['education.student.document'].sudo().browse(document_id) if not doc.exists(): return json_response(False, error_code="NOT_FOUND", error_message="Document not found.", status=404) if doc.student_id.id != student.id: return json_response(False, error_code="ACCESS_DENIED", error_message="Unauthorized access to document.", status=403) return json_response(True, data={ "document": { "id": doc.id, "name": doc.name, "category": doc.category, "category_label": dict(doc._fields['category'].selection).get(doc.category, doc.category), "file_name": doc.file_name, "file_size": doc.file_size, "mime_type": doc.mime_type, "verification_status": doc.verification_status, "upload_date": str(doc.upload_date or ''), "verification_date": str(doc.verification_date or '') if doc.verification_date else None, "rejection_reason": doc.rejection_reason or None, } }) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/documents//download', type='http', auth='public', methods=['GET'], cors='*') @require_student def download_document(self, student, document_id, **kwargs): try: doc = request.env['education.student.document'].sudo().browse(document_id) if not doc.exists(): return json_response(False, error_code="NOT_FOUND", error_message="Document not found.", status=404) if doc.student_id.id != student.id: return json_response(False, error_code="ACCESS_DENIED", error_message="Unauthorized access to document.", status=403) if not doc.document_file: return json_response(False, error_code="NOT_FOUND", error_message="Document content is empty.", status=404) file_data = base64.b64decode(doc.document_file) mime_type = doc.mime_type or 'application/octet-stream' filename = doc.file_name or f"document_{doc.id}" headers = [ ('Content-Type', mime_type), ('Content-Disposition', f'inline; filename="{filename}"'), ('Content-Length', str(len(file_data))), ] return request.make_response(file_data, headers=headers) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) # ------------------------------------------------------------------------- # Student Certificates API # ------------------------------------------------------------------------- @http.route('/api/mobile/v1/student/certificates', type='http', auth='public', methods=['GET'], cors='*') @require_student def get_certificates(self, student, **kwargs): try: certs = request.env['education.student.certificate'].sudo().search([ ('student_id', '=', student.id) ], order='id desc') data = [{ "id": c.id, "name": c.name or f"REQ-{c.id:04d}", "certificate_type": c.certificate_type, "certificate_type_label": dict(c._fields['certificate_type'].selection).get(c.certificate_type, c.certificate_type), "state": c.state, "request_date": str(c.request_date or ''), "issue_date": str(c.issue_date or '') if c.issue_date else None, "expiry_date": str(c.expiry_date or '') if c.expiry_date else None, "file_name": c.file_name or None, "remarks": c.remarks or None, "rejection_reason": c.rejection_reason or None, "has_file": bool(c.document_file), } for c in certs] return json_response(True, data={"certificates": data}) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500) @http.route('/api/mobile/v1/student/certificates/request', type='http', auth='public', methods=['POST'], cors='*', csrf=False) @require_student def request_certificate(self, student, **kwargs): try: raw_body = request.httprequest.data payload = {} if raw_body: try: payload = json.loads(raw_body.decode('utf-8')) except Exception: payload = {} if not payload: payload = dict(kwargs) cert_type = payload.get('certificate_type') valid_types = ['bonafide', 'id_card', 'course_completion', 'conduct'] if not cert_type or cert_type not in valid_types: return json_response(False, error_code="INVALID_TYPE", error_message=f"Invalid certificate type. Allowed: {', '.join(valid_types)}", status=400) if student.status not in ('active', 'admitted'): return json_response(False, error_code="INELIGIBLE_STATUS", error_message="Only active or admitted students may request certificates.", status=400) remarks = payload.get('remarks') or '' cert = request.env['education.student.certificate'].sudo().create({ 'student_id': student.id, 'certificate_type': cert_type, 'remarks': remarks, 'state': 'requested', }) return json_response(True, data={ "certificate": { "id": cert.id, "name": cert.name or f"REQ-{cert.id:04d}", "certificate_type": cert.certificate_type, "certificate_type_label": dict(cert._fields['certificate_type'].selection).get(cert.certificate_type, cert.certificate_type), "state": cert.state, "request_date": str(cert.request_date or ''), "remarks": cert.remarks, } }, status=201) except Exception as e: return json_response(False, error_code="REQUEST_FAILED", error_message=str(e), status=400) @http.route('/api/mobile/v1/student/certificates//download', type='http', auth='public', methods=['GET'], cors='*') @require_student def download_certificate(self, student, certificate_id, **kwargs): try: cert = request.env['education.student.certificate'].sudo().browse(certificate_id) if not cert.exists(): return json_response(False, error_code="NOT_FOUND", error_message="Certificate not found.", status=404) if cert.student_id.id != student.id: return json_response(False, error_code="ACCESS_DENIED", error_message="Unauthorized access to certificate.", status=403) if cert.state != 'issued': return json_response(False, error_code="NOT_ISSUED", error_message="Certificate has not been issued yet.", status=400) if not cert.document_file: report_ref = None if cert.certificate_type == 'bonafide': report_ref = 'education_reports.action_report_bonafide_certificate' elif cert.certificate_type == 'id_card': report_ref = 'education_reports.action_report_student_id_card' if report_ref and request.env.ref(report_ref, raise_if_not_found=False): try: pdf_content, _ = request.env['ir.actions.report'].sudo()._render_qweb_pdf(report_ref, student.ids) if pdf_content: cert.sudo().write({ 'document_file': base64.b64encode(pdf_content), 'file_name': f"certificate_{cert.name or cert.id}.pdf" }) except Exception: pass if not cert.document_file: return json_response(False, error_code="NOT_FOUND", error_message="Certificate document file is not yet available.", status=404) file_data = base64.b64decode(cert.document_file) filename = cert.file_name or f"certificate_{cert.id}.pdf" headers = [ ('Content-Type', 'application/pdf'), ('Content-Disposition', f'inline; filename="{filename}"'), ('Content-Length', str(len(file_data))), ] return request.make_response(file_data, headers=headers) except Exception as e: return json_response(False, error_code="SERVER_ERROR", error_message=str(e), status=500)