import json import logging from odoo import http from odoo.http import request from odoo.addons.encoach_api.controllers.base import ( jwt_required, _json_response, _error_response, _get_json_body, _paginate ) _logger = logging.getLogger(__name__) CEFR_ORDER = ['pre_a1', 'a1', 'a2', 'b1', 'b2', 'c1', 'c2'] STEP_DOWN_THRESHOLD = 0.4 PASS_THRESHOLD_DEFAULT = 0.7 def _cefr_index(level): try: return CEFR_ORDER.index(level) except ValueError: return 0 def _next_cefr(level): idx = _cefr_index(level) return CEFR_ORDER[min(idx + 1, len(CEFR_ORDER) - 1)] def _module_to_dict(mod): return { 'id': mod.id, 'name': mod.name, 'cefr_target': mod.cefr_target or '', 'auto_generated': mod.auto_generated, 'completion_criteria': mod.completion_criteria or 'all_resources', 'score_threshold': mod.score_threshold, 'prerequisite_module_id': mod.prerequisite_module_id.id if mod.prerequisite_module_id else None, 'status': mod.status, 'sequence': mod.sequence, 'generation_brief': mod.generation_brief or '', } def _course_to_dict(course, include_modules=True): data = { 'id': course.id, 'name': course.name, 'code': course.code if hasattr(course, 'code') and course.code else '', 'description': course.description if hasattr(course, 'description') else '', 'generation_source': course.generation_source if hasattr(course, 'generation_source') else '', 'progression_model': course.progression_model if hasattr(course, 'progression_model') else 'linear', 'target_band': course.target_band if hasattr(course, 'target_band') else 0.0, 'entity_id': course.entity_id.id if hasattr(course, 'entity_id') and course.entity_id else None, } if include_modules: Module = request.env['encoach.course.module'].sudo() modules = Module.search([('course_id', '=', course.id)], order='sequence asc') data['modules'] = [_module_to_dict(m) for m in modules] return data class EncoachCourseController(http.Controller): # ------------------------------------------------------------------ # POST /api/course/create # ------------------------------------------------------------------ @http.route('/api/course/create', type='http', auth='none', methods=['POST'], csrf=False) @jwt_required def create(self, **kw): try: body = _get_json_body() name = (body.get('name') or '').strip() if not name: return _error_response('name is required', 400) Course = request.env['op.course'].sudo() course_vals = { 'name': name, 'code': body.get('code', name[:20].upper().replace(' ', '_')), } if body.get('subject_id'): course_vals['subject_id'] = int(body['subject_id']) if body.get('description') and hasattr(Course, 'description'): course_vals['description'] = body['description'] if hasattr(Course, 'generation_source'): course_vals['generation_source'] = 'manual' if hasattr(Course, 'entity_id'): user = request.env.user.sudo() if user.entity_ids: course_vals['entity_id'] = user.entity_ids[0].id course = Course.create(course_vals) Module = request.env['encoach.course.module'].sudo() modules_data = body.get('modules', []) for idx, mod_data in enumerate(modules_data): Module.create({ 'name': mod_data.get('name', f'Module {idx + 1}'), 'course_id': course.id, 'cefr_target': mod_data.get('cefr_target', ''), 'completion_criteria': mod_data.get('completion_criteria', 'all_resources'), 'score_threshold': mod_data.get('score_threshold', 0.0), 'sequence': mod_data.get('sequence', (idx + 1) * 10), 'status': 'available' if idx == 0 else 'locked', }) return _json_response(_course_to_dict(course), 201) except Exception as e: _logger.exception('course create failed') return _error_response(str(e), 500) # ------------------------------------------------------------------ # GET /api/course/gap-analysis # ------------------------------------------------------------------ @http.route('/api/course/gap-analysis', type='http', auth='none', methods=['GET'], csrf=False) @jwt_required def gap_analysis(self, **kw): try: uid = request.env.user.id GapProfile = request.env['encoach.gap.profile'].sudo() profile = GapProfile.search([ ('student_id', '=', uid), ], limit=1, order='created_at desc') if not profile: CatSession = request.env['encoach.cat.session'].sudo() session = CatSession.search([ ('student_id', '=', uid), ('status', '=', 'completed'), ], limit=1, order='completed_at desc') if not session: return _error_response('No placement results or gap profile found', 404) Ability = request.env['encoach.student.ability.model'].sudo() abilities = Ability.search([('student_id', '=', uid)]) skill_gaps = {} for ab in abilities: if ab.theta < 0.5: skill_gaps[ab.skill] = { 'theta': ab.theta, 'sem': ab.sem, 'gap_severity': 'high' if ab.theta < -0.5 else 'medium' if ab.theta < 0.0 else 'low', } Attempt = request.env['encoach.student.attempt'].sudo() placement_attempt = Attempt.search([ ('student_id', '=', uid), ('is_placement', '=', True), ], limit=1, order='id desc') question_type_weaknesses = {} topic_weaknesses = {} if placement_attempt: Answer = request.env['encoach.student.answer'].sudo() answers = Answer.search([('attempt_id', '=', placement_attempt.id)]) type_stats = {} topic_stats = {} for ans in answers: q = ans.question_id qt = q.question_type or 'unknown' type_stats.setdefault(qt, {'correct': 0, 'total': 0}) type_stats[qt]['total'] += 1 if ans.is_correct: type_stats[qt]['correct'] += 1 if q.topic_id: tid = q.topic_id.name or str(q.topic_id.id) topic_stats.setdefault(tid, {'correct': 0, 'total': 0}) topic_stats[tid]['total'] += 1 if ans.is_correct: topic_stats[tid]['correct'] += 1 for qt, stats in type_stats.items(): pct = stats['correct'] / stats['total'] if stats['total'] else 0 if pct < 0.6: question_type_weaknesses[qt] = round(pct, 2) for topic, stats in topic_stats.items(): pct = stats['correct'] / stats['total'] if stats['total'] else 0 if pct < 0.6: topic_weaknesses[topic] = round(pct, 2) user = request.env.user.sudo() entity_id = user.entity_ids[0].id if user.entity_ids else False profile = GapProfile.create({ 'student_id': uid, 'source_type': 'placement', 'source_id': session.id, 'skill_gaps': json.dumps(skill_gaps), 'question_type_weaknesses': json.dumps(question_type_weaknesses), 'topic_weaknesses': json.dumps(topic_weaknesses), 'entity_id': entity_id, }) return _json_response({ 'gap_profile_id': profile.id, 'student_id': profile.student_id.id, 'source_type': profile.source_type, 'skill_gaps': json.loads(profile.skill_gaps) if profile.skill_gaps else {}, 'question_type_weaknesses': json.loads(profile.question_type_weaknesses) if profile.question_type_weaknesses else {}, 'topic_weaknesses': json.loads(profile.topic_weaknesses) if profile.topic_weaknesses else {}, 'created_at': profile.created_at, }) except Exception as e: _logger.exception('gap_analysis failed') return _error_response(str(e), 500) # ------------------------------------------------------------------ # POST /api/course/auto-generate # ------------------------------------------------------------------ @http.route('/api/course/auto-generate', type='http', auth='none', methods=['POST'], csrf=False) @jwt_required def auto_generate(self, **kw): try: body = _get_json_body() uid = request.env.user.id GapProfile = request.env['encoach.gap.profile'].sudo() gap_profile_id = body.get('gap_profile_id') if gap_profile_id: profile = GapProfile.browse(int(gap_profile_id)) if not profile.exists(): return _error_response('Gap profile not found', 404) else: profile = GapProfile.search([ ('student_id', '=', uid), ], limit=1, order='created_at desc') if not profile: return _error_response('No gap profile found. Run gap-analysis first.', 404) skill_gaps = json.loads(profile.skill_gaps) if profile.skill_gaps else {} topic_weaknesses = json.loads(profile.topic_weaknesses) if profile.topic_weaknesses else {} Ability = request.env['encoach.student.ability.model'].sudo() abilities = Ability.search([('student_id', '=', uid)]) ability_map = {} for ab in abilities: ability_map[ab.skill] = ab.theta user = request.env.user.sudo() entity_id = user.entity_ids[0].id if user.entity_ids else False Course = request.env['op.course'].sudo() course_vals = { 'name': f"Personalized Course for {user.name}", 'code': f"AUTO_{uid}_{profile.id}", } if hasattr(Course, 'generation_source'): course_vals['generation_source'] = 'auto_gap' if hasattr(Course, 'gap_profile_id'): course_vals['gap_profile_id'] = profile.id if hasattr(Course, 'entity_id'): course_vals['entity_id'] = entity_id course = Course.create(course_vals) Module = request.env['encoach.course.module'].sudo() seq = 10 created_modules = [] sorted_gaps = sorted(skill_gaps.items(), key=lambda x: x[1].get('theta', 0) if isinstance(x[1], dict) else 0) for skill, gap_info in sorted_gaps: theta = gap_info.get('theta', 0) if isinstance(gap_info, dict) else 0 if theta < -0.5: cefr = 'a1' elif theta < 0.0: cefr = 'a2' elif theta < 0.5: cefr = 'b1' else: cefr = 'b2' severity = gap_info.get('gap_severity', 'medium') if isinstance(gap_info, dict) else 'medium' mod = Module.create({ 'name': f"{skill.title()} Improvement ({severity})", 'course_id': course.id, 'cefr_target': cefr, 'auto_generated': True, 'generation_brief': json.dumps({ 'skill': skill, 'theta': theta, 'severity': severity, }), 'completion_criteria': 'score_threshold', 'score_threshold': PASS_THRESHOLD_DEFAULT, 'sequence': seq, 'status': 'available' if seq == 10 else 'locked', }) created_modules.append(mod) seq += 10 for topic_name, accuracy in sorted(topic_weaknesses.items(), key=lambda x: x[1]): mod = Module.create({ 'name': f"Topic Review: {topic_name}", 'course_id': course.id, 'cefr_target': 'b1', 'auto_generated': True, 'generation_brief': json.dumps({ 'topic': topic_name, 'accuracy': accuracy, }), 'completion_criteria': 'all_resources', 'sequence': seq, 'status': 'locked', }) created_modules.append(mod) seq += 10 if not created_modules: Module.create({ 'name': 'General Practice', 'course_id': course.id, 'cefr_target': 'b1', 'auto_generated': True, 'completion_criteria': 'all_resources', 'sequence': 10, 'status': 'available', }) for i in range(1, len(created_modules)): created_modules[i].write({ 'prerequisite_module_id': created_modules[i - 1].id, }) return _json_response(_course_to_dict(course), 201) except Exception as e: _logger.exception('auto_generate failed') return _error_response(str(e), 500) # ------------------------------------------------------------------ # GET /api/course/ # ------------------------------------------------------------------ @http.route('/api/course/', type='http', auth='none', methods=['GET'], csrf=False) @jwt_required def get(self, course_id, **kw): try: Course = request.env['op.course'].sudo() course = Course.browse(course_id) if not course.exists(): return _error_response('Course not found', 404) data = _course_to_dict(course) uid = request.env.user.id Progress = request.env.get('encoach.course.progress') if Progress is not None: Progress = Progress.sudo() for mod in data.get('modules', []): progress_recs = Progress.search([ ('student_id', '=', uid), ('module_id', '=', mod['id']), ]) mod['completed_resources'] = len( progress_recs.filtered(lambda p: p.completed if hasattr(p, 'completed') else False) ) return _json_response(data) except Exception as e: _logger.exception('course get failed') return _error_response(str(e), 500) # ------------------------------------------------------------------ # PUT /api/course/ # ------------------------------------------------------------------ @http.route('/api/course/', type='http', auth='none', methods=['PUT'], csrf=False) @jwt_required def update(self, course_id, **kw): try: body = _get_json_body() Course = request.env['op.course'].sudo() course = Course.browse(course_id) if not course.exists(): return _error_response('Course not found', 404) allowed_fields = {'name', 'description', 'code'} vals = {} for key in allowed_fields: if key in body and hasattr(course, key): vals[key] = body[key] if vals: course.write(vals) module_order = body.get('module_order') if module_order and isinstance(module_order, list): Module = request.env['encoach.course.module'].sudo() for idx, mod_id in enumerate(module_order): mod = Module.browse(int(mod_id)) if mod.exists() and mod.course_id.id == course.id: mod.write({'sequence': (idx + 1) * 10}) return _json_response(_course_to_dict(course)) except Exception as e: _logger.exception('course update failed') return _error_response(str(e), 500) # ------------------------------------------------------------------ # POST /api/course//progress # ------------------------------------------------------------------ @http.route('/api/course//progress', type='http', auth='none', methods=['POST'], csrf=False) @jwt_required def progress(self, course_id, **kw): try: body = _get_json_body() module_id = body.get('module_id') resource_id = body.get('resource_id') completed = body.get('completed', False) if not module_id: return _error_response('module_id is required', 400) Module = request.env['encoach.course.module'].sudo() mod = Module.browse(int(module_id)) if not mod.exists() or mod.course_id.id != course_id: return _error_response('Module not found in this course', 404) uid = request.env.user.id if mod.status == 'locked': return _error_response('Module is locked', 400) if mod.status in ('locked', 'available'): mod.write({'status': 'in_progress'}) Progress = request.env.get('encoach.course.progress') if Progress is not None and resource_id: Progress = Progress.sudo() existing = Progress.search([ ('student_id', '=', uid), ('module_id', '=', mod.id), ('resource_id', '=', int(resource_id)), ], limit=1) if existing: existing.write({'completed': completed}) else: Progress.create({ 'student_id': uid, 'module_id': mod.id, 'resource_id': int(resource_id), 'completed': completed, }) all_modules = Module.search([('course_id', '=', course_id)], order='sequence asc') total_modules = len(all_modules) completed_modules = len(all_modules.filtered(lambda m: m.status == 'completed')) module_status = mod.status if mod.completion_criteria == 'all_resources' and Progress is not None: Progress = request.env['encoach.course.progress'].sudo() mod_progress = Progress.search([ ('student_id', '=', uid), ('module_id', '=', mod.id), ]) all_done = all( getattr(p, 'completed', False) for p in mod_progress ) if mod_progress else False if all_done and mod_progress: mod.write({'status': 'completed'}) module_status = 'completed' completed_modules += 1 self._unlock_next_module(mod, all_modules) overall_pct = round((completed_modules / total_modules) * 100, 1) if total_modules else 0.0 return _json_response({ 'module_id': mod.id, 'module_status': module_status, 'overall_progress_pct': overall_pct, }) except Exception as e: _logger.exception('progress failed') return _error_response(str(e), 500) def _unlock_next_module(self, current_mod, all_modules): """Unlock the next sequential module when the current one is completed.""" Module = request.env['encoach.course.module'].sudo() next_mods = Module.search([ ('course_id', '=', current_mod.course_id.id), ('prerequisite_module_id', '=', current_mod.id), ('status', '=', 'locked'), ]) for nm in next_mods: nm.write({'status': 'available'}) if not next_mods: found_current = False for m in all_modules: if found_current and m.status == 'locked': m.write({'status': 'available'}) break if m.id == current_mod.id: found_current = True # ------------------------------------------------------------------ # POST /api/course//checkpoint # ------------------------------------------------------------------ @http.route('/api/course//checkpoint', type='http', auth='none', methods=['POST'], csrf=False) @jwt_required def checkpoint(self, course_id, **kw): try: body = _get_json_body() module_id = body.get('module_id') score = body.get('score') if not module_id or score is None: return _error_response('module_id and score are required', 400) score = float(score) Module = request.env['encoach.course.module'].sudo() mod = Module.browse(int(module_id)) if not mod.exists() or mod.course_id.id != course_id: return _error_response('Module not found in this course', 404) threshold = mod.score_threshold or PASS_THRESHOLD_DEFAULT passed = score >= threshold all_modules = Module.search([('course_id', '=', course_id)], order='sequence asc') next_module_id = None action = 'continue' if passed: mod.write({'status': 'completed'}) self._unlock_next_module(mod, all_modules) next_mods = Module.search([ ('course_id', '=', course_id), ('status', 'in', ['available', 'locked']), ('sequence', '>', mod.sequence), ], limit=1, order='sequence asc') if next_mods: next_module_id = next_mods[0].id action = 'next_module' else: action = 'course_complete' elif score < STEP_DOWN_THRESHOLD: action = 'remediation' else: action = 'retry' return _json_response({ 'module_id': mod.id, 'score': score, 'threshold': threshold, 'passed': passed, 'next_module_id': next_module_id, 'action': action, }) except Exception as e: _logger.exception('checkpoint failed') return _error_response(str(e), 500) # ------------------------------------------------------------------ # POST /api/course//post-test # ------------------------------------------------------------------ @http.route('/api/course//post-test', type='http', auth='none', methods=['POST'], csrf=False) @jwt_required def post_test(self, course_id, **kw): try: Course = request.env['op.course'].sudo() course = Course.browse(course_id) if not course.exists(): return _error_response('Course not found', 404) Module = request.env['encoach.course.module'].sudo() all_modules = Module.search([('course_id', '=', course_id)]) incomplete = all_modules.filtered(lambda m: m.status != 'completed') if incomplete: return _error_response( f'{len(incomplete)} module(s) not completed. Finish all modules first.', 400 ) subject_id = course.subject_id.id if hasattr(course, 'subject_id') and course.subject_id else False ExamCustom = request.env['encoach.exam.custom'].sudo() exam = None if subject_id: exam = ExamCustom.search([ ('subject_id', '=', subject_id), ('status', '=', 'published'), ], limit=1, order='id desc') if not exam: exam = ExamCustom.search([ ('status', '=', 'published'), ], limit=1, order='id desc') if not exam: return _error_response('No published exam found for post-test', 404) Assignment = request.env['encoach.exam.assignment'].sudo() uid = request.env.user.id existing = Assignment.search([ ('exam_id', '=', exam.id), ('student_id', '=', uid), ('status', '=', 'assigned'), ], limit=1) if existing: return _json_response({'exam_assignment_id': existing.id}) assignment = Assignment.create({ 'exam_id': exam.id, 'student_id': uid, 'status': 'assigned', }) return _json_response({'exam_assignment_id': assignment.id}, 201) except Exception as e: _logger.exception('post_test failed') return _error_response(str(e), 500)