Files
encoach_backend_v4/custom_addons/encoach_taxonomy/controllers/taxonomy.py
Yamen Ahmad 6ec68160c8 feat: institutional + support + training admin sections (backend + frontend)
Ship three fully-wired admin areas end-to-end with APIs, seeds, tests and docs.

Backend (new `encoach_lms_api` addon + existing addons):
- Institutional: academic years/terms, departments, admission registers & admissions,
  courses/batches, lessons, fees (terms + student fees + invoicing with income-account
  auto-wiring), gradebook (assignments/grades), library, facilities (encoach.asset),
  student leave, result templates + marksheets (incl. delete-with-cascade).
- Support: `encoach.ticket` model + CRUD/assignee routes; payment records derived
  from `op.student.fees.details` and `account.move`; platform settings backed by
  `encoach.code` and `ir.config_parameter` (packages + grading config).
- Training: `encoach.vocab.item` + `encoach.grammar.rule` (plus progress models)
  with CRUD, pagination, search/level filters, and upsert-style progress endpoints.
  Odoo 19 compatibility: `_sql_constraints` replaced with `@api.constrains`;
  `ValidationError`/`UserError` mapped to HTTP 400.

Frontend:
- Rewire institutional admin pages (Academic Year Manager, Admissions, Courses,
  Lessons, Fees, Gradebook, Library, Facilities, Student Leave, Marksheets,
  Taxonomy, Resources) to real APIs with React Query invalidation and dialogs.
- New typed services: `payments.service.ts`, `platformSettings.service.ts`,
  `training.service.ts`. Updated `fees/gradebook/lms/courseware/taxonomy/
  resources/student-progress/generation` services + related types.
- Rewrite `VocabularyPage`, `GrammarPage`, `PaymentRecordPage`, `SettingsPage`,
  `TicketsPage` to consume live data with search/filter/progress/CRUD flows.
- New shared components: `TaxonomyCascade`, `MaterialViewer`, `teacher/TeacherLibrary`.
- Favicons/branding assets and misc. UX polish across teacher/student pages.

Tooling & QA:
- Seeders: `seed_demo.py`, `seed_demo_data.py`, `seed_institutional.py` (idempotent,
  covers institutional + support + training fixtures incl. income-account wiring).
- API write-flow test suites: `test_write_flows.py` (institutional),
  `test_support_flows.py` (support), `test_training_flows.py` (training),
  `test_ai_full.py`. All suites pass end-to-end.
- Docs: add `docs/PROJECT_SUMMARY.md` with per-section scope, artifacts and QA.
- `.gitignore`: ignore `pgdata_bak_*/`, `frontend/.vite/`, `frontend/dist/`,
  `frontend/node_modules/`.

Made-with: Cursor
2026-04-19 03:13:23 +04:00

