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

feat(backend): Phase 2/3 hardening release

Browse Source 
Roadmap P0 — platform safety & ops
- Merge duplicate encoach.student.attempt/answer models into encoach_scoring
  and drop the stale encoach_exam_template copies.
- Remove duplicate /api/exam/* routes; canonicalize on one controller tree.
- Gate raw-SQL seeds in seed_demo_data.py behind an explicit env flag.
- Add /api/health and /api/health/ready (DB + LLM reachability) endpoints.
- Fix docker-compose + ship odoo-docker.conf for container-local runs.
- Enforce OpenAI request_timeout=30s and @jwt_required on all AI/coach routes.
- Promote canonical cefr_mapper to encoach_ai.services.cefr_mapper.
- JWT cache TTL=30s + invalidation hook on user mutation.

Roadmap P1 — exam correctness & data provenance
- Wire QualityChecker + IeltsValidator into exam submit with a
  pending_review gate (encoach_ai.services.question_validator).
- Populate RAG metadata (course_id, subject_id, entity_id, taxonomy) on
  encoach_vector embeddings and add a chunking pipeline (>2000 chars).
- Add provenance fields on encoach.question (model, prompt_hash, log_id)
  and validate LLM output with schema before DB insert.
- Unify response envelope to {items,total,page,size}.
- Approval reject rollback with savepoint atomicity.
- Ticket notifications on status/assignee change.

Roadmap P2 — performance & observability
- Reports: replace Python loops with SQL read_group aggregations.
- X-Request-ID middleware + structured JSON logs.
- In-process/Prometheus counters and openapi.py controller exporting a
  spec by scanning @http.route decorators.
- Paymob real checkout + HMAC-SHA512 webhook verification, backed by a
  new encoach.paymob.order model and ir.config_parameter credentials.
- JWT refresh tokens + revocation table.
- Composite DB indexes on hot report/ticket/attempt paths.

Roadmap P3 — human-in-the-loop & compliance
- Human-in-the-loop exam review workflow (pending_review → publish) with
  new review controller and status transitions.
- encoach.ai.prompt model + versioning + admin editor endpoints (one
  active version per key, render-preview dry run).
- Student feedback loop → encoach.ai.feedback (upsert per user/subject,
  admin triage + resolve endpoints).
- GDPR export (/api/gdpr/export) and right-to-erasure (/api/gdpr/delete)
  with anonymization, tombstone record, and admin-self-erasure guard.
- HttpCase smoke tests for /api/health and /api/health/ready.

Made-with: Cursor
This commit is contained in:
Yamen Ahmad
2026-04-19 14:16:09 +04:00
parent 1a0349c381
commit 3972023a30
64 changed files with 4121 additions and 701 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
custom_addons/encoach_ai/__manifest__.py
View File

@@ -15,7 +15,7 @@
- AI Search, Insights, Report Narrative
""",
"author": "EnCoach",
"depends": ["base", "encoach_core"],
"depends": ["base", "encoach_core", "encoach_api"],
"external_dependencies": {
"python": ["openai", "boto3"],
},

2
custom_addons/encoach_ai/controllers/__init__.py
View File

@@ -1,3 +1,5 @@
from . import ai_controller
from . import coach_controller
from . import media_controller
from . import prompt_controller
from . import feedback_controller

161
custom_addons/encoach_ai/controllers/ai_controller.py
View File

@@ -2,8 +2,9 @@
import json
import logging
from odoo import http
from odoo import fields, http
from odoo.http import request, Response
from odoo.addons.encoach_api.controllers.base import jwt_required
_logger = logging.getLogger(__name__)
@@ -27,7 +28,8 @@ class AIController(http.Controller):
"""Handles /api/ai/* endpoints consumed by frontend AI components."""
# ── POST /api/ai/search — AiSearchBar.tsx (RAG-enhanced) ──
@http.route("/api/ai/search", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/search", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_search(self, **kw):
body = _get_json()
query = body.get("query", "")
@@ -43,7 +45,8 @@ class AIController(http.Controller):
return _json_response({"answer": f"AI search unavailable: {e}", "suggestions": []})
# ── GET /api/ai/vector-search — pure semantic search without GPT ──
@http.route("/api/ai/vector-search", type="http", auth="public", methods=["GET"], csrf=False)
@http.route("/api/ai/vector-search", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def ai_vector_search(self, **kw):
query = request.params.get("q", "")
content_type = request.params.get("content_type")
@@ -60,7 +63,8 @@ class AIController(http.Controller):
return _json_response({"results": [], "query": query, "error": str(e)})
# ── POST /api/ai/insights — AiInsightsPanel.tsx ──
@http.route("/api/ai/insights", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/insights", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_insights(self, **kw):
body = _get_json()
try:
@@ -76,7 +80,8 @@ class AIController(http.Controller):
return _json_response({"insights": [{"title": "AI Unavailable", "description": str(e), "severity": "info", "recommendation": "Check AI settings."}]})
# ── GET /api/ai/alerts — AiAlertBanner.tsx ──
@http.route("/api/ai/alerts", type="http", auth="public", methods=["GET"], csrf=False)
@http.route("/api/ai/alerts", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def ai_alerts(self, **kw):
try:
from odoo.addons.encoach_ai.services.openai_service import OpenAIService
@@ -92,7 +97,8 @@ class AIController(http.Controller):
return _json_response({"alerts": []})
# ── POST /api/ai/report-narrative — AiReportNarrative.tsx ──
@http.route("/api/ai/report-narrative", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/report-narrative", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_report_narrative(self, **kw):
body = _get_json()
try:
@@ -107,7 +113,8 @@ class AIController(http.Controller):
return _json_response({"narrative": f"Report generation unavailable: {e}"})
# ── POST /api/ai/batch-optimize — AiBatchOptimizer.tsx ──
@http.route("/api/ai/batch-optimize", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/batch-optimize", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_batch_optimize(self, **kw):
body = _get_json()
try:
@@ -122,7 +129,8 @@ class AIController(http.Controller):
return _json_response({"optimized": [], "summary": str(e), "impact": "none"})
# ── POST /api/ai/grade-suggest — AiGradingAssistant.tsx ──
@http.route("/api/ai/grade-suggest", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/grade-suggest", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_grade_suggest(self, **kw):
body = _get_json()
try:
@@ -146,7 +154,8 @@ class AIController(http.Controller):
return _json_response({"scores": {}, "overall_band": 0, "feedback": str(e), "suggestions": []})
# ── POST /api/ai/generate-resource — ModuleBuilder.tsx (dedup-aware) ──
@http.route("/api/ai/generate-resource", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/generate-resource", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_generate_resource(self, **kw):
body = _get_json()
try:
@@ -162,7 +171,8 @@ class AIController(http.Controller):
return _json_response({"resource": None, "status": "error", "error": str(e)})
# ── POST /api/ai/detect — GPTZero AI detection ──
@http.route("/api/ai/detect", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/detect", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_detect(self, **kw):
body = _get_json()
try:
@@ -174,7 +184,8 @@ class AIController(http.Controller):
return _json_response({"is_ai_generated": False, "ai_probability": 0, "error": str(e)})
# ── POST /api/plagiarism/check — plagiarism.service.ts ──
@http.route("/api/plagiarism/check", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/plagiarism/check", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def plagiarism_check(self, **kw):
body = _get_json()
try:
@@ -187,7 +198,8 @@ class AIController(http.Controller):
return _json_response({"report_id": None, "error": str(e)})
# ── POST /api/domains/:domainId/ai-suggest — TaxonomyManager.tsx ──
@http.route("/api/domains/<int:domain_id>/ai-suggest", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/domains/<int:domain_id>/ai-suggest", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_suggest_topics(self, domain_id, **kw):
body = _get_json()
try:
@@ -206,7 +218,8 @@ class AIController(http.Controller):
return _json_response({"topics": [], "error": str(e)})
# ── POST /api/learning-plan/generate — LearningPlan.tsx ──
@http.route("/api/learning-plan/generate", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/learning-plan/generate", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def learning_plan_generate(self, **kw):
body = _get_json()
try:
@@ -227,7 +240,8 @@ class AIController(http.Controller):
return _json_response({"plan": None, "error": str(e)})
# ── Workbench endpoints — AiWorkbench.tsx ──
@http.route("/api/workbench/generate-outline", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/workbench/generate-outline", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def workbench_outline(self, **kw):
body = _get_json()
try:
@@ -244,7 +258,8 @@ class AIController(http.Controller):
except Exception as e:
return _json_response({"chapters": [], "error": str(e)})
@http.route("/api/workbench/generate-chapter", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/workbench/generate-chapter", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def workbench_chapter(self, **kw):
body = _get_json()
try:
@@ -262,7 +277,8 @@ class AIController(http.Controller):
except Exception as e:
return _json_response({"content": "", "error": str(e)})
@http.route("/api/workbench/generate-rubric", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/workbench/generate-rubric", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def workbench_rubric(self, **kw):
body = _get_json()
try:
@@ -280,11 +296,13 @@ class AIController(http.Controller):
except Exception as e:
return _json_response({"rubric": None, "error": str(e)})
@http.route("/api/workbench/regenerate", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/workbench/regenerate", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def workbench_regenerate(self, **kw):
return self.workbench_chapter(**kw)
@http.route("/api/workbench/publish", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/workbench/publish", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def workbench_publish(self, **kw):
body = _get_json()
try:
@@ -315,7 +333,8 @@ class AIController(http.Controller):
return _json_response({"status": "error", "error": str(e)}, 500)
# ── POST /api/ai/suggest-rubric-criteria — RubricsPage.tsx ──
@http.route("/api/ai/suggest-rubric-criteria", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/suggest-rubric-criteria", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_suggest_rubric_criteria(self, **kw):
from odoo.addons.encoach_api.controllers.base import validate_token
user = validate_token()
@@ -450,7 +469,8 @@ class AIController(http.Controller):
]
# ── Exam generation — GenerationPage.tsx ──
@http.route("/api/exam/<string:module>/generate", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/exam/<string:module>/generate", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def exam_generate(self, module, **kw):
from odoo.addons.encoach_api.controllers.base import validate_token
user = validate_token()
@@ -1252,7 +1272,8 @@ class AIController(http.Controller):
return {"questions": questions}
# ── POST /api/exam/generation/submit — create exam from generation page ──
@http.route("/api/exam/generation/submit", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/exam/generation/submit", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def generation_submit(self, **kw):
from odoo.addons.encoach_api.controllers.base import validate_token
user = validate_token()
@@ -1265,12 +1286,40 @@ class AIController(http.Controller):
if not title:
return _json_response({"error": "title is required"}, 400)
from odoo.addons.encoach_ai.services.question_validator import (
validate_payload, score_question, summarize_question_reports,
dumps_report, prompt_hash as _prompt_hash,
VERDICT_FAIL, VERDICT_WARN,
)
schema_report = validate_payload(body)
if schema_report["verdict"] == VERDICT_FAIL:
_logger.warning(
"generation_submit rejected %d schema errors: %s",
len(schema_report["errors"]), schema_report["errors"][:3],
)
return _json_response({
"error": "AI payload failed schema validation",
"details": schema_report,
}, 422)
label = body.get("label", "")
modules = body.get("modules", {})
skip_approval = body.get("skip_approval", False)
exam_mode = body.get("exam_mode", "official")
structure_id = body.get("structure_id")
ai_model_used = (
body.get("ai_model")
or request.env["ir.config_parameter"].sudo().get_param(
"encoach_ai.openai_fast_model", "gpt-4o-mini"
)
)
payload_hash = _prompt_hash(json.dumps(
{"title": title, "modules": modules}, sort_keys=True, default=str,
))
question_reports = [] # collected for aggregate gating below
first_mod = next(iter(modules.values()), {}) if modules else {}
entity_val = first_mod.get("entity", "none")
entity_id = int(entity_val) if entity_val and entity_val != "none" else False
@@ -1304,13 +1353,14 @@ class AIController(http.Controller):
except KeyError:
return _json_response({"error": "encoach.exam.custom model not available"}, 500)
initial_status = "published" if skip_approval else "draft"
exam_vals = {
"title": title,
"label": label,
"exam_mode": exam_mode,
"teacher_id": request.env.user.id,
"template_id": template_id,
"status": "published" if skip_approval else "draft",
"status": initial_status,
"total_time_min": sum(m.get("timer", 0) for m in modules.values()),
"total_marks": sum(float(m.get("totalMarks", 0)) for m in modules.values()),
"randomize_questions": any(m.get("shuffling", False) for m in modules.values()),
@@ -1344,6 +1394,27 @@ class AIController(http.Controller):
"form_completion": "form_completion", "map_labelling": "map_labelling",
}
now = fields.Datetime.now() if hasattr(fields, 'Datetime') else None
def _create_question(vals, *, cefr_level=None):
stem = vals.get("stem") or ""
q_report = score_question(
stem, skill=vals.get("skill"), cefr_level=cefr_level
)
question_reports.append(q_report)
status = vals.pop("status", "draft")
if q_report["verdict"] == VERDICT_FAIL:
status = "flagged"
vals.update({
"status": status,
"ai_model_used": ai_model_used,
"ai_prompt_hash": payload_hash,
"ai_generated_at": now,
"quality_score": q_report.get("score"),
"quality_report": dumps_report(q_report),
})
return Question.create(vals)
seq = 10
total_questions = 0
for mod_key, mod_data in modules.items():
@@ -1381,7 +1452,7 @@ class AIController(http.Controller):
for ex in (passage.get("exercises") or []):
q_type = QUESTION_TYPE_MAP.get(ex.get("type", "mcq"), "mcq")
opts = ex.get("options", [])
q = Question.create({
q = _create_question({
"skill": mod_key if mod_key in ("reading", "listening", "writing", "speaking", "grammar", "vocabulary", "math", "it") else "reading",
"source_type": "passage",
"question_type": q_type,
@@ -1392,7 +1463,7 @@ class AIController(http.Controller):
"difficulty": _q_difficulty_for(ex),
"status": "active",
"ai_generated": True,
})
}, cefr_level=(ex.get("cefr_level") or cefr_level))
question_ids.append(q.id)
sections_data = mod_data.get("sections") or []
@@ -1402,7 +1473,7 @@ class AIController(http.Controller):
for ex in (s_data.get("exercises") or []):
q_type = QUESTION_TYPE_MAP.get(ex.get("type", "mcq"), "mcq")
opts = ex.get("options", [])
q = Question.create({
q = _create_question({
"skill": "listening",
"source_type": "audio",
"question_type": q_type,
@@ -1413,12 +1484,12 @@ class AIController(http.Controller):
"difficulty": _q_difficulty_for(ex),
"status": "active",
"ai_generated": True,
})
}, cefr_level=(ex.get("cefr_level") or cefr_level))
question_ids.append(q.id)
tasks = mod_data.get("tasks") or []
for t_idx, task in enumerate(tasks):
q = Question.create({
q = _create_question({
"skill": "writing",
"source_type": "writing_prompt",
"question_type": "short_answer",
@@ -1429,12 +1500,12 @@ class AIController(http.Controller):
"difficulty": q_difficulty,
"status": "active",
"ai_generated": True,
})
}, cefr_level=cefr_level)
question_ids.append(q.id)
parts = mod_data.get("parts") or []
for p_idx, part in enumerate(parts):
q = Question.create({
q = _create_question({
"skill": "speaking",
"source_type": "speaking_card",
"question_type": "short_answer",
@@ -1445,7 +1516,7 @@ class AIController(http.Controller):
"difficulty": q_difficulty,
"status": "active",
"ai_generated": True,
})
}, cefr_level=cefr_level)
question_ids.append(q.id)
if question_ids:
@@ -1455,18 +1526,37 @@ class AIController(http.Controller):
})
total_questions += len(question_ids)
quality_summary = summarize_question_reports(question_reports)
if (
exam.status != "draft"
and quality_summary["verdict"] in (VERDICT_FAIL, VERDICT_WARN)
and quality_summary["total"] > 0
):
exam.sudo().write({"status": "pending_review"})
_logger.info(
"exam %s forced pending_review: avg_score=%.2f failed=%d warned=%d",
exam.id, quality_summary["avg_score"],
quality_summary["failed"], quality_summary["warned"],
)
return _json_response({
"exam_id": exam.id,
"status": exam.status,
"template_id": template_id,
"total_questions": total_questions,
"quality": quality_summary,
"schema_validation": {
"verdict": schema_report["verdict"],
"warnings": schema_report["warnings"],
},
}, 201)
except Exception as e:
_logger.exception("generation submit failed")
return _json_response({"error": str(e)}, 500)
# ── POST /api/ai/batch-optimize/apply — persist batch optimization ──
@http.route("/api/ai/batch-optimize/apply", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/ai/batch-optimize/apply", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_batch_optimize_apply(self, **kw):
body = _get_json()
optimized = body.get("optimized", [])
@@ -1481,7 +1571,8 @@ class AIController(http.Controller):
return _json_response({"applied": 0, "error": str(e)}, 500)
# ── POST /api/exam/<module>/generate/save — save generated exam items ──
@http.route("/api/exam/<string:module>/generate/save", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/exam/<string:module>/generate/save", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def exam_generate_save(self, module, **kw):
from odoo.addons.encoach_api.controllers.base import validate_token
user = validate_token()
@@ -1521,7 +1612,8 @@ class AIController(http.Controller):
return _json_response({"saved": 0, "error": str(e)}, 500)
# ── POST /api/workbench/suggest-materials — AI material suggestions ──
@http.route("/api/workbench/suggest-materials", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/workbench/suggest-materials", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def workbench_suggest_materials(self, **kw):
body = _get_json()
try:
@@ -1541,7 +1633,8 @@ class AIController(http.Controller):
return _json_response({"materials": [], "error": str(e)})
# ── Topic content generation — adaptive ──
@http.route("/api/topics/<int:topic_id>/generate-content", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/topics/<int:topic_id>/generate-content", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def topic_generate_content(self, topic_id, **kw):
body = _get_json()
try:

22
custom_addons/encoach_ai/controllers/coach_controller.py
View File

@@ -4,6 +4,10 @@ import json
import logging
from odoo import http
from odoo.http import request, Response
from odoo.addons.encoach_api.controllers.base import (
jwt_required, _json_response as _base_json_response, _get_json_body,
)
_logger = logging.getLogger(__name__)
@@ -27,7 +31,8 @@ class CoachController(http.Controller):
return CoachService(request.env)
# ── POST /api/coach/chat — AiAssistantDrawer.tsx ──
@http.route("/api/coach/chat", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/coach/chat", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def coach_chat(self, **kw):
body = _get_json()
try:
@@ -43,7 +48,8 @@ class CoachController(http.Controller):
return _json_response({"reply": f"I'm having trouble right now. Error: {e}"})
# ── GET /api/coach/tip — AiTipBanner.tsx ──
@http.route("/api/coach/tip", type="http", auth="public", methods=["GET"], csrf=False)
@http.route("/api/coach/tip", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def coach_tip(self, **kw):
context = request.params.get("context", "general")
try:
@@ -53,7 +59,8 @@ class CoachController(http.Controller):
return _json_response({"tip": "Keep practising every day — consistency beats intensity!", "category": "general"})
# ── POST /api/coach/explain — AiGradeExplainer.tsx ──
@http.route("/api/coach/explain", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/coach/explain", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def coach_explain(self, **kw):
body = _get_json()
try:
@@ -67,7 +74,8 @@ class CoachController(http.Controller):
return _json_response({"explanation": f"Could not generate explanation: {e}"})
# ── POST /api/coach/suggest — AiStudyCoach.tsx ──
@http.route("/api/coach/suggest", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/coach/suggest", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def coach_suggest(self, **kw):
body = _get_json()
try:
@@ -82,7 +90,8 @@ class CoachController(http.Controller):
})
# ── POST /api/coach/writing-help — AiWritingHelper.tsx ──
@http.route("/api/coach/writing-help", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/coach/writing-help", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def coach_writing_help(self, **kw):
body = _get_json()
try:
@@ -97,7 +106,8 @@ class CoachController(http.Controller):
return _json_response({"improved_text": "", "changes": [], "tips": [str(e)]})
# ── POST /api/coach/hint — (unused component, wired for completeness) ──
@http.route("/api/coach/hint", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/coach/hint", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def coach_hint(self, **kw):
body = _get_json()
try:

217
custom_addons/encoach_ai/controllers/feedback_controller.py Normal file
View File

@@ -0,0 +1,217 @@
"""Endpoints for submitting & aggregating AI feedback."""
import logging
from odoo import fields, http
from odoo.http import request
from odoo.addons.encoach_api.controllers.base import (
_error_response,
_get_json_body,
_json_response,
_paginate,
jwt_required,
)
_logger = logging.getLogger(__name__)
def _feedback_to_dict(row):
return {
"id": row.id,
"subject_type": row.subject_type,
"subject_id": row.subject_id,
"subject_key": row.subject_key,
"rating": row.rating,
"comment": row.comment or "",
"tags": (row.tags or "").split(",") if row.tags else [],
"prompt_key": row.prompt_key or None,
"prompt_version": row.prompt_version or None,
"ai_log_id": row.ai_log_id.id if row.ai_log_id else None,
"user_id": row.user_id.id,
"user_name": row.user_id.name,
"entity_id": row.entity_id.id if row.entity_id else None,
"course_id": row.course_id.id if row.course_id else None,
"status": row.status,
"create_date": (
fields.Datetime.to_string(row.create_date) if row.create_date else None
),
"resolved_at": (
fields.Datetime.to_string(row.resolved_at) if row.resolved_at else None
),
"resolution_notes": row.resolution_notes or "",
}
class EncoachAIFeedbackController(http.Controller):
"""Feedback write + admin triage endpoints."""
# ------------------------------------------------------------------
# POST /api/ai/feedback — student submits/updates feedback
#
# Upsert semantics: we enforce uniqueness per (user, subject_type,
# subject_id) at the DB layer, so the client doesn't need to know whether
# the row already exists.
# ------------------------------------------------------------------
@http.route("/api/ai/feedback", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def submit(self, **kw):
try:
body = _get_json_body() or {}
subject_type = body.get("subject_type")
subject_id = body.get("subject_id")
rating = body.get("rating")
if subject_type not in {
"question", "coach", "explanation", "translation", "narrative", "other",
}:
return _error_response("Invalid subject_type", 400)
if not isinstance(subject_id, int) or subject_id <= 0:
return _error_response("subject_id must be a positive integer", 400)
if rating not in {"up", "down"}:
return _error_response("rating must be 'up' or 'down'", 400)
comment = body.get("comment") or ""
if rating == "down" and not (comment or "").strip():
# We keep this a soft constraint so the UI can decide how
# strict to be; returning an error here gives clients a clear
# path to prompt the user when needed.
return _error_response(
"A short comment is required when rating is 'down'.", 400,
)
vals = {
"subject_type": subject_type,
"subject_id": subject_id,
"rating": rating,
"comment": comment,
"tags": (
",".join(body.get("tags") or [])
if isinstance(body.get("tags"), list)
else (body.get("tags") or "")
),
"prompt_key": body.get("prompt_key") or False,
"prompt_version": body.get("prompt_version") or 0,
"ai_log_id": body.get("ai_log_id") or False,
"entity_id": body.get("entity_id") or False,
"course_id": body.get("course_id") or False,
"user_id": request.env.user.id,
}
Feedback = request.env["encoach.ai.feedback"].sudo()
existing = Feedback.search([
("user_id", "=", request.env.user.id),
("subject_type", "=", subject_type),
("subject_id", "=", subject_id),
], limit=1)
with request.env.cr.savepoint():
if existing:
existing.write(vals)
row = existing
else:
row = Feedback.create(vals)
return _json_response(_feedback_to_dict(row), 200 if existing else 201)
except Exception as e:
_logger.exception("ai feedback submit failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# GET /api/ai/feedback/summary?subject_type=question&subject_id=42
# ------------------------------------------------------------------
@http.route(
"/api/ai/feedback/summary", type="http", auth="none", methods=["GET"], csrf=False,
)
@jwt_required
def summary(self, subject_type=None, subject_id=None, **kw):
try:
if not subject_type or not subject_id:
return _error_response("subject_type and subject_id are required", 400)
Feedback = request.env["encoach.ai.feedback"].sudo()
domain = [
("subject_type", "=", subject_type),
("subject_id", "=", int(subject_id)),
]
rows = Feedback.search(domain)
up = sum(1 for r in rows if r.rating == "up")
down = sum(1 for r in rows if r.rating == "down")
my_row = Feedback.search([
*domain,
("user_id", "=", request.env.user.id),
], limit=1)
return _json_response({
"subject_type": subject_type,
"subject_id": int(subject_id),
"up": up,
"down": down,
"total": up + down,
"my_rating": my_row.rating if my_row else None,
"my_comment": my_row.comment or "" if my_row else "",
})
except Exception as e:
_logger.exception("ai feedback summary failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# GET /api/ai/feedback — admin triage list
# ------------------------------------------------------------------
@http.route("/api/ai/feedback", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def list_feedback(self, **kw):
try:
if not request.env.user.has_group("base.group_system"):
return _error_response("Admin privileges required", 403)
Feedback = request.env["encoach.ai.feedback"].sudo()
domain = []
if kw.get("status"):
domain.append(("status", "=", kw["status"]))
if kw.get("rating"):
domain.append(("rating", "=", kw["rating"]))
if kw.get("subject_type"):
domain.append(("subject_type", "=", kw["subject_type"]))
if kw.get("prompt_key"):
domain.append(("prompt_key", "=", kw["prompt_key"]))
offset, limit, page = _paginate(kw)
total = Feedback.search_count(domain)
rows = Feedback.search(domain, offset=offset, limit=limit, order="create_date desc")
items = [_feedback_to_dict(r) for r in rows]
return _json_response({
"items": items,
"data": items,
"total": total,
"page": page,
"size": limit,
})
except Exception as e:
_logger.exception("ai feedback list failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# POST /api/ai/feedback/<id>/resolve — admin triage
# ------------------------------------------------------------------
@http.route(
'/api/ai/feedback/<int:feedback_id>/resolve',
type="http", auth="none", methods=["POST"], csrf=False,
)
@jwt_required
def resolve(self, feedback_id, **kw):
try:
if not request.env.user.has_group("base.group_system"):
return _error_response("Admin privileges required", 403)
body = _get_json_body() or {}
status = body.get("status")
if status not in {"acknowledged", "fixed", "dismissed"}:
return _error_response(
"status must be one of acknowledged|fixed|dismissed", 400,
)
row = request.env["encoach.ai.feedback"].sudo().browse(feedback_id)
if not row.exists():
return _error_response("Feedback not found", 404)
with request.env.cr.savepoint():
row.write({
"status": status,
"resolved_by_id": request.env.user.id,
"resolved_at": fields.Datetime.now(),
"resolution_notes": body.get("notes") or "",
})
return _json_response(_feedback_to_dict(row))
except Exception as e:
_logger.exception("ai feedback resolve failed")
return _error_response(str(e), 500)

22
custom_addons/encoach_ai/controllers/media_controller.py
View File

@@ -5,6 +5,10 @@ import json
import logging
from odoo import http
from odoo.http import request, Response
from odoo.addons.encoach_api.controllers.base import (
jwt_required, _json_response as _base_json_response, _get_json_body,
)
_logger = logging.getLogger(__name__)
@@ -52,7 +56,8 @@ class MediaController(http.Controller):
)
# ── POST /api/exam/listening/media — generate listening audio ──
@http.route("/api/exam/listening/media", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/exam/listening/media", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def listening_media(self, **kw):
body = _get_json()
text = body.get("text", "")
@@ -72,7 +77,8 @@ class MediaController(http.Controller):
return _json_response({"error": str(e)}, 500)
# ── POST /api/exam/speaking/media — generate speaking prompt audio ──
@http.route("/api/exam/speaking/media", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/exam/speaking/media", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def speaking_media(self, **kw):
body = _get_json()
text = body.get("text", "")
@@ -89,7 +95,8 @@ class MediaController(http.Controller):
return _json_response({"error": str(e)}, 500)
# ── GET /api/exam/avatars — list ELAI avatars ──
@http.route("/api/exam/avatars", type="http", auth="public", methods=["GET"], csrf=False)
@http.route("/api/exam/avatars", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def list_avatars(self, **kw):
try:
from odoo.addons.encoach_ai.services.elai_service import ElaiService
@@ -100,7 +107,8 @@ class MediaController(http.Controller):
return _json_response({"avatars": [], "note": str(e)})
# ── POST /api/exam/avatar/video — create avatar video ──
@http.route("/api/exam/avatar/video", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/exam/avatar/video", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def create_avatar_video(self, **kw):
body = _get_json()
try:
@@ -116,7 +124,8 @@ class MediaController(http.Controller):
return _json_response({"error": str(e)}, 500)
# ── GET /api/exam/avatar/video/:id — check video status ──
@http.route("/api/exam/avatar/video/<string:video_id>", type="http", auth="public", methods=["GET"], csrf=False)
@http.route("/api/exam/avatar/video/<string:video_id>", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def video_status(self, video_id, **kw):
try:
from odoo.addons.encoach_ai.services.elai_service import ElaiService
@@ -126,7 +135,8 @@ class MediaController(http.Controller):
return _json_response({"video_id": video_id, "status": "error", "error": str(e)})
# ── POST /api/courses/ai-generate — AiCreationAssistant.tsx ──
@http.route("/api/courses/ai-generate", type="http", auth="public", methods=["POST"], csrf=False)
@http.route("/api/courses/ai-generate", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def ai_generate_course(self, **kw):
body = _get_json()
try:

219
custom_addons/encoach_ai/controllers/prompt_controller.py Normal file
View File

@@ -0,0 +1,219 @@
"""Admin endpoints for the versioned AI prompt editor."""
import logging
from odoo import fields, http
from odoo.http import request
from odoo.addons.encoach_api.controllers.base import (
_error_response,
_get_json_body,
_json_response,
_paginate,
jwt_required,
)
_logger = logging.getLogger(__name__)
def _prompt_to_dict(prompt, *, include_content=True):
data = {
"id": prompt.id,
"key": prompt.key,
"version": prompt.version,
"title": prompt.title or "",
"description": prompt.description or "",
"is_active": bool(prompt.is_active),
"variables": (prompt.variables or "").split(",") if prompt.variables else [],
"author_id": prompt.author_id.id if prompt.author_id else None,
"author_name": prompt.author_id.name if prompt.author_id else "",
"activated_at": (
fields.Datetime.to_string(prompt.activated_at) if prompt.activated_at else None
),
"create_date": (
fields.Datetime.to_string(prompt.create_date) if prompt.create_date else None
),
}
if include_content:
data["content"] = prompt.content or ""
return data
class EncoachAIPromptController(http.Controller):
"""Versioned prompt CRUD.
Auth: every endpoint requires a valid JWT. The write endpoints additionally
check for admin rights via ``res.users.has_group('base.group_system')`` so
non-admins can read prompts but not mutate them.
"""
# ------------------------------------------------------------------
# GET /api/ai/prompts — list keys (latest version per key)
# ------------------------------------------------------------------
@http.route("/api/ai/prompts", type="http", auth="none", methods=["GET"], csrf=False)
@jwt_required
def list_keys(self, **kw):
try:
Prompt = request.env["encoach.ai.prompt"].sudo()
# One row per key = the latest version. We fetch candidates then
# deduplicate in Python because Odoo's ORM doesn't expose DISTINCT
# ON ergonomically.
search = (kw.get("search") or "").strip()
domain = [("key", "ilike", search)] if search else []
# Sort by key then descending version so the first occurrence per
# key is the latest.
all_prompts = Prompt.search(domain, order="key asc, version desc")
seen = set()
latest = []
for p in all_prompts:
if p.key in seen:
continue
seen.add(p.key)
latest.append(p)
offset, per_page, page = _paginate(kw)
total = len(latest)
window = latest[offset : offset + per_page]
items = [_prompt_to_dict(p, include_content=False) for p in window]
# Attach version-count for the UI to show "v5 (of 7)" hints.
counts = {}
for p in all_prompts:
counts[p.key] = counts.get(p.key, 0) + 1
for entry in items:
entry["total_versions"] = counts.get(entry["key"], 1)
return _json_response({
"items": items,
"data": items,
"total": total,
"page": page,
"size": per_page,
})
except Exception as e:
_logger.exception("list ai prompts failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# GET /api/ai/prompts/<key>/versions
# ------------------------------------------------------------------
@http.route(
'/api/ai/prompts/<string:key>/versions',
type="http", auth="none", methods=["GET"], csrf=False,
)
@jwt_required
def list_versions(self, key, **kw):
try:
Prompt = request.env["encoach.ai.prompt"].sudo()
prompts = Prompt.search([("key", "=", key)], order="version desc")
if not prompts:
return _error_response("Prompt key not found", 404)
items = [_prompt_to_dict(p, include_content=False) for p in prompts]
return _json_response({
"key": key,
"items": items,
"data": items,
"total": len(items),
"page": 1,
"size": len(items),
})
except Exception as e:
_logger.exception("list prompt versions failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# GET /api/ai/prompts/<int:prompt_id>
# ------------------------------------------------------------------
@http.route(
'/api/ai/prompts/<int:prompt_id>',
type="http", auth="none", methods=["GET"], csrf=False,
)
@jwt_required
def get_version(self, prompt_id, **kw):
try:
prompt = request.env["encoach.ai.prompt"].sudo().browse(prompt_id)
if not prompt.exists():
return _error_response("Prompt not found", 404)
return _json_response(_prompt_to_dict(prompt))
except Exception as e:
_logger.exception("get prompt failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# POST /api/ai/prompts — create a new version of a key (or a new key)
# ------------------------------------------------------------------
@http.route("/api/ai/prompts", type="http", auth="none", methods=["POST"], csrf=False)
@jwt_required
def create_version(self, **kw):
try:
if not request.env.user.has_group("base.group_system"):
return _error_response("Admin privileges required", 403)
body = _get_json_body() or {}
key = (body.get("key") or "").strip()
content = body.get("content") or ""
title = (body.get("title") or "").strip()
if not key or not content.strip() or not title:
return _error_response(
"key, title, and content are required", 400,
)
vals = {
"key": key,
"title": title,
"description": body.get("description") or "",
"content": content,
"is_active": bool(body.get("activate", False)),
}
with request.env.cr.savepoint():
prompt = request.env["encoach.ai.prompt"].sudo().create(vals)
return _json_response(_prompt_to_dict(prompt), 201)
except Exception as e:
_logger.exception("create prompt version failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# POST /api/ai/prompts/<id>/activate
# ------------------------------------------------------------------
@http.route(
'/api/ai/prompts/<int:prompt_id>/activate',
type="http", auth="none", methods=["POST"], csrf=False,
)
@jwt_required
def activate(self, prompt_id, **kw):
try:
if not request.env.user.has_group("base.group_system"):
return _error_response("Admin privileges required", 403)
prompt = request.env["encoach.ai.prompt"].sudo().browse(prompt_id)
if not prompt.exists():
return _error_response("Prompt not found", 404)
with request.env.cr.savepoint():
prompt.write({"is_active": True})
return _json_response(_prompt_to_dict(prompt))
except Exception as e:
_logger.exception("activate prompt failed")
return _error_response(str(e), 500)
# ------------------------------------------------------------------
# POST /api/ai/prompts/<id>/render — dry-run render with sample variables
# ------------------------------------------------------------------
@http.route(
'/api/ai/prompts/<int:prompt_id>/render',
type="http", auth="none", methods=["POST"], csrf=False,
)
@jwt_required
def render_preview(self, prompt_id, **kw):
try:
prompt = request.env["encoach.ai.prompt"].sudo().browse(prompt_id)
if not prompt.exists():
return _error_response("Prompt not found", 404)
body = _get_json_body() or {}
variables = body.get("variables") or {}
if not isinstance(variables, dict):
return _error_response("variables must be a JSON object", 400)
rendered = prompt.render(variables)
return _json_response({
"rendered": rendered,
"key": prompt.key,
"version": prompt.version,
})
except Exception as e:
_logger.exception("render prompt failed")
return _error_response(str(e), 400)

2
custom_addons/encoach_ai/models/__init__.py
View File

@@ -1,3 +1,5 @@
from . import ai_settings
from . import ai_log
from . import ai_prompt
from . import ai_feedback
from . import constants

134
custom_addons/encoach_ai/models/ai_feedback.py Normal file
View File

@@ -0,0 +1,134 @@
"""Student feedback on AI-generated content (thumbs up/down).
This closes the AI quality loop started by the human-review workflow (P3.3)
and the versioned prompts (P3.4):
* Every AI-generated artefact (a question, a coach reply, an explanation, a
translation, ...) can have any number of feedback rows attached to it.
* Feedback is always coupled to a **subject type + subject id** so it can
point at anything (``encoach.question``, ``encoach.ai.log``, ...) without
forcing a hard FK per model.
* When the student flags feedback as negative, we also store their free-text
note so product can triage.
Downstream consumers (prompt library, RAG indexers, report dashboards) can
read ``encoach.ai.feedback`` to compute win rates per prompt key, per model,
per entity, etc.
"""
from __future__ import annotations
from odoo import api, fields, models
from odoo.exceptions import ValidationError
class EncoachAIFeedback(models.Model):
_name = "encoach.ai.feedback"
_description = "Student feedback on AI-generated content"
_order = "create_date desc"
_rec_name = "subject_key"
# ------------------------------------------------------------------
# What was rated
# ------------------------------------------------------------------
subject_type = fields.Selection(
[
("question", "Exam question"),
("coach", "AI Coach reply"),
("explanation", "Content explanation"),
("translation", "Translation"),
("narrative", "Report narrative"),
("other", "Other AI output"),
],
required=True, index=True,
)
subject_id = fields.Integer(
required=True, index=True,
help="Id of the rated artefact (semantics depend on subject_type).",
)
subject_key = fields.Char(
compute="_compute_subject_key", store=True, index=True,
help="Denormalised '<type>:<id>' key for quick lookups/aggregations.",
)
# Optional link to the AI call that produced this artefact. When set, it
# lets us correlate feedback with the underlying model/prompt used.
ai_log_id = fields.Many2one(
"encoach.ai.log", ondelete="set null", index=True,
)
prompt_key = fields.Char(
index=True,
help="If the artefact was produced via encoach.ai.prompt, record the "
"key here so we can compute win-rates per prompt.",
)
prompt_version = fields.Integer()
# ------------------------------------------------------------------
# Rating
# ------------------------------------------------------------------
rating = fields.Selection(
[("up", "Thumbs up"), ("down", "Thumbs down")],
required=True, index=True,
)
comment = fields.Text(
help="Free-text note. Required when rating is 'down' and the UI "
"prompts the student for a reason.",
)
tags = fields.Char(
help="Comma-separated tags (e.g. 'wrong-answer,confusing-stem').",
)
# ------------------------------------------------------------------
# Who / context
# ------------------------------------------------------------------
user_id = fields.Many2one(
"res.users", required=True, index=True,
default=lambda self: self.env.user,
)
entity_id = fields.Many2one(
"encoach.entity", ondelete="set null", index=True,
)
course_id = fields.Many2one(
"encoach.course", ondelete="set null", index=True,
)
# ------------------------------------------------------------------
# Triage status (filled in by admins/coaches reviewing feedback).
# ------------------------------------------------------------------
status = fields.Selection(
[
("open", "Open"),
("acknowledged", "Acknowledged"),
("fixed", "Fixed"),
("dismissed", "Dismissed"),
],
default="open", index=True,
)
resolved_by_id = fields.Many2one(
"res.users", ondelete="set null",
)
resolved_at = fields.Datetime()
resolution_notes = fields.Text()
# ------------------------------------------------------------------
# Constraints
# ------------------------------------------------------------------
_sql_constraints = [
(
"uniq_user_subject",
"unique(user_id, subject_type, subject_id)",
"A user can only leave one feedback row per AI artefact. "
"Subsequent submissions should update the existing row.",
),
]
@api.depends("subject_type", "subject_id")
def _compute_subject_key(self):
for rec in self:
rec.subject_key = f"{rec.subject_type or ''}:{rec.subject_id or 0}"
@api.constrains("subject_type", "subject_id")
def _check_subject(self):
for rec in self:
if not rec.subject_type or not rec.subject_id:
raise ValidationError("subject_type and subject_id are required.")

213
custom_addons/encoach_ai/models/ai_prompt.py Normal file
View File

@@ -0,0 +1,213 @@
"""Versioned AI prompt templates.
Context (ADR-0004, P3.4 hardening release): we want non-engineers to iterate
on LLM prompts without shipping code. Every time an author edits a prompt we
keep the previous revision around so we can A/B test, audit, and roll back.
Design:
* A **prompt** is identified by its ``key`` (e.g. ``exam.mcq.generate``) — a
stable, dotted string that callers reference from Python.
* Each save produces a new **version** row (bumped ``version`` int). The old
rows stay with ``is_active = False`` so history is preserved; only one row
per ``key`` can be active at a time (enforced in ``write``/``create``).
* Callers look the prompt up with ``get_active(key)`` and render it with
``render(variables)``. Rendering uses :py:meth:`str.format_map` so templates
are just ``{variable}`` placeholders — no new DSL to learn.
This module intentionally does **not** call the LLM itself. It is a
content-management layer that the existing ``encoach_ai.services.*`` pipelines
can adopt at their own pace (falling back to their hard-coded strings until a
prompt with the matching key is created in the UI).
"""
from __future__ import annotations
import logging
import re
from typing import Iterable
from odoo import api, fields, models
from odoo.exceptions import UserError, ValidationError
_logger = logging.getLogger(__name__)
# Keys look like ``namespace.sub_namespace.name`` — lowercase, dotted.
_KEY_RE = re.compile(r"^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$")
# Variables pulled from ``{foo}`` / ``{foo.bar}`` placeholders.
_VAR_RE = re.compile(r"\{([a-zA-Z_][a-zA-Z0-9_.]*)\}")
class EncoachAIPrompt(models.Model):
_name = "encoach.ai.prompt"
_description = "Versioned AI prompt template"
_order = "key, version desc"
_rec_name = "display_name"
key = fields.Char(
required=True,
index=True,
help="Stable dotted identifier (e.g. 'exam.mcq.generate'). "
"Callers reference this from Python.",
)
version = fields.Integer(required=True, default=1, index=True)
title = fields.Char(
required=True,
help="Short human label shown in the admin editor.",
)
description = fields.Text(
help="Author intent, expected variables, known limitations.",
)
content = fields.Text(
required=True,
help="Template body. Use {variable} placeholders; "
"render() fills them via str.format_map.",
)
is_active = fields.Boolean(
default=False, index=True,
help="Exactly one active row per key. Creating/activating a new row "
"automatically deactivates the previous active version.",
)
author_id = fields.Many2one("res.users", ondelete="set null", readonly=True)
activated_at = fields.Datetime(readonly=True)
# Denormalised for quick display/search. Recomputed on save.
variables = fields.Char(
compute="_compute_variables", store=True,
help="Comma-separated list of {variable} placeholders found in content.",
)
display_name = fields.Char(
compute="_compute_display_name", store=True,
)
_sql_constraints = [
(
"key_version_uniq",
"unique(key, version)",
"A prompt version must be unique for its key.",
),
]
# ------------------------------------------------------------------
# Computed fields
# ------------------------------------------------------------------
@api.depends("content")
def _compute_variables(self):
for rec in self:
if not rec.content:
rec.variables = ""
continue
found = sorted(set(_VAR_RE.findall(rec.content)))
rec.variables = ",".join(found)
@api.depends("key", "version", "title")
def _compute_display_name(self):
for rec in self:
rec.display_name = f"{rec.key} v{rec.version} — {rec.title or ''}".rstrip(" —")
# ------------------------------------------------------------------
# Validation
# ------------------------------------------------------------------
@api.constrains("key")
def _check_key_shape(self):
for rec in self:
if not rec.key or not _KEY_RE.match(rec.key):
raise ValidationError(
f"Invalid prompt key {rec.key!r}. "
"Use lowercase dotted identifiers like 'exam.mcq.generate'."
)
@api.constrains("content")
def _check_content_non_empty(self):
for rec in self:
if not (rec.content or "").strip():
raise ValidationError("Prompt content cannot be empty.")
# ------------------------------------------------------------------
# Create / write — enforce "one active per key"
# ------------------------------------------------------------------
@api.model_create_multi
def create(self, vals_list):
# Default version = max(existing) + 1 when caller omits it.
for vals in vals_list:
if "version" not in vals or not vals.get("version"):
key = vals.get("key")
existing = self.search(
[("key", "=", key)], order="version desc", limit=1,
)
vals["version"] = (existing.version + 1) if existing else 1
vals.setdefault("author_id", self.env.user.id)
if vals.get("is_active"):
vals.setdefault("activated_at", fields.Datetime.now())
records = super().create(vals_list)
# Post-create: enforce exclusivity for any rows created as active.
for rec in records:
if rec.is_active:
rec._deactivate_siblings()
return records
def write(self, vals):
res = super().write(vals)
if vals.get("is_active"):
now = fields.Datetime.now()
for rec in self:
if rec.is_active and not rec.activated_at:
super(EncoachAIPrompt, rec).write({"activated_at": now})
rec._deactivate_siblings()
return res
def _deactivate_siblings(self):
for rec in self:
if not rec.is_active:
continue
siblings = self.search([
("key", "=", rec.key),
("id", "!=", rec.id),
("is_active", "=", True),
])
if siblings:
siblings.write({"is_active": False})
# ------------------------------------------------------------------
# Public API (called by pipelines / controllers)
# ------------------------------------------------------------------
@api.model
def get_active(self, key: str):
"""Return the active prompt record for ``key`` (or an empty set)."""
return self.sudo().search(
[("key", "=", key), ("is_active", "=", True)], limit=1,
)
def render(self, variables: dict | None = None) -> str:
"""Fill ``{placeholders}`` using ``variables`` (dict)."""
self.ensure_one()
variables = variables or {}
missing = self._missing_variables(variables.keys())
if missing:
raise UserError(
f"Missing prompt variables for {self.key} v{self.version}: "
f"{', '.join(sorted(missing))}"
)
try:
return self.content.format_map(_SafeFormatDict(variables))
except Exception as exc:
_logger.exception("prompt render failed for %s v%s", self.key, self.version)
raise UserError(f"Prompt render failed: {exc}") from exc
def _missing_variables(self, provided: Iterable[str]) -> set[str]:
self.ensure_one()
required = set(_VAR_RE.findall(self.content or ""))
provided_root = {p.split(".")[0] for p in provided}
return {v for v in required if v.split(".")[0] not in provided_root}
class _SafeFormatDict(dict):
"""Dict that returns ``''`` for dotted/nested lookups str.format can't resolve.
We don't advertise ``{foo.bar}`` as supported (the editor shows top-level
variables only), but we also don't want rendering to explode if a caller
passes an object with attributes.
"""
def __missing__(self, key):
return "{" + key + "}"

4
custom_addons/encoach_ai/security/ir.model.access.csv
View File

@@ -1,3 +1,7 @@
id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
access_ai_log_admin,encoach.ai.log admin,model_encoach_ai_log,base.group_system,1,1,1,1
access_ai_log_user,encoach.ai.log user,model_encoach_ai_log,base.group_user,1,0,1,0
access_ai_prompt_admin,encoach.ai.prompt admin,model_encoach_ai_prompt,base.group_system,1,1,1,1
access_ai_prompt_user,encoach.ai.prompt user,model_encoach_ai_prompt,base.group_user,1,0,0,0
access_ai_feedback_admin,encoach.ai.feedback admin,model_encoach_ai_feedback,base.group_system,1,1,1,1
access_ai_feedback_user,encoach.ai.feedback user,model_encoach_ai_feedback,base.group_user,1,1,1,0
1 id name model_id:id group_id:id perm_read perm_write perm_create perm_unlink
2 access_ai_log_admin encoach.ai.log admin model_encoach_ai_log base.group_system 1 1 1 1
3 access_ai_log_user encoach.ai.log user model_encoach_ai_log base.group_user 1 0 1 0
4 access_ai_prompt_admin encoach.ai.prompt admin model_encoach_ai_prompt base.group_system 1 1 1 1
5 access_ai_prompt_user encoach.ai.prompt user model_encoach_ai_prompt base.group_user 1 0 0 0
6 access_ai_feedback_admin encoach.ai.feedback admin model_encoach_ai_feedback base.group_system 1 1 1 1
7 access_ai_feedback_user encoach.ai.feedback user model_encoach_ai_feedback base.group_user 1 1 1 0

2
custom_addons/encoach_ai/services/__init__.py
View File

@@ -5,3 +5,5 @@ from .elevenlabs_service import ElevenLabsService
from .gptzero_service import GPTZeroService
from .elai_service import ElaiService
from .coach_service import CoachService
from . import cefr_mapper # canonical CEFR / band / theta mapper (P0.9)
from . import question_validator # schema + quality gate for AI-generated questions (P1.6/P1.1)

149
custom_addons/encoach_ai/services/cefr_mapper.py Normal file
View File

@@ -0,0 +1,149 @@
"""Canonical CEFR / IELTS band / IRT theta mapper.
This is the SINGLE SOURCE OF TRUTH for band ↔ CEFR ↔ theta conversions across
the platform. All modules MUST import from here rather than hand-rolling their
own mapping tables. See §21 of docs/PROJECT_SUMMARY.md for the rationale.
"""
# IRT theta band thresholds (ability parameter → CEFR level)
THETA_TO_CEFR = [
(-4.0, -2.5, 'pre_a1'),
(-2.5, -1.5, 'a1'),
(-1.5, -0.5, 'a2'),
(-0.5, 0.5, 'b1'),
(0.5, 1.5, 'b2'),
(1.5, 2.5, 'c1'),
(2.5, 4.0, 'c2'),
]
# CEFR code → canonical IELTS band centre value
CEFR_TO_BAND = {
'pre_a1': 2.0,
'a1': 3.0,
'a2': 4.0,
'b1': 5.0,
'b2': 6.5,
'c1': 7.5,
'c2': 9.0,
}
# Lowercase CEFR code → human-readable label
CEFR_LABELS = {
'pre_a1': 'Pre-A1 (Beginner)',
'a1': 'A1 (Elementary)',
'a2': 'A2 (Pre-Intermediate)',
'b1': 'B1 (Intermediate)',
'b2': 'B2 (Upper-Intermediate)',
'c1': 'C1 (Advanced)',
'c2': 'C2 (Proficient)',
}
# Monotonic (highest band → lowest) thresholds used by band_to_cefr.
# Picked to align CEFR_TO_BAND with widely-published IELTS/CEFR crosswalks
# (Cambridge, Council of Europe): 8.0+ → C2, 7.0-7.9 → C1, 5.5-6.9 → B2, etc.
_BAND_THRESHOLDS = [
(8.0, 'c2'),
(7.0, 'c1'),
(5.5, 'b2'),
(4.5, 'b1'),
(3.5, 'a2'),
(2.5, 'a1'),
(0.0, 'pre_a1'),
]
def band_to_cefr(band):
"""Map an IELTS band (0-9, may be float) to a canonical CEFR code."""
if band is None:
return None
try:
b = float(band)
except (TypeError, ValueError):
return None
for threshold, level in _BAND_THRESHOLDS:
if b >= threshold:
return level
return 'pre_a1'
def theta_to_cefr(theta):
"""Map an IRT theta ability parameter (typically -4..+4) to a CEFR code."""
try:
t = float(theta)
except (TypeError, ValueError):
return 'pre_a1'
for low, high, level in THETA_TO_CEFR:
if low <= t < high:
return level
return 'c2' if t >= 2.5 else 'pre_a1'
def theta_to_band(theta):
"""Interpolate an IRT theta to an IELTS band, rounded to nearest 0.5."""
try:
t = float(theta)
except (TypeError, ValueError):
return 0.0
cefr = theta_to_cefr(t)
base_band = CEFR_TO_BAND.get(cefr, 5.0)
for low, high, level in THETA_TO_CEFR:
if level == cefr:
width = high - low
position = (t - low) / width if width > 0 else 0.5
cefr_list = list(CEFR_TO_BAND.keys())
idx = cefr_list.index(cefr)
next_band = CEFR_TO_BAND.get(
cefr_list[min(idx + 1, len(cefr_list) - 1)], base_band + 1.0
)
band = base_band + position * (next_band - base_band)
return round(band * 2) / 2
return base_band
def cefr_to_band(cefr):
"""Return the canonical centre-of-range band for a CEFR code."""
if not cefr:
return None
return CEFR_TO_BAND.get(str(cefr).strip().lower().replace('-', '_'))
def normalize_cefr(label):
"""Canonicalise a CEFR label to the uppercase display form (e.g. 'A2', 'Pre-A1').
Accepts 'a2', 'A2', 'pre_a1', 'Pre-A1', 'pre a1' etc. Returns ``None``
if the input is falsy; returns the input unchanged if unrecognised.
"""
if not label:
return None
s = str(label).strip().lower().replace('-', '_').replace(' ', '_')
return {
'pre_a1': 'Pre-A1',
'a1': 'A1',
'a2': 'A2',
'b1': 'B1',
'b2': 'B2',
'c1': 'C1',
'c2': 'C2',
}.get(s, label)
def cefr_label(cefr_code):
"""Return the human-readable label (e.g. 'B2 (Upper-Intermediate)') for a code."""
return CEFR_LABELS.get(cefr_code, cefr_code)
class CefrMapper:
"""Class-style facade kept for backward compatibility with ``encoach_placement``."""
THETA_TO_CEFR = THETA_TO_CEFR
CEFR_TO_BAND = CEFR_TO_BAND
CEFR_LABELS = CEFR_LABELS
theta_to_cefr = staticmethod(theta_to_cefr)
theta_to_band = staticmethod(theta_to_band)
band_to_cefr = staticmethod(band_to_cefr)
@staticmethod
def get_cefr_label(cefr_code):
return cefr_label(cefr_code)

10
custom_addons/encoach_ai/services/openai_service.py
View File

@@ -20,16 +20,20 @@ class OpenAIService:
self._get_param = env["ir.config_parameter"].sudo().get_param
self.enabled = self._get_param("encoach_ai.enabled", "True").lower() in ("1", "true", "yes")
self.max_retries = int(self._get_param("encoach_ai.max_retries", "3"))
# Hard wall-clock timeout (seconds) enforced by the OpenAI SDK on each
# HTTP call. Prevents a single slow LLM request from tying up an Odoo
# worker for minutes. See P0.5 in docs/PROJECT_SUMMARY.md §21.
self.request_timeout = float(self._get_param("encoach_ai.request_timeout", "30"))
api_key = self._get_param("encoach_ai.openai_api_key", "")
if not api_key:
import os
api_key = os.environ.get("OPENAI_API_KEY", "")
if _openai_mod and api_key:
self.client = _openai_mod.OpenAI(api_key=api_key)
self.client = _openai_mod.OpenAI(api_key=api_key, timeout=self.request_timeout)
else:
self.client = None
self.model = self._get_param("encoach_ai.openai_model", "gpt-4o")
self.fast_model = self._get_param("encoach_ai.openai_fast_model", "gpt-3.5-turbo")
self.fast_model = self._get_param("encoach_ai.openai_fast_model", "gpt-4o-mini")
def _log(self, action, model, usage, latency, status="success", error=None, inp=None, out=None):
try:
@@ -86,6 +90,7 @@ class OpenAIService:
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
timeout=self.request_timeout,
)
resp = self._retry_with_backoff(_call, action, model)
content = resp.choices[0].message.content
@@ -112,6 +117,7 @@ class OpenAIService:
temperature=temperature,
max_tokens=max_tokens,
response_format={"type": "json_object"},
timeout=self.request_timeout,
)
resp = self._retry_with_backoff(_call, action, model)
raw = resp.choices[0].message.content

318
custom_addons/encoach_ai/services/question_validator.py Normal file
View File

@@ -0,0 +1,318 @@
"""Validation + quality gate for AI-generated exam content.
This module is the single entry point used by exam-generation controllers to:
1. **Schema-validate** raw LLM output before any DB insert (``validate_payload``),
catching malformed AI responses (missing prompt, empty options for MCQ,
wrong correct_answer format, etc.) before they corrupt the question pool.
2. **Score content quality** of persisted questions (``score_question``), wrapping
:class:`encoach_quality_gate.services.quality_checker.QualityChecker` and
:class:`encoach_ielts_validation.services.validator.IeltsValidator` with
graceful degradation when those addons are not installed or their optional
Python deps (``textstat``, ``language_tool_python``) are missing.
Design choices:
- Zero hard dependency on ``jsonschema`` or ``pydantic`` — the schema is a
lightweight, introspectable dict used by :func:`_validate_dict`. Good enough
for the shapes we actually emit; keeps our external-deps footprint small.
- Quality scoring is best-effort and **never** raises to callers — failures
degrade to a ``warn`` verdict so the HTTP request still completes.
"""
from __future__ import annotations
import hashlib
import json
import logging
from typing import Any
_logger = logging.getLogger(__name__)
VERDICT_OK = "ok"
VERDICT_WARN = "warn"
VERDICT_FAIL = "fail"
MIN_STEM_LEN = 8
MIN_QUALITY_SCORE = 0.55
_VALID_QUESTION_TYPES = {
"mcq", "mcq_multi", "tfng", "ynng", "gap_fill", "short_answer",
"form_completion", "note_completion", "map_labelling", "matching",
"summary_completion", "heading_matching", "matching_features",
"numerical", "expression", "code_completion", "code_output",
"sql_query", "matrix", "graph",
}
_SKILL_VALUES = {
"listening", "reading", "writing", "speaking", "grammar",
"vocabulary", "math", "it",
}
EXERCISE_SCHEMA: dict[str, dict[str, Any]] = {
"prompt": {"type": str, "required": True, "min_len": MIN_STEM_LEN},
"type": {"type": str, "required": False, "choices": _VALID_QUESTION_TYPES | {
"multiple_choice", "true_false", "fill_blanks", "matching_headings",
"paragraph_match", "sentence_completion", "matching_information",
}},
"options": {"type": list, "required": False},
"correct_answer": {"type": (str, list, int, float), "required": False},
"marks": {"type": (int, float), "required": False, "min": 0, "max": 20},
"cefr_level": {"type": str, "required": False},
}
MODULE_SCHEMA: dict[str, dict[str, Any]] = {
"difficulty": {"type": (list, str), "required": False},
"timer": {"type": (int, float), "required": False, "min": 0, "max": 600},
"totalMarks": {"type": (int, float), "required": False, "min": 0, "max": 500},
"passages": {"type": list, "required": False},
"sections": {"type": list, "required": False},
"tasks": {"type": list, "required": False},
"parts": {"type": list, "required": False},
}
def _validate_dict(obj: Any, schema: dict, path: str = "") -> list[str]:
"""Return list of human-readable validation errors for ``obj`` against ``schema``."""
errors: list[str] = []
if not isinstance(obj, dict):
errors.append(f"{path or 'root'}: expected object, got {type(obj).__name__}")
return errors
for key, rule in schema.items():
here = f"{path}.{key}" if path else key
if key not in obj or obj[key] in (None, ""):
if rule.get("required"):
errors.append(f"{here}: missing required field")
continue
val = obj[key]
expected_type = rule.get("type")
if expected_type and not isinstance(val, expected_type):
got = type(val).__name__
want = (
expected_type.__name__ if isinstance(expected_type, type)
else "/".join(t.__name__ for t in expected_type)
)
errors.append(f"{here}: expected {want}, got {got}")
continue
if isinstance(val, str):
min_len = rule.get("min_len")
if min_len and len(val.strip()) < min_len:
errors.append(f"{here}: too short (< {min_len} chars)")
choices = rule.get("choices")
if choices and val not in choices:
errors.append(f"{here}: '{val}' not in allowed values")
if isinstance(val, (int, float)):
if "min" in rule and val < rule["min"]:
errors.append(f"{here}: {val} < min {rule['min']}")
if "max" in rule and val > rule["max"]:
errors.append(f"{here}: {val} > max {rule['max']}")
return errors
def _validate_exercise(ex: Any, path: str = "exercise") -> list[str]:
"""Validate a single AI-generated exercise dict."""
errors = _validate_dict(ex, EXERCISE_SCHEMA, path)
if errors:
return errors
q_type = (ex.get("type") or "mcq").lower()
options = ex.get("options") or []
correct = ex.get("correct_answer")
if q_type in ("mcq", "multiple_choice"):
if not options or len(options) < 2:
errors.append(f"{path}.options: MCQ requires >= 2 options")
if correct in (None, ""):
errors.append(f"{path}.correct_answer: MCQ requires a correct_answer")
if q_type in ("mcq_multi",):
if not isinstance(correct, list) or len(correct) < 1:
errors.append(f"{path}.correct_answer: mcq_multi requires a list of answers")
if q_type in ("true_false", "tfng", "ynng"):
if isinstance(correct, str) and correct.lower() not in {
"true", "false", "not_given", "yes", "no", "t", "f"
}:
errors.append(f"{path}.correct_answer: '{correct}' not a valid T/F/NG answer")
return errors
def validate_payload(payload: Any) -> dict[str, Any]:
"""Validate an entire exam-generation submit payload.
Expected shape (best-effort, since the frontend already normalizes it):
.. code-block:: text
{
"title": str (required),
"modules": {
"<skill>": {
"passages": [{"text": str, "exercises": [ExerciseSchema, ...]}],
"sections": [...], "tasks": [...], "parts": [...],
...
},
...
}
}
Returns a dict with::
{"verdict": "ok"|"warn"|"fail", "errors": [...], "warnings": [...]}
``verdict == "fail"`` means the caller MUST NOT persist this payload as-is.
"""
errors: list[str] = []
warnings: list[str] = []
if not isinstance(payload, dict):
return {"verdict": VERDICT_FAIL,
"errors": [f"payload: expected object, got {type(payload).__name__}"],
"warnings": []}
title = (payload.get("title") or "").strip()
if not title:
errors.append("title: missing required field")
modules = payload.get("modules") or {}
if not isinstance(modules, dict) or not modules:
errors.append("modules: missing or empty — nothing to persist")
return {"verdict": VERDICT_FAIL, "errors": errors, "warnings": warnings}
total_exercises = 0
for mod_key, mod in modules.items():
mpath = f"modules.{mod_key}"
if not isinstance(mod, dict):
errors.append(f"{mpath}: expected object")
continue
errors.extend(_validate_dict(mod, MODULE_SCHEMA, mpath))
for p_idx, passage in enumerate(mod.get("passages") or []):
if not isinstance(passage, dict):
errors.append(f"{mpath}.passages[{p_idx}]: expected object")
continue
for e_idx, ex in enumerate(passage.get("exercises") or []):
total_exercises += 1
errors.extend(_validate_exercise(
ex, f"{mpath}.passages[{p_idx}].exercises[{e_idx}]"))
for s_idx, sec in enumerate(mod.get("sections") or []):
if not isinstance(sec, dict):
errors.append(f"{mpath}.sections[{s_idx}]: expected object")
continue
for e_idx, ex in enumerate(sec.get("exercises") or []):
total_exercises += 1
errors.extend(_validate_exercise(
ex, f"{mpath}.sections[{s_idx}].exercises[{e_idx}]"))
if total_exercises == 0 and not any(
(m.get("tasks") or m.get("parts")) for m in modules.values() if isinstance(m, dict)
):
warnings.append("no exercises/tasks/parts found — exam will have no questions")
if errors:
verdict = VERDICT_FAIL
elif warnings:
verdict = VERDICT_WARN
else:
verdict = VERDICT_OK
return {"verdict": verdict, "errors": errors, "warnings": warnings,
"total_exercises": total_exercises}
def prompt_hash(prompt: str) -> str:
"""SHA-256 hex digest of a rendered prompt, for provenance tracking."""
return hashlib.sha256((prompt or "").encode("utf-8")).hexdigest()
def score_question(stem: str, *, skill: str | None = None,
cefr_level: str | None = None) -> dict[str, Any]:
"""Best-effort quality score for a persisted question stem.
Wraps QualityChecker + IeltsValidator defensively — any import/runtime
failure downgrades the verdict to ``warn`` without raising. Returns::
{
"verdict": "ok"|"warn"|"fail",
"score": float in [0, 1],
"readability": {...} | None,
"grammar_issues": int,
"checks_ran": ["readability", "grammar", ...],
"errors": [str, ...],
}
"""
report: dict[str, Any] = {
"verdict": VERDICT_OK,
"score": 1.0,
"readability": None,
"grammar_issues": 0,
"checks_ran": [],
"errors": [],
}
if not stem or len(stem.strip()) < MIN_STEM_LEN:
report["verdict"] = VERDICT_FAIL
report["score"] = 0.0
report["errors"].append(f"stem shorter than {MIN_STEM_LEN} chars")
return report
try:
from odoo.addons.encoach_quality_gate.services.quality_checker import QualityChecker
except Exception as exc:
report["verdict"] = VERDICT_WARN
report["errors"].append(f"quality_gate unavailable: {exc}")
return report
try:
readability = QualityChecker.check_readability(stem)
report["readability"] = readability
report["checks_ran"].append("readability")
if cefr_level:
cefr_match = QualityChecker.check_cefr_alignment(stem, cefr_level.lower())
report["cefr_alignment"] = cefr_match
report["checks_ran"].append("cefr_alignment")
if not cefr_match.get("aligned", True):
report["score"] -= 0.2
grammar = QualityChecker.check_grammar(stem)
report["grammar_issues"] = len(grammar.get("issues", [])) if isinstance(grammar, dict) else 0
report["checks_ran"].append("grammar")
if report["grammar_issues"] > 3:
report["score"] -= 0.2
except Exception as exc:
_logger.warning("QualityChecker failed for stem (skill=%s): %s", skill, exc)
report["verdict"] = VERDICT_WARN
report["errors"].append(f"quality_checker error: {exc}")
report["score"] = max(0.0, min(1.0, report["score"]))
if report["verdict"] == VERDICT_OK and report["score"] < MIN_QUALITY_SCORE:
report["verdict"] = VERDICT_WARN
return report
def summarize_question_reports(reports: list[dict]) -> dict[str, Any]:
"""Reduce a list of per-question reports to an aggregate verdict."""
if not reports:
return {"verdict": VERDICT_OK, "total": 0, "failed": 0,
"warned": 0, "avg_score": 1.0}
failed = sum(1 for r in reports if r.get("verdict") == VERDICT_FAIL)
warned = sum(1 for r in reports if r.get("verdict") == VERDICT_WARN)
avg = sum(float(r.get("score") or 0) for r in reports) / max(1, len(reports))
if failed:
verdict = VERDICT_FAIL
elif warned:
verdict = VERDICT_WARN
else:
verdict = VERDICT_OK
return {
"verdict": verdict,
"total": len(reports),
"failed": failed,
"warned": warned,
"avg_score": round(avg, 3),
}
def dumps_report(report: dict) -> str:
"""Compact JSON dump safe for storing in a Text column."""
try:
return json.dumps(report, ensure_ascii=False, default=str)
except Exception:
return "{}"