463 lines
19 KiB
Python
463 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
|
import json
|
|
import logging
|
|
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
|
|
|
|
@http.route('/api/mobile/v1/student/profile', type='http', auth='public', methods=['GET'], cors='*')
|
|
@require_student
|
|
def get_profile(self, student, **kwargs):
|
|
try:
|
|
data = {
|
|
"id": student.id,
|
|
"student_id": student.student_id,
|
|
"admission_number": student.admission_number,
|
|
"roll_number": student.roll_number,
|
|
"full_name": student.full_name,
|
|
"email": student.email,
|
|
"mobile": student.mobile,
|
|
"gender": student.gender,
|
|
"date_of_birth": student.date_of_birth,
|
|
"status": student.status,
|
|
"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)
|
|
}
|
|
return json_response(True, data=data)
|
|
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
|
|
}
|
|
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),
|
|
], 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/<int:announcement_id>', 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)
|