523 lines
20 KiB
Python

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__)
TAXONOMY_MODELS = {
'subject': 'encoach.subject',
'domain': 'encoach.domain',
'topic': 'encoach.topic',
'learning_objective': 'encoach.learning.objective',
}
PARENT_FIELD_MAP = {
'domain': 'subject_id',
'topic': 'domain_id',
'learning_objective': 'topic_id',
}
def _subject_dict(s):
return {
'id': s.id,
'name': s.name,
'code': s.code or '',
'description': s.description or '',
'is_active': s.active,
'active': s.active,
'domain_count': len(s.domain_ids),
'topic_count': sum(len(d.topic_ids) for d in s.domain_ids),
'course_count': s.course_count if hasattr(s, 'course_count') else 0,
'resource_count': s.resource_count if hasattr(s, 'resource_count') else 0,
'mastery_threshold': 80,
'grading_scale': 'percentage',
'diagnostic_config': {
'questions_per_domain': 5,
'total_question_cap': 30,
'time_limit_minutes': 45,
'starting_difficulty': 'medium',
'mastery_per_correct': 10,
'mastery_per_incorrect': -5,
},
}
def _domain_dict(d):
return {
'id': d.id,
'name': d.name,
'code': '',
'subject_id': d.subject_id.id,
'subject_name': d.subject_id.name,
'sequence': d.sequence,
'description': d.description or '',
'topic_count': len(d.topic_ids),
}
def _topic_dict(t):
return {
'id': t.id,
'name': t.name,
'code': '',
'domain_id': t.domain_id.id,
'domain_name': t.domain_id.name,
'description': t.description or '',
'difficulty_level': 'medium',
'estimated_hours': 1,
'prerequisite_ids': [],
'prerequisite_names': [],
'question_type_weights': {},
'objective_count': len(t.learning_objective_ids),
'resource_count': t.resource_count if hasattr(t, 'resource_count') else 0,
'chapter_count': t.chapter_count if hasattr(t, 'chapter_count') else 0,
}
def _objective_dict(o):
return {
'id': o.id,
'name': o.name,
'topic_id': o.topic_id.id,
'bloom_level': o.bloom_level or 'remember',
'sequence': o.sequence if hasattr(o, 'sequence') else 0,
}
class EncoachTaxonomyController(http.Controller):
# ══════════════════════════════════════════════════════════════════
# SUBJECTS /api/subjects
# ══════════════════════════════════════════════════════════════════
@http.route(['/api/subjects', '/api/taxonomy/subjects'], type='http',
auth='public', methods=['GET'], csrf=False)
@jwt_required
def list_subjects(self, **kw):
try:
recs = request.env['encoach.subject'].sudo().search([])
items = [_subject_dict(s) for s in recs]
return _json_response({'data': items, 'subjects': items})
except Exception as e:
_logger.exception('list_subjects')
return _error_response(str(e), 500)
@http.route('/api/subjects/<int:sid>', type='http',
auth='public', methods=['GET'], csrf=False)
@jwt_required
def get_subject(self, sid, **kw):
try:
rec = request.env['encoach.subject'].sudo().browse(sid)
if not rec.exists():
return _error_response('Subject not found', 404)
return _json_response({'data': _subject_dict(rec)})
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/subjects', type='http', auth='public',
methods=['POST'], csrf=False)
@jwt_required
def create_subject(self, **kw):
try:
body = _get_json_body()
vals = {'name': body.get('name', ''), 'code': body.get('code', '')}
if body.get('description'):
vals['description'] = body['description']
rec = request.env['encoach.subject'].sudo().create(vals)
return _json_response(_subject_dict(rec), 201)
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/subjects/<int:sid>', type='http', auth='public',
methods=['PATCH', 'PUT'], csrf=False)
@jwt_required
def update_subject(self, sid, **kw):
try:
rec = request.env['encoach.subject'].sudo().browse(sid)
if not rec.exists():
return _error_response('Not found', 404)
body = _get_json_body()
vals = {}
for k in ('name', 'code', 'description'):
if k in body:
vals[k] = body[k]
if 'is_active' in body:
vals['active'] = bool(body['is_active'])
if vals:
rec.write(vals)
return _json_response(_subject_dict(rec))
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/subjects/<int:sid>', type='http', auth='public',
methods=['DELETE'], csrf=False)
@jwt_required
def delete_subject(self, sid, **kw):
try:
rec = request.env['encoach.subject'].sudo().browse(sid)
if rec.exists():
rec.unlink()
return _json_response({'success': True})
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/subjects/<int:sid>/taxonomy', type='http',
auth='public', methods=['GET'], csrf=False)
@jwt_required
def get_subject_taxonomy(self, sid, **kw):
try:
s = request.env['encoach.subject'].sudo().browse(sid)
if not s.exists():
return _error_response('Subject not found', 404)
domains = []
for d in s.domain_ids:
topics = []
for t in d.topic_ids:
objectives = [_objective_dict(o) for o in t.learning_objective_ids]
td = _topic_dict(t)
td['objectives'] = objectives
topics.append(td)
dd = _domain_dict(d)
dd['topics'] = topics
domains.append(dd)
result = _subject_dict(s)
result['domains'] = domains
return _json_response({'data': {'subject': _subject_dict(s), 'domains': domains}})
except Exception as e:
_logger.exception('get_subject_taxonomy')
return _error_response(str(e), 500)
# ══════════════════════════════════════════════════════════════════
# DOMAINS /api/domains
# ══════════════════════════════════════════════════════════════════
@http.route('/api/domains', type='http', auth='public',
methods=['GET'], csrf=False)
@jwt_required
def list_domains(self, **kw):
try:
domain_filter = []
if kw.get('subject_id'):
domain_filter.append(('subject_id', '=', int(kw['subject_id'])))
recs = request.env['encoach.domain'].sudo().search(domain_filter)
return _json_response({'data': [_domain_dict(d) for d in recs]})
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/domains', type='http', auth='public',
methods=['POST'], csrf=False)
@jwt_required
def create_domain(self, **kw):
try:
body = _get_json_body()
vals = {
'name': body.get('name', ''),
'subject_id': int(body['subject_id']),
}
if body.get('description'):
vals['description'] = body['description']
rec = request.env['encoach.domain'].sudo().create(vals)
return _json_response(_domain_dict(rec), 201)
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/domains/<int:did>', type='http', auth='public',
methods=['PATCH', 'PUT'], csrf=False)
@jwt_required
def update_domain(self, did, **kw):
try:
rec = request.env['encoach.domain'].sudo().browse(did)
if not rec.exists():
return _error_response('Not found', 404)
body = _get_json_body()
vals = {}
for k in ('name', 'description'):
if k in body:
vals[k] = body[k]
if vals:
rec.write(vals)
return _json_response(_domain_dict(rec))
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/domains/<int:did>', type='http', auth='public',
methods=['DELETE'], csrf=False)
@jwt_required
def delete_domain(self, did, **kw):
try:
rec = request.env['encoach.domain'].sudo().browse(did)
if rec.exists():
rec.unlink()
return _json_response({'success': True})
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/domains/<int:did>/ai-suggest', type='http',
auth='public', methods=['POST'], csrf=False)
@jwt_required
def ai_suggest_topics(self, did, **kw):
try:
domain = request.env['encoach.domain'].sudo().browse(did)
if not domain.exists():
return _error_response('Domain not found', 404)
suggestions = [
{'name': f'{domain.name} - Fundamentals', 'difficulty_level': 'easy', 'estimated_hours': 2},
{'name': f'{domain.name} - Intermediate Concepts', 'difficulty_level': 'medium', 'estimated_hours': 3},
{'name': f'{domain.name} - Advanced Topics', 'difficulty_level': 'hard', 'estimated_hours': 4},
]
return _json_response({'suggestions': suggestions})
except Exception as e:
return _error_response(str(e), 500)
# ══════════════════════════════════════════════════════════════════
# TOPICS /api/topics
# ══════════════════════════════════════════════════════════════════
@http.route('/api/topics', type='http', auth='public',
methods=['GET'], csrf=False)
@jwt_required
def list_topics(self, **kw):
try:
domain_filter = []
if kw.get('domain_id'):
domain_filter.append(('domain_id', '=', int(kw['domain_id'])))
if kw.get('subject_id'):
domains = request.env['encoach.domain'].sudo().search([('subject_id', '=', int(kw['subject_id']))])
domain_filter.append(('domain_id', 'in', domains.ids))
recs = request.env['encoach.topic'].sudo().search(domain_filter)
return _json_response({'data': [_topic_dict(t) for t in recs]})
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/topics', type='http', auth='public',
methods=['POST'], csrf=False)
@jwt_required
def create_topic(self, **kw):
try:
body = _get_json_body()
vals = {
'name': body.get('name', ''),
'domain_id': int(body['domain_id']),
}
if body.get('description'):
vals['description'] = body['description']
rec = request.env['encoach.topic'].sudo().create(vals)
return _json_response(_topic_dict(rec), 201)
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/topics/<int:tid>', type='http', auth='public',
methods=['PATCH', 'PUT'], csrf=False)
@jwt_required
def update_topic(self, tid, **kw):
try:
rec = request.env['encoach.topic'].sudo().browse(tid)
if not rec.exists():
return _error_response('Not found', 404)
body = _get_json_body()
vals = {}
for k in ('name', 'description'):
if k in body:
vals[k] = body[k]
if vals:
rec.write(vals)
return _json_response(_topic_dict(rec))
except Exception as e:
return _error_response(str(e), 500)
@http.route('/api/topics/<int:tid>', type='http', auth='public',
methods=['DELETE'], csrf=False)
@jwt_required
def delete_topic(self, tid, **kw):
try:
rec = request.env['encoach.topic'].sudo().browse(tid)
if rec.exists():
rec.unlink()
return _json_response({'success': True})
except Exception as e:
return _error_response(str(e), 500)
# ══════════════════════════════════════════════════════════════════
# Legacy /api/taxonomy/* routes
# ══════════════════════════════════════════════════════════════════
@http.route('/api/taxonomy/tree', type='http', auth='public',
methods=['GET'], csrf=False)
@jwt_required
def get_tree(self, **kw):
try:
Subject = request.env['encoach.subject'].sudo()
subjects = Subject.search([('active', '=', True)])
tree = []
for s in subjects:
domains = []
for d in s.domain_ids:
topics = []
for t in d.topic_ids:
objectives = []
for o in t.learning_objective_ids:
objectives.append({
'id': o.id,
'name': o.name,
'bloom_level': o.bloom_level or '',
'description': o.description or '',
})
topics.append({
'id': t.id,
'name': t.name,
'description': t.description or '',
'learning_objectives': objectives,
})
domains.append({
'id': d.id,
'name': d.name,
'description': d.description or '',
'topics': topics,
})
tree.append({
'id': s.id,
'name': s.name,
'code': s.code or '',
'description': s.description or '',
'domains': domains,
})
return _json_response({'tree': tree})
except Exception as e:
_logger.exception('get_tree failed')
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# POST /api/taxonomy/node
# ------------------------------------------------------------------
@http.route('/api/taxonomy/node', type='http', auth='none',
methods=['POST'], csrf=False)
@jwt_required
def create_node(self, **kw):
try:
body = _get_json_body()
node_type = body.get('type')
if not node_type or node_type not in TAXONOMY_MODELS:
return _error_response(
f'type must be one of: {", ".join(TAXONOMY_MODELS.keys())}', 400,
)
name = body.get('name')
if not name:
return _error_response('name is required', 400)
model_name = TAXONOMY_MODELS[node_type]
Model = request.env[model_name].sudo()
vals = {'name': name}
if body.get('description'):
vals['description'] = body['description']
if node_type == 'subject':
code = body.get('code')
if not code:
return _error_response('code is required for subjects', 400)
vals['code'] = code
else:
parent_id = body.get('parent_id')
if not parent_id:
parent_field = PARENT_FIELD_MAP[node_type]
return _error_response(
f'parent_id ({parent_field}) is required for {node_type}', 400,
)
vals[PARENT_FIELD_MAP[node_type]] = int(parent_id)
if node_type == 'learning_objective' and body.get('bloom_level'):
vals['bloom_level'] = body['bloom_level']
record = Model.create(vals)
return _json_response({
'id': record.id,
'type': node_type,
'name': record.name,
}, 201)
except Exception as e:
_logger.exception('create_node failed')
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# PUT /api/taxonomy/node/<int:node_id>
# ------------------------------------------------------------------
@http.route('/api/taxonomy/node/<int:node_id>', type='http', auth='none',
methods=['PUT'], csrf=False)
@jwt_required
def update_node(self, node_id, **kw):
try:
body = _get_json_body()
node_type = body.get('type')
if not node_type or node_type not in TAXONOMY_MODELS:
return _error_response(
f'type must be one of: {", ".join(TAXONOMY_MODELS.keys())}', 400,
)
model_name = TAXONOMY_MODELS[node_type]
Model = request.env[model_name].sudo()
record = Model.browse(node_id)
if not record.exists():
return _error_response(f'{node_type} not found', 404)
vals = {}
if body.get('name'):
vals['name'] = body['name']
if 'description' in body:
vals['description'] = body.get('description') or ''
if node_type == 'subject':
if body.get('code'):
vals['code'] = body['code']
if 'active' in body:
vals['active'] = bool(body['active'])
elif node_type == 'learning_objective' and body.get('bloom_level'):
vals['bloom_level'] = body['bloom_level']
if vals:
record.write(vals)
return _json_response(record.to_api_dict())
except Exception as e:
_logger.exception('update_node failed')
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# DELETE /api/taxonomy/node/<int:node_id>
# ------------------------------------------------------------------
@http.route('/api/taxonomy/node/<int:node_id>', type='http', auth='none',
methods=['DELETE'], csrf=False)
@jwt_required
def delete_node(self, node_id, **kw):
try:
body = _get_json_body()
node_type = body.get('type') or kw.get('type')
if not node_type or node_type not in TAXONOMY_MODELS:
return _error_response(
f'type must be one of: {", ".join(TAXONOMY_MODELS.keys())}', 400,
)
model_name = TAXONOMY_MODELS[node_type]
Model = request.env[model_name].sudo()
record = Model.browse(node_id)
if not record.exists():
return _error_response(f'{node_type} not found', 404)
record.unlink()
return _json_response({}, 204)
except Exception as e:
_logger.exception('delete_node failed')
return _error_response(str(e), 500)