devops/encoach_backend_new_v2
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat(backend): course-plan student visibility, multi-voice TTS, OCR sources, branches

Browse Source 
Ported from monorepo v4 commit 3b62075d (backend/* portion).

encoach_ai_course:
- workbook_attempt model + scoring + REST endpoints for student attempts
- dialogue_parser splits scripts by speaker, classifies gender, strips labels
- media_service: multi-voice TTS via Polly/ElevenLabs, ffmpeg concatenation,
  manual media upload endpoint (audio/image/video) with size validation
- source_indexer: OCR fallback (pytesseract + pdf2image) for scanned PDFs,
  page-streaming to stay under memory limit
- exercise_extractor + rag_context for RAG-grounded interactive workbooks
- course_plan_pipeline: v2 generator that grounds week material on indexed
  sources and persists grounded_on_json metadata
- security: access rules for new models

encoach_lms_api:
- branches model + controller (entity-scoped LMS branches)
- classroom_ext + course_ext (assignment + section workflow)
- classrooms controller: students/teachers/assign-course endpoints

Made-with: Cursor
This commit is contained in:
Yamen Ahmad
2026-04-28 00:20:35 +04:00
parent cd47d01f53
commit 565ddd5ff7
21 changed files with 4045 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
custom_addons/encoach_ai_course/models/__init__.py
View File

@@ -4,3 +4,4 @@ from . import course_plan
from . import course_plan_source
from . import course_plan_media
from . import course_plan_assignment
from . import workbook_attempt

16
custom_addons/encoach_ai_course/models/course_plan.py
View File

@@ -44,6 +44,7 @@ MATERIAL_TYPE_SELECTION = [
('grammar_lesson', 'Grammar Lesson'),
('vocabulary_list', 'Vocabulary List'),
('practice', 'Practice Exercises'),
('interactive_workbook', 'Interactive Workbook'),
('other', 'Other'),
]
@@ -302,6 +303,17 @@ class CoursePlanMaterial(models.Model):
help='Plain-text rendering for easy preview / copy-paste when the '
'structured body is not needed.',
)
grounded_on_json = fields.Text(
help='JSON list of source citations used to ground this material via '
'RAG: [{source_id, title, chunks_used}]. Surfaced in the UI as '
'a "Grounded on N references" badge.',
)
extracted_from_json = fields.Text(
help='JSON {source_id, page_hint, …} when the material was mined '
'from an indexed PDF/DOCX by ExerciseExtractor. Distinct from '
'grounded_on_json which records *retrieval* rather than '
'*extraction* provenance.',
)
media_ids = fields.One2many(
'encoach.course.plan.media', 'material_id', string='Generated media',
@@ -317,6 +329,8 @@ class CoursePlanMaterial(models.Model):
def to_api_dict(self, include_media=True):
self.ensure_one()
grounded = self._loads(self.grounded_on_json, [])
extracted = self._loads(self.extracted_from_json, None)
out = {
'id': self.id,
'plan_id': self.plan_id.id,
@@ -330,6 +344,8 @@ class CoursePlanMaterial(models.Model):
'summary': self.summary or '',
'body': self._loads(self.body_json, {}),
'body_text': self.body_text or '',
'grounded_on': grounded if isinstance(grounded, list) else [],
'extracted_from': extracted if isinstance(extracted, dict) else None,
}
if include_media:
out['media'] = [m.to_api_dict() for m in self.media_ids]

337
custom_addons/encoach_ai_course/models/workbook_attempt.py Normal file
View File

@@ -0,0 +1,337 @@
"""Per-student workbook attempt — server-side scoring + persistence.
A student opens an ``interactive_workbook`` material in
``InteractiveWorkbook.tsx``, types/picks/drags answers, and either
clicks "Check answers" (debounced save) or "Submit final" (final save).
Each save creates or updates one ``encoach.course.plan.workbook.attempt``
row keyed by ``(material_id, student_id, attempt_number)``.
The score is always recomputed server-side from the persisted answers
so a tampered client cannot inflate the score. The grading function
handles each of the six exercise types defined in the plan schema:
* ``gap_fill`` — case-insensitive equality, ``alt`` list match.
* ``multiple_choice`` — exact match against ``answer`` (string or letter).
* ``match_pairs`` — set-equality of pair tuples.
* ``reorder_words`` — token-by-token equality (whitespace-tolerant).
* ``transformation`` — equality with normalised punctuation, ``alt``
fallback, capitalisation tolerated.
* ``short_answer`` — glob-style ``accepted`` template (``*`` matches
any phrase) plus exact ``answer`` fallback.
Entity isolation
----------------
Every attempt carries the same ``entity_id`` as the parent plan. A SQL
constraint mirrors the LMS isolation pattern: a student in entity A
cannot create an attempt against a material owned by entity B.
"""
from __future__ import annotations
import json
import logging
import re
from odoo import api, fields, models
from odoo.exceptions import ValidationError
_logger = logging.getLogger(__name__)
# ----------------------------------------------------------------------
# Scoring helpers — pure functions so the unit tests can hit them
# directly without spinning up an ORM.
# ----------------------------------------------------------------------
def _norm_text(s) -> str:
if s is None:
return ''
return re.sub(r'\s+', ' ', str(s)).strip().lower()
def _norm_punct(s) -> str:
"""Normalise final punctuation/quotes for transformation checks."""
return re.sub(r'[\s\.\?\!,]+$', '', _norm_text(s))
def _glob_match(pattern: str, value: str) -> bool:
"""Tiny ``*``-only glob matcher used by short-answer templates."""
pat = _norm_text(pattern)
val = _norm_text(value)
if not pat:
return False
if '*' not in pat:
return pat == val
rx = '^' + re.escape(pat).replace(r'\*', r'.*') + '$'
return bool(re.match(rx, val))
def _pairs_equal(answer, given) -> bool:
"""Set-equality check for match_pairs answers."""
def _to_set(p):
out = set()
for it in (p or []):
if isinstance(it, (list, tuple)) and len(it) == 2:
try:
out.add((int(it[0]), int(it[1])))
except (TypeError, ValueError):
return None
return out
a = _to_set(answer)
g = _to_set(given)
if a is None or g is None:
return False
return a == g
def _check_one(ex: dict, given) -> bool:
"""Return True iff ``given`` is a correct answer for ``ex``."""
t = (ex.get('type') or '').lower()
if t == 'gap_fill':
if given is None:
return False
accepted = [ex.get('answer')] + list(ex.get('alt') or [])
n = _norm_text(given)
return any(_norm_text(a) == n for a in accepted if a is not None)
if t == 'multiple_choice':
opts = ex.get('options') or []
ans = ex.get('answer')
# The "answer" in the schema can be the literal option string
# ("Paris") OR a single letter ("A","B",…) — accept either.
if ans is None:
return False
if isinstance(given, str):
n = _norm_text(given)
if n == _norm_text(ans):
return True
# Letter-mode: convert "A"/"B" to index → option string.
if isinstance(ans, str) and len(ans) == 1 and ans.isalpha():
idx = ord(ans.upper()) - ord('A')
if 0 <= idx < len(opts) and _norm_text(opts[idx]) == n:
return True
if isinstance(given, str) and len(given) == 1 and given.isalpha():
idx = ord(given.upper()) - ord('A')
if 0 <= idx < len(opts) and _norm_text(opts[idx]) == _norm_text(ans):
return True
return False
if t == 'match_pairs':
return _pairs_equal(ex.get('answer'), given)
if t == 'reorder_words':
if given is None:
return False
# Accept either a list of tokens or a joined string.
if isinstance(given, list):
given_str = ' '.join(str(t) for t in given)
else:
given_str = str(given)
return _norm_text(given_str) == _norm_text(ex.get('answer'))
if t == 'transformation':
if given is None:
return False
accepted = [ex.get('answer')] + list(ex.get('alt') or [])
n = _norm_punct(given)
return any(_norm_punct(a) == n for a in accepted if a is not None)
if t == 'short_answer':
if given is None or str(given).strip() == '':
return False
accepted = list(ex.get('accepted') or [])
if ex.get('answer'):
accepted.append(str(ex['answer']))
return any(_glob_match(p, given) for p in accepted)
# Unknown / un-checkable type — never auto-mark correct.
return False
def score_exercises(exercises: list[dict], answers: dict) -> dict:
"""Grade ``answers`` against ``exercises``. Pure / safe / deterministic."""
items: list[dict] = []
correct = 0
for ex in exercises or []:
eid = str(ex.get('id') or '')
if not eid:
continue
given = answers.get(eid) if isinstance(answers, dict) else None
ok = _check_one(ex, given)
if ok:
correct += 1
items.append({
'id': eid,
'correct': bool(ok),
'expected': ex.get('answer'),
'given': given,
})
total = len(items)
return {
'items': items,
'correct': correct,
'total': total,
'percent': round((correct / total) * 100, 1) if total else 0.0,
}
# ----------------------------------------------------------------------
# Odoo model
# ----------------------------------------------------------------------
class CoursePlanWorkbookAttempt(models.Model):
_name = 'encoach.course.plan.workbook.attempt'
_description = 'Course-plan interactive workbook attempt'
_order = 'submitted_at desc, id desc'
_rec_name = 'display_name'
material_id = fields.Many2one(
'encoach.course.plan.material',
required=True, ondelete='cascade', index=True,
)
plan_id = fields.Many2one(
related='material_id.plan_id', store=True, index=True,
)
student_id = fields.Many2one(
'res.users', required=True, ondelete='cascade', index=True,
string='Student user',
)
entity_id = fields.Many2one(
'encoach.entity', index=True, string='Entity',
help='Mirrors the parent plan\'s entity for LMS isolation.',
)
answers_json = fields.Text(
default='{}',
help='Per-exercise dict keyed by exercise ID '
'— shape mirrors the schema output by InteractiveWorkbook.tsx.',
)
score_json = fields.Text(
default='{}',
help='Server-side grading output: '
'{items: [{id, correct, expected, given}], correct, total, percent}.',
)
correct_count = fields.Integer(default=0, string='Correct')
total_count = fields.Integer(default=0, string='Total')
percent = fields.Float(default=0.0, digits=(5, 1))
submitted_at = fields.Datetime(
help='Set when the student clicks "Submit final". '
'Until then the row is editable on every "Check answers".',
)
last_updated_at = fields.Datetime(default=fields.Datetime.now)
attempt_number = fields.Integer(default=1)
is_final = fields.Boolean(default=False)
display_name = fields.Char(compute='_compute_display_name', store=False)
_sql_constraints = [
(
'workbook_attempt_unique',
'unique(material_id, student_id, attempt_number)',
'Only one workbook attempt per (material, student, attempt#).',
),
]
# ------------------------------------------------------------------
@api.depends('material_id', 'student_id', 'attempt_number')
def _compute_display_name(self):
for rec in self:
mat = rec.material_id.title or 'Workbook'
who = rec.student_id.name or rec.student_id.login or '?'
rec.display_name = f'{mat} — {who} (attempt {rec.attempt_number})'
# ------------------------------------------------------------------
@api.constrains('material_id', 'entity_id')
def _check_entity_isolation(self):
for rec in self:
plan = rec.material_id.plan_id
plan_entity = plan.entity_id.id if (plan and plan.entity_id) else False
if rec.entity_id and plan_entity and rec.entity_id.id != plan_entity:
raise ValidationError(
'Workbook attempt entity must match its plan\'s entity.'
)
# ------------------------------------------------------------------
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
mat = self.env['encoach.course.plan.material'].sudo().browse(
int(vals.get('material_id') or 0),
)
if mat and mat.exists() and not vals.get('entity_id'):
vals['entity_id'] = (
mat.plan_id.entity_id.id if mat.plan_id and mat.plan_id.entity_id
else False
)
return super().create(vals_list)
# ------------------------------------------------------------------
@staticmethod
def _exercises_from_material(material) -> list[dict]:
try:
body = json.loads(material.body_json or '{}')
except (TypeError, ValueError):
return []
ex = body.get('exercises')
if isinstance(ex, list):
return ex
# Some skill bodies (grammar / vocabulary) embed the workbook
# one level deeper. Search a couple of known keys before giving up.
nested = (
(body.get('interactive_workbook') or {}).get('exercises')
if isinstance(body.get('interactive_workbook'), dict) else None
)
if isinstance(nested, list):
return nested
return []
# ------------------------------------------------------------------
def grade(self, answers: dict, *, finalize: bool = False) -> dict:
"""Re-grade ``answers`` server-side and persist on this row."""
self.ensure_one()
exercises = self._exercises_from_material(self.material_id)
score = score_exercises(exercises, answers or {})
vals = {
'answers_json': json.dumps(answers or {}, ensure_ascii=False),
'score_json': json.dumps(score, ensure_ascii=False),
'correct_count': score['correct'],
'total_count': score['total'],
'percent': score['percent'],
'last_updated_at': fields.Datetime.now(),
}
if finalize:
vals['submitted_at'] = fields.Datetime.now()
vals['is_final'] = True
self.write(vals)
return score
# ------------------------------------------------------------------
def to_api_dict(self) -> dict:
self.ensure_one()
try:
answers = json.loads(self.answers_json or '{}')
except (TypeError, ValueError):
answers = {}
try:
score = json.loads(self.score_json or '{}')
except (TypeError, ValueError):
score = {}
return {
'id': self.id,
'material_id': self.material_id.id,
'plan_id': self.plan_id.id if self.plan_id else None,
'student_id': self.student_id.id,
'student_name': self.student_id.name or self.student_id.login or '',
'attempt_number': self.attempt_number,
'is_final': bool(self.is_final),
'submitted_at': (
self.submitted_at.isoformat() if self.submitted_at else None
),
'last_updated_at': (
self.last_updated_at.isoformat() if self.last_updated_at else None
),
'correct_count': self.correct_count,
'total_count': self.total_count,
'percent': self.percent,
'answers': answers,
'score': score,
}