bf3981028b
- ticket: add user_id, priority, category fields; fix _serialize_ticket -> _serialize - package: fix is_active -> active (Odoo auto-filter), remove invalid entity_id filter - walkthrough: add sequence field; fix controller to use name/content instead of title/steps - users: fix encoach_user_type -> encoach_type, encoach_entity_id -> entity_rel_ids.entity_id - generation: use EncoachGenerationService class instead of nonexistent ORM model, fix /api/exam/level/ route conflict (separate GET and POST URLs) Made-with: Cursor
352 lines
14 KiB
Python
352 lines
14 KiB
Python
import json
|
|
import logging
|
|
|
|
from odoo import http
|
|
from odoo.http import request
|
|
|
|
from .base import EncoachMixin
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _get_gen_service():
|
|
from odoo.addons.encoach_ai_generation.services.generation_service import (
|
|
EncoachGenerationService,
|
|
)
|
|
return EncoachGenerationService(request.env)
|
|
|
|
|
|
class GenerationController(EncoachMixin, http.Controller):
|
|
|
|
# ------------------------------------------------- reading passage
|
|
@http.route(
|
|
'/api/exam/reading/<int:passage>', type='http', auth='public',
|
|
methods=['GET', 'OPTIONS'], csrf=False, cors='*',
|
|
)
|
|
def generate_reading_passage(self, passage, **kwargs):
|
|
"""Generate a reading passage (1, 2, or 3) using GPT-4o."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
if passage not in (1, 2, 3):
|
|
return self._error_response('passage must be 1, 2, or 3', 400)
|
|
|
|
topic = kwargs.get('topic')
|
|
word_count = int(kwargs.get('word_count', 500))
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_reading_passage(
|
|
passage_num=passage,
|
|
topic=topic,
|
|
word_count=word_count,
|
|
)
|
|
|
|
return self._json_response(result)
|
|
except Exception:
|
|
_logger.exception("Generate reading passage error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# ------------------------------------------ reading exercises
|
|
@http.route('/api/exam/reading/', type='http', auth='public', methods=['POST', 'OPTIONS'], csrf=False, cors='*')
|
|
def generate_reading_exercises(self, **kwargs):
|
|
"""Generate reading exercises from a passage."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
|
|
body = self._get_json_body()
|
|
text = body.get('text', '')
|
|
exercises = body.get('exercises', [])
|
|
difficulty = body.get('difficulty', ['B1', 'B2'])
|
|
|
|
if not text or not exercises:
|
|
return self._error_response('text and exercises are required', 400)
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_reading_exercises(
|
|
text=text,
|
|
exercises_config=exercises,
|
|
difficulty=difficulty,
|
|
)
|
|
|
|
return self._json_response({'exercises': result})
|
|
except Exception:
|
|
_logger.exception("Generate reading exercises error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# ----------------------------------------------- listening dialog
|
|
@http.route(
|
|
'/api/exam/listening/<int:section>', type='http', auth='public',
|
|
methods=['GET', 'OPTIONS'], csrf=False, cors='*',
|
|
)
|
|
def generate_listening_dialog(self, section, **kwargs):
|
|
"""Generate a listening dialog/monologue (section 1-4) using GPT-4o."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
if section not in (1, 2, 3, 4):
|
|
return self._error_response('section must be 1, 2, 3, or 4', 400)
|
|
|
|
topic = kwargs.get('topic')
|
|
difficulty = kwargs.get('difficulty')
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_listening_dialog(
|
|
section=section,
|
|
topic=topic,
|
|
difficulty=difficulty,
|
|
)
|
|
|
|
return self._json_response(result)
|
|
except Exception:
|
|
_logger.exception("Generate listening dialog error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# ---------------------------------------- listening exercises
|
|
@http.route('/api/exam/listening/', type='http', auth='public', methods=['POST', 'OPTIONS'], csrf=False, cors='*')
|
|
def generate_listening_exercises(self, **kwargs):
|
|
"""Generate listening exercises from a dialog transcript."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
|
|
body = self._get_json_body()
|
|
text = body.get('text', '')
|
|
exercises = body.get('exercises', [])
|
|
difficulty = body.get('difficulty', ['B1'])
|
|
|
|
if not text or not exercises:
|
|
return self._error_response('text and exercises are required', 400)
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_listening_exercises(
|
|
text=text,
|
|
exercises_config=exercises,
|
|
difficulty=difficulty,
|
|
)
|
|
|
|
return self._json_response({'exercises': result})
|
|
except Exception:
|
|
_logger.exception("Generate listening exercises error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# ------------------------------------------------- writing task
|
|
@http.route(
|
|
'/api/exam/writing/<int:task>', type='http', auth='public',
|
|
methods=['GET', 'OPTIONS'], csrf=False, cors='*',
|
|
)
|
|
def generate_writing_task(self, task, **kwargs):
|
|
"""Generate a writing task prompt (task 1 or 2) using GPT-4o."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
if task not in (1, 2):
|
|
return self._error_response('task must be 1 or 2', 400)
|
|
|
|
topic = kwargs.get('topic')
|
|
difficulty = kwargs.get('difficulty')
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_writing_task(
|
|
task=task,
|
|
topic=topic,
|
|
difficulty=difficulty,
|
|
)
|
|
|
|
return self._json_response(result)
|
|
except Exception:
|
|
_logger.exception("Generate writing task error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# ------------------------------------------------ speaking task
|
|
@http.route(
|
|
'/api/exam/speaking/<int:task>', type='http', auth='public',
|
|
methods=['GET', 'OPTIONS'], csrf=False, cors='*',
|
|
)
|
|
def generate_speaking_task(self, task, **kwargs):
|
|
"""Generate a speaking task prompt (part 1, 2, or 3) using GPT-4o."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
if task not in (1, 2, 3):
|
|
return self._error_response('task must be 1, 2, or 3', 400)
|
|
|
|
topic = kwargs.get('topic')
|
|
first_topic = kwargs.get('first_topic')
|
|
second_topic = kwargs.get('second_topic')
|
|
difficulty = kwargs.get('difficulty')
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_speaking_task(
|
|
part=task,
|
|
topic=topic,
|
|
first_topic=first_topic,
|
|
second_topic=second_topic,
|
|
difficulty=difficulty,
|
|
)
|
|
|
|
return self._json_response(result)
|
|
except Exception:
|
|
_logger.exception("Generate speaking task error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# --------------------------------------------- level exercises (POST)
|
|
@http.route('/api/exam/level/generate', type='http', auth='public', methods=['POST', 'OPTIONS'], csrf=False, cors='*')
|
|
def generate_level_exercises(self, **kwargs):
|
|
"""Generate level-test exercises using GPT-4o."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
|
|
body = self._get_json_body()
|
|
exercises = body.get('exercises', [])
|
|
difficulty = body.get('difficulty', ['B1'])
|
|
|
|
if not exercises:
|
|
return self._error_response('exercises specification is required', 400)
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_level_exercises(
|
|
exercises_config=exercises,
|
|
difficulty=difficulty,
|
|
)
|
|
|
|
return self._json_response({'exercises': result})
|
|
except Exception:
|
|
_logger.exception("Generate level exercises error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# --------------------------------------------- writing attachment (vision)
|
|
@http.route(
|
|
'/api/exam/writing/<int:task>/attachment', type='http', auth='public',
|
|
methods=['POST', 'OPTIONS'], csrf=False, cors='*',
|
|
)
|
|
def generate_writing_attachment(self, task, **kwargs):
|
|
"""Academic writing Task 1 with image upload for GPT-4o vision."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
if task not in (1, 2):
|
|
return self._error_response('task must be 1 or 2', 400)
|
|
|
|
files = request.httprequest.files
|
|
form = request.httprequest.form
|
|
image_file = files.get('file')
|
|
difficulty = form.get('difficulty', 'B2')
|
|
|
|
if not image_file:
|
|
return self._error_response('file (image) is required', 400)
|
|
|
|
import base64
|
|
image_data = base64.b64encode(image_file.read()).decode('utf-8')
|
|
mime_type = image_file.content_type or 'image/png'
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_writing_task(
|
|
task=task,
|
|
difficulty=difficulty,
|
|
)
|
|
|
|
return self._json_response(result)
|
|
except Exception:
|
|
_logger.exception("Generate writing attachment error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# -------------------------------------------- pre-built level exam (GET)
|
|
@http.route('/api/exam/level', type='http', auth='public', methods=['GET', 'OPTIONS'], csrf=False, cors='*')
|
|
def get_level_exam(self, **kwargs):
|
|
"""Return a pre-built level exam (no AI generation)."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
|
|
exams = request.env['encoach.exam'].sudo().search([
|
|
('module', '=', 'level'),
|
|
('access', '=', 'public'),
|
|
], limit=1, order='create_date desc')
|
|
if not exams:
|
|
return self._error_response('No level exam found', 404)
|
|
return self._json_response(exams[0].to_encoach_dict())
|
|
except Exception:
|
|
_logger.exception("Get level exam error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# -------------------------------------------- UTAS level exam
|
|
@http.route('/api/exam/level/utas', type='http', auth='public', methods=['GET', 'OPTIONS'], csrf=False, cors='*')
|
|
def get_utas_level_exam(self, **kwargs):
|
|
"""Return a UTAS-format level exam."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
|
|
exams = request.env['encoach.exam'].sudo().search([
|
|
('module', '=', 'level'),
|
|
('label', 'ilike', 'utas'),
|
|
], limit=1, order='create_date desc')
|
|
if not exams:
|
|
return self._error_response('No UTAS level exam found', 404)
|
|
return self._json_response(exams[0].to_encoach_dict())
|
|
except Exception:
|
|
_logger.exception("Get UTAS level exam error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# -------------------------------------------- level exam import
|
|
@http.route('/api/exam/level/import', type='http', auth='public', methods=['POST', 'OPTIONS'], csrf=False, cors='*')
|
|
def import_level_exam(self, **kwargs):
|
|
"""Import level exam from uploaded file."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
|
|
files = request.httprequest.files
|
|
exercises_file = files.get('exercises')
|
|
if not exercises_file:
|
|
return self._error_response('exercises file is required', 400)
|
|
|
|
content = exercises_file.read().decode('utf-8')
|
|
data = json.loads(content)
|
|
|
|
exam = request.env['encoach.exam'].sudo().create({
|
|
'module': 'level',
|
|
'label': data.get('label', 'Imported Level Exam'),
|
|
'parts': data.get('parts', []),
|
|
'access': 'private',
|
|
})
|
|
return self._json_response(exam.to_encoach_dict())
|
|
except Exception:
|
|
_logger.exception("Import level exam error")
|
|
return self._error_response('Internal server error', 500)
|
|
|
|
# -------------------------------------------- custom level exam
|
|
@http.route('/api/exam/level/custom', type='http', auth='public', methods=['POST', 'OPTIONS'], csrf=False, cors='*')
|
|
def generate_custom_level(self, **kwargs):
|
|
"""Generate custom level exam from JSON specification."""
|
|
try:
|
|
user = self._authenticate()
|
|
if not user:
|
|
return self._error_response('Unauthorized', 401)
|
|
|
|
body = self._get_json_body()
|
|
spec = body.get('spec', body)
|
|
|
|
service = _get_gen_service()
|
|
result = service.generate_level_exercises(
|
|
exercises_config=spec.get('exercises', []),
|
|
difficulty=spec.get('difficulty', ['B1', 'B2']),
|
|
)
|
|
return self._json_response({'exercises': result})
|
|
except Exception:
|
|
_logger.exception("Generate custom level error")
|
|
return self._error_response('Internal server error', 500)
|